#!/usr/bin/env python3 """Create and execute per-part JSON model specifications. Two reconstruction modes are supported: * ``native_generator`` loads a task-local Python generator. The generator reads the same model-spec.json, so named parameter edits remain the source of truth. * ``exact_step_base`` imports an immutable task-local STEP file and applies a parametric modification layer. With no modifications, the imported B-Rep is preserved geometrically even though the exported STEP text may differ. """ from __future__ import annotations import argparse import hashlib import importlib.util import json import math import re import shutil import subprocess import sys from pathlib import Path from typing import Any from build123d import ( Align, Box, Cylinder, Part, Pos, export_step, export_stl, import_step, ) SCHEMA_VERSION = "1.0" MODEL_SPEC_KIND = "parametric_cad_model" MODEL_ID_RE = re.compile(r"[^a-zA-Z0-9._-]+") SUPPORTED_OPERATIONS = { "add_box", "cut_box", "add_cylinder", "cut_cylinder", } class ModelSpecError(ValueError): """Raised when a model specification is invalid or unsafe to execute.""" def _read_json(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ModelSpecError(f"Cannot read model spec {path}: {exc}") from exc if not isinstance(value, dict): raise ModelSpecError("Model spec must be a JSON object") return value def _write_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) temporary.replace(path) def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _safe_model_id(value: str) -> str: result = MODEL_ID_RE.sub("_", value.strip()).strip("._-") if not result: raise ModelSpecError("model_id must contain a letter or number") return result def _resolve_relative(spec_path: Path, value: str, field: str) -> Path: candidate = Path(value) if candidate.is_absolute(): raise ModelSpecError(f"{field} must be task-relative, not absolute") return (spec_path.parent / candidate).resolve() def _parameter_value(spec: dict[str, Any], name: str) -> float: parameters = spec.get("parameters", {}) entry = parameters.get(name) if isinstance(parameters, dict) else None if not isinstance(entry, dict) or "value" not in entry: raise ModelSpecError(f"Unknown parameter reference: {name}") value = entry["value"] if isinstance(value, bool) or not isinstance(value, (int, float)): raise ModelSpecError(f"Parameter {name} must contain a numeric value") if not math.isfinite(float(value)): raise ModelSpecError(f"Parameter {name} must be finite") return float(value) def resolve_number(spec: dict[str, Any], value: Any, field: str) -> float: if isinstance(value, bool): raise ModelSpecError(f"{field} must be numeric") if isinstance(value, (int, float)): result = float(value) elif isinstance(value, dict) and "parameter" in value: unexpected = set(value) - {"parameter", "scale", "offset"} if unexpected: raise ModelSpecError( f"{field} parameter expression has unsupported keys: " + ", ".join(sorted(unexpected)) ) scale = value.get("scale", 1.0) offset = value.get("offset", 0.0) if ( isinstance(scale, bool) or not isinstance(scale, (int, float)) or isinstance(offset, bool) or not isinstance(offset, (int, float)) ): raise ModelSpecError(f"{field} scale and offset must be numeric") result = ( _parameter_value(spec, str(value["parameter"])) * float(scale) + float(offset) ) else: raise ModelSpecError( f"{field} must be a number or a parameter expression" ) if not math.isfinite(result): raise ModelSpecError(f"{field} must be finite") return result def _vector3(spec: dict[str, Any], value: Any, field: str) -> tuple[float, float, float]: if not isinstance(value, list) or len(value) != 3: raise ModelSpecError(f"{field} must contain three values") return tuple( resolve_number(spec, item, f"{field}[{index}]") for index, item in enumerate(value) ) def validate_model_spec(spec: dict[str, Any]) -> None: if spec.get("schema_version") != SCHEMA_VERSION: raise ModelSpecError(f"Unsupported model spec schema: {spec.get('schema_version')}") if spec.get("model_spec_kind") != MODEL_SPEC_KIND: raise ModelSpecError("Not a parametric CAD model specification") _safe_model_id(str(spec.get("model_id", ""))) if spec.get("units") != "mm": raise ModelSpecError("V1 model specs use millimetres") reconstruction = spec.get("reconstruction") if not isinstance(reconstruction, dict): raise ModelSpecError("reconstruction must be an object") mode = reconstruction.get("mode") if mode not in {"native_generator", "exact_step_base"}: raise ModelSpecError(f"Unsupported reconstruction mode: {mode}") if mode == "native_generator": generator = reconstruction.get("generator") if not isinstance(generator, dict) or not generator.get("path"): raise ModelSpecError("native_generator requires generator.path") else: source = reconstruction.get("source") if ( not isinstance(source, dict) or not source.get("path") or not source.get("sha256") ): raise ModelSpecError("exact_step_base requires source.path and source.sha256") parameters = spec.get("parameters", {}) if not isinstance(parameters, dict): raise ModelSpecError("parameters must be an object") for name, entry in parameters.items(): if not isinstance(entry, dict) or "value" not in entry: raise ModelSpecError(f"Parameter {name} must be an object with value") _parameter_value(spec, name) modifications = spec.get("modifications", []) if not isinstance(modifications, list): raise ModelSpecError("modifications must be an array") seen_ids: set[str] = set() for index, modification in enumerate(modifications): if not isinstance(modification, dict): raise ModelSpecError(f"modifications[{index}] must be an object") feature_id = str(modification.get("id", "")).strip() if not feature_id or feature_id in seen_ids: raise ModelSpecError("Every modification requires a unique id") seen_ids.add(feature_id) operation = modification.get("operation") if operation not in SUPPORTED_OPERATIONS: raise ModelSpecError(f"Unsupported modification operation: {operation}") outputs = spec.get("outputs") if not isinstance(outputs, dict) or not outputs.get("step"): raise ModelSpecError("outputs.step is required") def geometry_facts(shape: Any) -> dict[str, Any]: box = shape.bounding_box() solids = list(shape.solids()) volume = sum(float(solid.volume) for solid in solids) return { "size_mm": [ round(float(box.size.X), 9), round(float(box.size.Y), 9), round(float(box.size.Z), 9), ], "center_mm": [ round(float(box.center().X), 9), round(float(box.center().Y), 9), round(float(box.center().Z), 9), ], "solid_count": len(solids), "face_count": len(shape.faces()), "edge_count": len(shape.edges()), "volume_mm3": round(volume, 6), } def _rotation_for_axis(axis: str) -> tuple[float, float, float]: normalized = axis.lower() if normalized == "z": return (0.0, 0.0, 0.0) if normalized == "x": return (0.0, 90.0, 0.0) if normalized == "y": return (-90.0, 0.0, 0.0) raise ModelSpecError(f"Unsupported cylinder axis: {axis}") def _make_tool(spec: dict[str, Any], modification: dict[str, Any]) -> Any: operation = str(modification["operation"]) center = _vector3(spec, modification.get("center", [0, 0, 0]), "center") if operation.endswith("_cylinder"): diameter = resolve_number(spec, modification.get("diameter"), "diameter") length = resolve_number(spec, modification.get("length"), "length") if diameter <= 0 or length <= 0: raise ModelSpecError("Cylinder diameter and length must be positive") tool = Cylinder( diameter / 2.0, length, align=(Align.CENTER, Align.CENTER, Align.CENTER), rotation=_rotation_for_axis(str(modification.get("axis", "z"))), ) else: size = _vector3(spec, modification.get("size"), "size") if any(item <= 0 for item in size): raise ModelSpecError("Box dimensions must be positive") tool = Box( *size, align=(Align.CENTER, Align.CENTER, Align.CENTER), ) return Pos(*center) * tool def _apply_modifications(shape: Any, spec: dict[str, Any]) -> Any: result = shape for modification in spec.get("modifications", []): if modification.get("enabled", True) is False: continue tool = _make_tool(spec, modification) operation = modification["operation"] result = result + tool if operation.startswith("add_") else result - tool return result def _load_generator(spec_path: Path, reconstruction: dict[str, Any]) -> Any: generator = reconstruction["generator"] source_path = _resolve_relative(spec_path, str(generator["path"]), "generator.path") if not source_path.is_file(): raise ModelSpecError(f"Generator does not exist: {source_path}") module_spec = importlib.util.spec_from_file_location( f"cad_model_{hash(source_path)}", source_path, ) if module_spec is None or module_spec.loader is None: raise ModelSpecError(f"Cannot load generator: {source_path}") module = importlib.util.module_from_spec(module_spec) module_spec.loader.exec_module(module) entrypoint = str(generator.get("entrypoint", "gen_step")) function = getattr(module, entrypoint, None) if not callable(function): raise ModelSpecError(f"Generator has no callable {entrypoint}: {source_path}") return function() def build_shape(spec_path: Path, spec: dict[str, Any] | None = None) -> Any: resolved_spec = spec_path.expanduser().resolve() payload = spec or _read_json(resolved_spec) validate_model_spec(payload) reconstruction = payload["reconstruction"] if reconstruction["mode"] == "native_generator": shape = _load_generator(resolved_spec, reconstruction) else: source = reconstruction["source"] source_path = _resolve_relative(resolved_spec, str(source["path"]), "source.path") if not source_path.is_file(): raise ModelSpecError(f"Exact STEP base does not exist: {source_path}") actual_hash = _sha256(source_path) if actual_hash != source["sha256"]: raise ModelSpecError( "Exact STEP base checksum changed; import it again instead of " "silently rebuilding from a different source" ) imported = import_step(source_path) # build123d's STEP importer returns a topology wrapper that is readable # and boolean-capable but is not always directly accepted by its STEP # exporter. Normalize it to a Part while preserving the wrapped B-Rep. shape = Part(imported.wrapped) return _apply_modifications(shape, payload) def build_model(spec_path: Path) -> dict[str, Any]: resolved_spec = spec_path.expanduser().resolve() payload = _read_json(resolved_spec) validate_model_spec(payload) outputs = payload["outputs"] step_path = _resolve_relative(resolved_spec, str(outputs["step"]), "outputs.step") step_path.parent.mkdir(parents=True, exist_ok=True) reconstruction = payload["reconstruction"] stl_value = outputs.get("stl") stl_path = ( _resolve_relative(resolved_spec, str(stl_value), "outputs.stl") if stl_value else None ) if reconstruction["mode"] == "native_generator": generator_path = _resolve_relative( resolved_spec, str(reconstruction["generator"]["path"]), "generator.path", ) if generator_path.with_suffix(".step").resolve() != step_path: raise ModelSpecError( "V1 native generator output must use the generator basename" ) cad_step = Path(__file__).resolve().parents[2] / "cad" / "scripts" / "step" command = [sys.executable, str(cad_step), str(generator_path), "--force"] if stl_path is not None: if stl_path.parent != generator_path.parent: raise ModelSpecError( "V1 native generator STL must remain in the task directory" ) command.extend(["--stl", stl_path.name]) completed = subprocess.run( command, cwd=resolved_spec.parent, text=True, capture_output=True, ) if completed.returncode != 0: raise ModelSpecError( "CAD generator failed: " + (completed.stderr.strip() or completed.stdout.strip()) ) if not step_path.is_file(): raise ModelSpecError(f"CAD generator did not write {step_path}") else: shape = build_shape(resolved_spec, payload) source_path = _resolve_relative( resolved_spec, str(reconstruction["source"]["path"]), "source.path", ) if not payload.get("modifications"): # Empty exact-base rebuilds preserve the STEP byte-for-byte. if source_path != step_path: shutil.copy2(source_path, step_path) else: export_step(shape, step_path) if stl_path is not None: stl_path.parent.mkdir(parents=True, exist_ok=True) export_stl(shape, stl_path) exported_shape = Part(import_step(step_path).wrapped) result: dict[str, Any] = { "model_spec": str(resolved_spec), "step": str(step_path), "facts": geometry_facts(exported_shape), } if stl_path is not None: result["stl"] = str(stl_path) return result def import_step_model( source_step: Path, task_dir: Path, model_id: str | None = None, ) -> dict[str, Any]: source = source_step.expanduser().resolve() if not source.is_file() or source.suffix.lower() not in {".step", ".stp"}: raise ModelSpecError(f"Input must be an existing STEP/STP file: {source}") target_dir = task_dir.expanduser().resolve() target_dir.mkdir(parents=True, exist_ok=True) normalized_id = _safe_model_id(model_id or source.stem) copied_source = target_dir / "source.step" if copied_source.resolve() != source: shutil.copy2(source, copied_source) source_shape = import_step(copied_source) source_facts = geometry_facts(source_shape) spec_path = target_dir / "model-spec.json" payload: dict[str, Any] = { "schema_version": SCHEMA_VERSION, "model_spec_kind": MODEL_SPEC_KIND, "model_id": normalized_id, "units": "mm", "reconstruction": { "mode": "exact_step_base", "source": { "path": "source.step", "sha256": _sha256(copied_source), "geometry_signature": source_facts, }, "history_recovery": { "status": "not_present_in_step", "contract": ( "Preserve the imported B-Rep exactly as the immutable base; " "represent later edits as named parametric modifications." ), }, }, "parameters": {}, "modifications": [], "outputs": { "step": f"{normalized_id}.step", "stl": f"{normalized_id}.stl", }, "validation": { "exact_base_required": True, "geometry_signature_tolerance_mm": 1e-7, }, } _write_json(spec_path, payload) result = build_model(spec_path) result["source"] = str(copied_source) result["source_sha256"] = payload["reconstruction"]["source"]["sha256"] return result def _numbers_close(first: Any, second: Any, tolerance: float) -> bool: if isinstance(first, list) and isinstance(second, list): return len(first) == len(second) and all( _numbers_close(left, right, tolerance) for left, right in zip(first, second) ) if isinstance(first, (int, float)) and isinstance(second, (int, float)): return math.isclose( float(first), float(second), rel_tol=tolerance, abs_tol=tolerance, ) return first == second def verify_model(spec_path: Path) -> dict[str, Any]: resolved_spec = spec_path.expanduser().resolve() payload = _read_json(resolved_spec) shape = build_shape(resolved_spec, payload) facts = geometry_facts(shape) reconstruction = payload["reconstruction"] result: dict[str, Any] = { "model_spec": str(resolved_spec), "valid": True, "facts": facts, "checks": [], } if reconstruction["mode"] == "exact_step_base" and not payload.get("modifications"): expected = reconstruction["source"].get("geometry_signature", {}) source_path = _resolve_relative( resolved_spec, str(reconstruction["source"]["path"]), "source.path", ) output_path = _resolve_relative( resolved_spec, str(payload["outputs"]["step"]), "outputs.step", ) output_shape = ( Part(import_step(output_path).wrapped) if output_path.is_file() else None ) output_facts = geometry_facts(output_shape) if output_shape is not None else {} tolerance = float( payload.get("validation", {}).get( "geometry_signature_tolerance_mm", 1e-7, ) ) signature_exact = all( _numbers_close(expected.get(key), facts.get(key), tolerance) for key in ( "size_mm", "center_mm", "solid_count", "face_count", "edge_count", "volume_mm3", ) ) exported_exact = ( output_path.is_file() and _sha256(output_path) == _sha256(source_path) and all( _numbers_close(expected.get(key), output_facts.get(key), tolerance) for key in ( "size_mm", "center_mm", "solid_count", "face_count", "edge_count", "volume_mm3", ) ) ) exact = signature_exact and exported_exact result["checks"].append( { "check": "exact_base_geometry_signature", "passed": exact, "expected": expected, "actual": facts, "exported": output_facts, "byte_identical_step": ( output_path.is_file() and _sha256(output_path) == _sha256(source_path) ), } ) result["valid"] = exact else: result["checks"].append( { "check": "model_spec_execution", "passed": True, "detail": "The parameterized model spec executed successfully.", } ) return result def set_parameter(spec_path: Path, name: str, value: float) -> dict[str, Any]: resolved_spec = spec_path.expanduser().resolve() payload = _read_json(resolved_spec) validate_model_spec(payload) parameters = payload["parameters"] if name not in parameters: raise ModelSpecError( f"Unknown parameter {name}; add a named parameter and feature binding first" ) parameters[name]["value"] = value validate_model_spec(payload) _write_json(resolved_spec, payload) return { "model_spec": str(resolved_spec), "parameter": name, "value": value, } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Create, build, modify, and verify per-part model-spec.json files." ) subparsers = parser.add_subparsers(dest="command", required=True) import_parser = subparsers.add_parser( "import-step", help="Create an exact STEP-base model spec in a task-owned directory.", ) import_parser.add_argument("source_step", type=Path) import_parser.add_argument("--task-dir", type=Path, required=True) import_parser.add_argument("--model-id") build_command = subparsers.add_parser( "build", help="Regenerate STEP/STL from a model specification.", ) build_command.add_argument("model_spec", type=Path) verify_command = subparsers.add_parser( "verify", help="Validate a model specification and exact-base signature.", ) verify_command.add_argument("model_spec", type=Path) set_command = subparsers.add_parser( "set", help="Change one existing named parameter in a model specification.", ) set_command.add_argument("model_spec", type=Path) set_command.add_argument("name") set_command.add_argument("value", type=float) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: if args.command == "import-step": result = import_step_model(args.source_step, args.task_dir, args.model_id) elif args.command == "build": result = build_model(args.model_spec) elif args.command == "verify": result = verify_model(args.model_spec) else: result = set_parameter(args.model_spec, args.name, args.value) except ModelSpecError as exc: print(json.dumps({"error": str(exc)}, ensure_ascii=False, indent=2)) return 2 print(json.dumps(result, ensure_ascii=False, indent=2)) if args.command == "verify" and result.get("valid") is False: return 1 return 0 if __name__ == "__main__": raise SystemExit(main())