usage.py

"""
usage.py

Usage:
    python usage.py <input_dpp.json> <schema_name>

Arguments:
    input_dpp.json: Path to a JSON file containing a serialized DigitalProductPassport.
    schema_name:    Target schema to map to ('ECLASS' or 'ISA-95').

Example:
    python usage.py my_dpp.json ECLASS
"""

import argparse
import json
import ast
import sys
from typing import Dict, Any, Type
from pathlib import Path

from nmis_dpp import get_global_registry, register_default_mappers
from nmis_dpp.model import (
    IdentityLayer, StructureLayer, LifecycleLayer, RiskLayer,
    SustainabilityLayer, ProvenanceLayer, DigitalProductPassport
)
from nmis_dpp.part_class import (
    PartClass, OntologyBinding,
    Actuator, Sensor, PowerConversion, Thermal, Fluidics,
    Structural, ControlUnit, UserInterface, Fastener, Connectivity,
    EnergyStorage, SoftwareModule, Consumable, Protection, Transmission
)
from nmis_dpp.utils import to_json

# Registry is initialized on module import via nmis_dpp.__init__ but explicit call is safe
register_default_mappers()

# Map type strings to classes for deserialization
PART_CLASS_MAP: Dict[str, Type[PartClass]] = {
    "Actuator": Actuator,
    "Sensor": Sensor,
    "PowerConversion": PowerConversion,
    "Thermal": Thermal,
    "Fluidics": Fluidics,
    "Structural": Structural,
    "ControlUnit": ControlUnit,
    "UserInterface": UserInterface,
    "Fastener": Fastener,
    "Connectivity": Connectivity,
    "EnergyStorage": EnergyStorage,
    "SoftwareModule": SoftwareModule,
    "Consumable": Consumable,
    "Protection": Protection,
    "Transmission": Transmission,
    "PartClass": PartClass
}

def reconstruct_part(data: Dict[str, Any]) -> PartClass:
    """
    Reconstruct a specific PartClass subclass from a dictionary.
    Handles nested OntologyBinding objects.
    """
    # 1. Extract core PartClass fields that might need special handling
    p_type = data.get("type", "PartClass")
    bindings_data = data.pop("ontology_bindings", {})
    
    # 2. Reconstruct OntologyBinding objects
    bindings = {}
    for key, b_data in bindings_data.items():
        if isinstance(b_data, dict):
            bindings[key] = OntologyBinding(**b_data)
        else:
            # Fallback if somehow already object (unlikely from JSON)
            bindings[key] = b_data
            
    # 3. Resolve class
    cls = PART_CLASS_MAP.get(p_type, PartClass)
    
    # 4. Instantiate (passing remaining data as kwargs)
    # Check if 'properties' is in data, preserve it
    # Note: dataclasses constructor expects fields. Extra fields in JSON might cause error if not careful.
    # However, PartClass has 'properties' field for extras.
    # But usually JSON matches fields. if strict dataclass, we must filter.
    # For now assuming JSON strictly matches schema generated by to_dict.
    
    try:
        part = cls(**data)
    except TypeError as e:
        # Fallback: if data contains extra keys, we might need to filter them
        # or put them into 'properties'
        valid_fields = cls.__dataclass_fields__.keys()
        filtered_data = {k: v for k, v in data.items() if k in valid_fields}
        part = cls(**filtered_data)
        
        # If there were extra fields, maybe log them or add to properties?
        # part.properties.update({k: v for k, v in data.items() if k not in valid_fields})

    part.ontology_bindings = bindings
    return part

def reconstruct_dpp(data: Dict[str, Any]) -> DigitalProductPassport:
    """
    Reconstruct a DigitalProductPassport object from a dictionary.
    """
    # Reconstruct layers
    identity = IdentityLayer(**data.get("identity", {}))
    
    # Structure needs part reconstruction
    struct_data = data.get("structure", {})
    parts_data = struct_data.get("parts", [])
    parts = [reconstruct_part(p) for p in parts_data]
    
    # Update structure dict to have object list
    struct_args = struct_data.copy()
    struct_args["parts"] = parts
    
    # Handle optional fields or list of objects in other layers?
    # Usually they are dicts or lists of dicts which match standard types (e.g. interfaces, materials are list[dict])
    # So simple unpacking works for other layers.
    
    structure = StructureLayer(**struct_args)
    lifecycle = LifecycleLayer(**data.get("lifecycle", {}))
    risk = RiskLayer(**data.get("risk", {}))
    sustainability = SustainabilityLayer(**data.get("sustainability", {}))
    provenance = ProvenanceLayer(**data.get("provenance", {}))
    
    return DigitalProductPassport(
        identity=identity,
        structure=structure,
        lifecycle=lifecycle,
        risk=risk,
        sustainability=sustainability,
        provenance=provenance
    )

def main():
    parser = argparse.ArgumentParser(description="Map a DPP JSON file to a specific schema.")
    parser.add_argument("input_file", help="Path to the input DPP JSON file.")
    parser.add_argument("schema", help="Target schema (ECLASS or ISA-95).")
    
    args = parser.parse_args()
    
    input_path = Path(args.input_file)
    schema_target = args.schema
    
    if not input_path.exists():
        print(f"Error: File {input_path} not found.")
        sys.exit(1)
        
    print(f"Reading {input_path}...")
    try:
        with input_path.open("r", encoding="utf-8") as f:
            data = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error parsing JSON: {e}")
        sys.exit(1)
        
    print("Reconstructing DPP object...")
    try:
        dpp = reconstruct_dpp(data)
    except Exception as e:
        print(f"Error reconstructing DPP object: {e}")
        # import traceback
        # traceback.print_exc()
        sys.exit(1)
        
    print(f"DPP Reconstructed. Parts: {len(dpp.structure.parts)}")
    
    registry = get_global_registry()
    print(f"Mapping to {schema_target}...")
    
    try:
        mapper = registry.get_mapper(schema_target)
        output = mapper.map_dpp(dpp)
        print("Mapping Successful.")
        print(json.dumps(output, indent=2))
    except KeyError:
        print(f"Error: Schema '{schema_target}' not found. Available: {registry.list_schemas()}")
        sys.exit(1)
    except Exception as e:
        print(f"Mapping Failed: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()