#!/usr/bin/env python3 """Compile and validate DesignIR 2.0 without depending on teacher geometry.""" from __future__ import annotations import argparse import ast import copy import json import math import re import shutil import sys import tempfile from pathlib import Path from typing import Any from build123d import Align, Box, CenterOf, Cylinder, Part, Pos, export_step, import_step SCHEMA_VERSION = "2.0" DESIGNIR_KIND = "executable_parametric_design" SUPPORTED_OPERATIONS = { "extrude_circle", "extrude_rectangle", "add_cylinder", "through_hole", "polar_hole_pattern", } REQUIRED_TOP_LEVEL = { "schema_version", "designir_kind", "model_id", "family", "units", "reconstruction_status", "coordinate_system", "datums", "parameters", "expressions", "sketches", "constraints", "features", "patterns", "attachments", "construction_stages", "edit_interface", "validation_contract", } FORBIDDEN_KEYS = { "source_step", "teacher_step", "source_path", "brep", "b_rep", "step_text", "mesh", "triangles", "vertices", "face_refs", "surface_ids", "topology_refs", "base_step", } FORBIDDEN_TEXT = ( "source.step", "teacher.step", ".stp", ".step", "import_step", "read_step", "stepcontrol_reader", ) SAFE_EXPRESSION_NODES = ( ast.Expression, ast.BinOp, ast.UnaryOp, ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.Mod, ast.USub, ast.UAdd, ast.Constant, ast.Name, ast.Load, ) class DesignIRError(ValueError): pass 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 DesignIRError(f"Cannot read JSON {path}: {exc}") from exc if not isinstance(value, dict): raise DesignIRError("DesignIR 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) path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def find_forbidden(value: Any, trail: str = "$") -> str | None: if isinstance(value, dict): for key, child in value.items(): normalized = str(key).lower() if normalized in FORBIDDEN_KEYS: return f"{trail}.{key}" found = find_forbidden(child, f"{trail}.{key}") if found: return found elif isinstance(value, list): for index, child in enumerate(value): found = find_forbidden(child, f"{trail}[{index}]") if found: return found elif isinstance(value, str): lowered = value.lower() if any(token in lowered for token in FORBIDDEN_TEXT): return trail return None def _vector3(value: Any, label: str) -> None: if ( not isinstance(value, list) or len(value) != 3 or any(not isinstance(item, (int, float)) for item in value) ): raise DesignIRError(f"{label} must be a three-number vector") def validate_designir(payload: dict[str, Any]) -> dict[str, Any]: missing = sorted(REQUIRED_TOP_LEVEL - set(payload)) if missing: raise DesignIRError("Missing DesignIR fields: " + ", ".join(missing)) if payload["schema_version"] != SCHEMA_VERSION: raise DesignIRError("DesignIR schema_version must be 2.0") if payload["designir_kind"] != DESIGNIR_KIND: raise DesignIRError(f"designir_kind must be {DESIGNIR_KIND}") if payload["units"] != "mm": raise DesignIRError("Only millimetres are supported") if payload["reconstruction_status"] not in { "ready", "partial", "unsupported_feature_vocabulary", }: raise DesignIRError("Invalid reconstruction_status") forbidden = find_forbidden(payload) if forbidden: raise DesignIRError(f"Teacher geometry dependency is forbidden at {forbidden}") coordinate_system = payload["coordinate_system"] if not isinstance(coordinate_system, dict): raise DesignIRError("coordinate_system must be an object") for name in ("origin", "x_axis", "y_axis", "z_axis"): _vector3(coordinate_system.get(name), f"coordinate_system.{name}") parameters = payload["parameters"] if not isinstance(parameters, dict): raise DesignIRError("parameters must be an object") for name, parameter in parameters.items(): if not isinstance(parameter, dict) or not isinstance( parameter.get("value"), (int, float) ): raise DesignIRError(f"Parameter {name} requires a numeric value") if parameter.get("unit") not in {"mm", "deg", "count", "ratio"}: raise DesignIRError(f"Parameter {name} has an unsupported unit") if not isinstance(parameter.get("editable"), bool): raise DesignIRError(f"Parameter {name} requires editable=true/false") features = payload["features"] if not isinstance(features, list) or not features: raise DesignIRError("features must be a non-empty array") feature_ids: set[str] = set() for feature in features: if not isinstance(feature, dict): raise DesignIRError("Every feature must be an object") feature_id = str(feature.get("id", "")).strip() if not feature_id or feature_id in feature_ids: raise DesignIRError("Every feature requires a unique id") feature_ids.add(feature_id) operation = feature.get("operation") if operation not in SUPPORTED_OPERATIONS: raise DesignIRError(f"Unsupported DesignIR operation: {operation}") editable = payload["edit_interface"].get("editable_parameters", []) unknown = sorted(set(editable) - set(parameters)) if unknown: raise DesignIRError("Unknown editable parameters: " + ", ".join(unknown)) for test in payload["validation_contract"].get("perturbations", []): if test.get("parameter") not in parameters: raise DesignIRError( f"Perturbation references unknown parameter {test.get('parameter')}" ) return payload def _safe_expression(expression: str, values: dict[str, float]) -> float: try: tree = ast.parse(expression, mode="eval") except SyntaxError as exc: raise DesignIRError(f"Invalid expression {expression!r}") from exc if not all(isinstance(node, SAFE_EXPRESSION_NODES) for node in ast.walk(tree)): raise DesignIRError(f"Unsafe expression {expression!r}") unknown = { node.id for node in ast.walk(tree) if isinstance(node, ast.Name) and node.id not in values } if unknown: raise DesignIRError("Unknown expression symbols: " + ", ".join(sorted(unknown))) return float(eval(compile(tree, "", "eval"), {"__builtins__": {}}, values)) def parameter_values(payload: dict[str, Any]) -> dict[str, float]: values = { name: float(parameter["value"]) for name, parameter in payload["parameters"].items() } pending = dict(payload.get("expressions", {})) while pending: progressed = False for name, expression in list(pending.items()): try: values[name] = _safe_expression(str(expression), values) except DesignIRError as exc: if "Unknown expression symbols" in str(exc): continue raise del pending[name] progressed = True if not progressed: raise DesignIRError( "Unresolved or cyclic expressions: " + ", ".join(sorted(pending)) ) return values def resolve_value(value: Any, values: dict[str, float], label: str) -> float: if isinstance(value, (int, float)): return float(value) if isinstance(value, dict) and set(value) == {"parameter"}: name = str(value["parameter"]) if name not in values: raise DesignIRError(f"{label} references unknown parameter {name}") return values[name] if isinstance(value, dict) and set(value) == {"expression"}: name = str(value["expression"]) if name not in values: raise DesignIRError(f"{label} references unknown expression {name}") return values[name] raise DesignIRError(f"{label} must be a number, parameter, or expression") def _axis_is_z(feature: dict[str, Any]) -> None: if feature.get("axis", "primary_axis") not in {"z", "primary_axis"}: raise DesignIRError("The initial compiler supports only the primary Z axis") def build_shape(payload: dict[str, Any]) -> Part: validate_designir(payload) if payload["reconstruction_status"] != "ready": missing = ", ".join(payload.get("missing_capabilities", [])) or "unspecified" raise DesignIRError( f"DesignIR is not independently rebuildable; missing capabilities: {missing}" ) values = parameter_values(payload) shape: Any = None for feature in payload["features"]: operation = feature["operation"] _axis_is_z(feature) if operation in {"extrude_circle", "add_cylinder"}: radius = resolve_value(feature["radius"], values, f"{feature['id']}.radius") height = resolve_value(feature["height"], values, f"{feature['id']}.height") center = feature.get("center", [0, 0, 0]) _vector3(center, f"{feature['id']}.center") tool = Pos(*map(float, center)) * Cylinder( radius, height, align=(Align.CENTER, Align.CENTER, Align.MIN), ) shape = tool if shape is None else shape + tool elif operation == "extrude_rectangle": width = resolve_value(feature["width"], values, f"{feature['id']}.width") depth = resolve_value(feature["depth"], values, f"{feature['id']}.depth") height = resolve_value(feature["height"], values, f"{feature['id']}.height") center = feature.get("center", [0, 0, 0]) _vector3(center, f"{feature['id']}.center") tool = Pos(*map(float, center)) * Box( width, depth, height, align=(Align.CENTER, Align.CENTER, Align.MIN), ) shape = tool if shape is None else shape + tool elif operation in {"through_hole", "polar_hole_pattern"}: if shape is None: raise DesignIRError(f"{feature['id']} requires a host solid") diameter = resolve_value( feature["diameter"], values, f"{feature['id']}.diameter" ) bounds = shape.bounding_box() cutter_height = float(bounds.size.Z) + 2.0 cutter_z = float(bounds.min.Z) - 1.0 centers: list[tuple[float, float]] if operation == "through_hole": center = feature.get("center", [0, 0, 0]) _vector3(center, f"{feature['id']}.center") centers = [(float(center[0]), float(center[1]))] else: count = int( round(resolve_value(feature["count"], values, f"{feature['id']}.count")) ) pitch = resolve_value( feature["pitch_diameter"], values, f"{feature['id']}.pitch_diameter", ) if count < 1: raise DesignIRError(f"{feature['id']}.count must be positive") centers = [ ( math.cos(2.0 * math.pi * index / count) * pitch / 2.0, math.sin(2.0 * math.pi * index / count) * pitch / 2.0, ) for index in range(count) ] for x, y in centers: cutter = Pos(x, y, cutter_z) * Cylinder( diameter / 2.0, cutter_height, align=(Align.CENTER, Align.CENTER, Align.MIN), ) shape = shape - cutter if shape is None: raise DesignIRError("DesignIR produced no geometry") result = Part(shape.wrapped) if not list(result.solids()) or result.volume <= 0: raise DesignIRError("DesignIR did not produce a valid solid") return result def build_simplecad_shape(payload: dict[str, Any]) -> tuple[Any, str]: validate_designir(payload) if payload["reconstruction_status"] != "ready": missing = ", ".join(payload.get("missing_capabilities", [])) or "unspecified" raise DesignIRError( f"DesignIR is not independently rebuildable; missing capabilities: {missing}" ) simplecad_src = Path(__file__).resolve().parents[2] / "SimpleCADAPI" / "src" if str(simplecad_src) not in sys.path: sys.path.insert(0, str(simplecad_src)) try: import simplecadapi as scad except ImportError as exc: raise DesignIRError("SimpleCADAPI backend is unavailable") from exc values = parameter_values(payload) shape: Any = None z_min: float | None = None z_max: float | None = None with scad.GraphSession() as session: for feature in payload["features"]: operation = feature["operation"] _axis_is_z(feature) if operation in {"extrude_circle", "add_cylinder"}: radius = resolve_value( feature["radius"], values, f"{feature['id']}.radius" ) height = resolve_value( feature["height"], values, f"{feature['id']}.height" ) center = feature.get("center", [0, 0, 0]) _vector3(center, f"{feature['id']}.center") point = tuple(map(float, center)) tool = scad.make_cylinder_rsolid( radius=radius, height=height, bottom_face_center=point, axis=(0.0, 0.0, 1.0), ) shape = ( tool if shape is None else scad.union_rsolid([shape, tool], clean=True) ) z_min = point[2] if z_min is None else min(z_min, point[2]) z_max = ( point[2] + height if z_max is None else max(z_max, point[2] + height) ) elif operation == "extrude_rectangle": width = resolve_value( feature["width"], values, f"{feature['id']}.width" ) depth = resolve_value( feature["depth"], values, f"{feature['id']}.depth" ) height = resolve_value( feature["height"], values, f"{feature['id']}.height" ) center = feature.get("center", [0, 0, 0]) _vector3(center, f"{feature['id']}.center") point = tuple(map(float, center)) tool = scad.make_box_rsolid( width=width, height=depth, depth=height, bottom_face_center=point, ) shape = ( tool if shape is None else scad.union_rsolid([shape, tool], clean=True) ) z_min = point[2] if z_min is None else min(z_min, point[2]) z_max = ( point[2] + height if z_max is None else max(z_max, point[2] + height) ) elif operation in {"through_hole", "polar_hole_pattern"}: if shape is None or z_min is None or z_max is None: raise DesignIRError(f"{feature['id']} requires a host solid") diameter = resolve_value( feature["diameter"], values, f"{feature['id']}.diameter" ) if operation == "through_hole": center = feature.get("center", [0, 0, 0]) _vector3(center, f"{feature['id']}.center") centers = [(float(center[0]), float(center[1]))] else: count = int( round( resolve_value( feature["count"], values, f"{feature['id']}.count" ) ) ) pitch = resolve_value( feature["pitch_diameter"], values, f"{feature['id']}.pitch_diameter", ) if count < 1: raise DesignIRError( f"{feature['id']}.count must be positive" ) centers = [ ( math.cos(2.0 * math.pi * index / count) * pitch / 2.0, math.sin(2.0 * math.pi * index / count) * pitch / 2.0, ) for index in range(count) ] cutter_height = z_max - z_min + 2.0 cutters = [ scad.make_cylinder_rsolid( radius=diameter / 2.0, height=cutter_height, bottom_face_center=(x, y, z_min - 1.0), axis=(0.0, 0.0, 1.0), ) for x, y in centers ] shape = scad.cut_rsolid( shape, cutters, skip_non_intersecting=False ) print( f"simplecad grounding {feature['id']}: volume={shape.get_volume():.6f}", file=sys.stderr, ) model_json = scad.export_model_json(session=session, indent=2) if shape is None or shape.get_volume() <= 0: raise DesignIRError("DesignIR did not produce a valid SimpleCAD solid") return shape, model_json def geometry_facts(shape: Any) -> dict[str, Any]: bounds = shape.bounding_box() center = shape.center(CenterOf.MASS) return { "solid_count": len(list(shape.solids())), "face_count": len(list(shape.faces())), "edge_count": len(list(shape.edges())), "vertex_count": len(list(shape.vertices())), "bounds_min_mm": [float(bounds.min.X), float(bounds.min.Y), float(bounds.min.Z)], "bounds_max_mm": [float(bounds.max.X), float(bounds.max.Y), float(bounds.max.Z)], "size_mm": [float(bounds.size.X), float(bounds.size.Y), float(bounds.size.Z)], "volume_mm3": float(shape.volume), "center_of_mass_mm": [float(center.X), float(center.Y), float(center.Z)], } def compile_designir( payload: dict[str, Any], output: Path, backend: str | None = None ) -> dict[str, Any]: selected_backend = backend or str(payload.get("backend_hint", "build123d")) output = output.expanduser().resolve() output.parent.mkdir(parents=True, exist_ok=True) model_json_path: Path | None = None if selected_backend == "build123d": shape = build_shape(payload) export_step(shape, output) elif selected_backend == "simplecadapi": shape, model_json = build_simplecad_shape(payload) from simplecadapi import export_step as simplecad_export_step simplecad_export_step(shapes=shape, filename=str(output)) model_json_path = output.with_suffix(".simplecad.model.json") model_json_path.write_text(model_json, encoding="utf-8") else: raise DesignIRError(f"Unsupported DesignIR compiler backend: {selected_backend}") rebuilt = import_step(output) return { "output": str(output), "backend": selected_backend, "model_json": str(model_json_path) if model_json_path else None, "facts": geometry_facts(rebuilt), } def anti_cheat(payload: dict[str, Any], generator: Path | None = None) -> dict[str, Any]: validate_designir(payload) violations: list[str] = [] forbidden = find_forbidden(payload) if forbidden: violations.append(f"DesignIR teacher dependency at {forbidden}") if generator: text = generator.read_text(encoding="utf-8").lower() for token in FORBIDDEN_TEXT + ("breptools", "stl", "trimesh"): if token in text: violations.append(f"Generator contains forbidden token {token}") return {"valid": not violations, "violations": violations} def isolated_rebuild( designir_path: Path, output: Path, backend: str | None = None ) -> dict[str, Any]: payload = read_json(designir_path) guard = anti_cheat(payload) if not guard["valid"]: raise DesignIRError("; ".join(guard["violations"])) with tempfile.TemporaryDirectory(prefix="designir-isolated-") as temporary: isolated_root = Path(temporary) isolated_designir = isolated_root / "model.designir.json" write_json(isolated_designir, payload) isolated_payload = read_json(isolated_designir) result = compile_designir( isolated_payload, isolated_root / "rebuilt.step", backend=backend ) final_output = output.expanduser().resolve() final_output.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(result["output"], final_output) final_model_json: Path | None = None if result.get("model_json"): final_model_json = final_output.with_suffix(".simplecad.model.json") shutil.copy2(str(result["model_json"]), final_model_json) return { "valid": True, "teacher_step_available": False, "designir_only": True, "backend": result["backend"], "output": str(final_output), "model_json": str(final_model_json) if final_model_json else None, "facts": geometry_facts(import_step(final_output)), } def _max_axis_error(first: list[float], second: list[float]) -> float: return max(abs(a - b) for a, b in zip(first, second)) def _distance(first: list[float], second: list[float]) -> float: return math.sqrt(sum((a - b) ** 2 for a, b in zip(first, second))) def _symmetric_difference_volume(first: Part, second: Part) -> float: try: return float((first - second).volume + (second - first).volume) except Exception as exc: # OpenCascade failures become an explicit metric state. raise DesignIRError(f"Symmetric-difference boolean failed: {exc}") from exc def acceptance_report( teacher_path: Path, rebuilt_path: Path, payload: dict[str, Any], ) -> dict[str, Any]: validate_designir(payload) teacher = import_step(teacher_path.expanduser().resolve()) rebuilt = import_step(rebuilt_path.expanduser().resolve()) teacher_facts = geometry_facts(teacher) rebuilt_facts = geometry_facts(rebuilt) volume_base = max(teacher_facts["volume_mm3"], 1e-12) symmetric_difference = _symmetric_difference_volume(teacher, rebuilt) geometry = { "teacher": teacher_facts, "rebuilt": rebuilt_facts, "relative_volume_error": abs( rebuilt_facts["volume_mm3"] - teacher_facts["volume_mm3"] ) / volume_base, "center_of_mass_error_mm": _distance( rebuilt_facts["center_of_mass_mm"], teacher_facts["center_of_mass_mm"], ), "bounding_box_axis_error_mm": _max_axis_error( rebuilt_facts["size_mm"], teacher_facts["size_mm"] ), "symmetric_difference_volume_mm3": symmetric_difference, "symmetric_difference_ratio": symmetric_difference / volume_base, "topology_count_delta": { key: rebuilt_facts[key] - teacher_facts[key] for key in ("solid_count", "face_count", "edge_count", "vertex_count") }, } perturbations: list[dict[str, Any]] = [] with tempfile.TemporaryDirectory(prefix="designir-edits-") as temporary: for index, test in enumerate( payload["validation_contract"].get("perturbations", []) ): modified = copy.deepcopy(payload) name = test["parameter"] original = float(modified["parameters"][name]["value"]) modified["parameters"][name]["value"] = original * float(test["scale"]) try: output = Path(temporary) / f"perturbation-{index}.step" result = compile_designir(modified, output) perturbations.append( { "parameter": name, "original_value": original, "modified_value": modified["parameters"][name]["value"], "expected_change": test.get("expected_change"), "preserve": test.get("preserve", []), "success": result["facts"]["solid_count"] > 0, "facts": result["facts"], } ) except Exception as exc: perturbations.append( {"parameter": name, "success": False, "error": str(exc)} ) success_rate = ( sum(1 for item in perturbations if item["success"]) / len(perturbations) if perturbations else 1.0 ) thresholds = payload["validation_contract"].get("thresholds", {}) threshold_checks = { "relative_volume_error": geometry["relative_volume_error"] <= float(thresholds.get("relative_volume_error_max", 0.01)), "center_of_mass_error_mm": geometry["center_of_mass_error_mm"] <= float(thresholds.get("center_of_mass_error_mm_max", 0.1)), "bounding_box_axis_error_mm": geometry["bounding_box_axis_error_mm"] <= float(thresholds.get("bounding_box_axis_error_mm_max", 0.1)), "symmetric_difference_ratio": geometry["symmetric_difference_ratio"] <= float(thresholds.get("symmetric_difference_ratio_max", 0.02)), "parameter_perturbation_success_rate": success_rate >= float(thresholds.get("parameter_perturbation_success_rate_min", 1.0)), } passed = all(threshold_checks.values()) return { "schema_version": "1.0", "report_kind": "independent_designir_acceptance", "model_id": payload["model_id"], "geometry": geometry, "feature_semantics": { "declared_feature_count": len(payload["features"]), "declared_constraint_count": len(payload["constraints"]), "unsupported_capabilities": payload.get("missing_capabilities", []), }, "editability": { "editable_parameter_count": len( payload["edit_interface"]["editable_parameters"] ), "perturbations": perturbations, "parameter_perturbation_success_rate": success_rate, }, "threshold_checks": threshold_checks, "verdict": "pass" if passed else "fail", "next_state": "replay_validated" if passed else "quarantine", } def promotion_check(reports: list[Path], policy_path: Path) -> dict[str, Any]: policy = read_json(policy_path) loaded = [read_json(path) for path in reports] model_ids = {str(item.get("model_id")) for item in loaded} passing = [item for item in loaded if item.get("next_state") == "replay_validated"] minimum = int(policy["minimum_independent_cases"]) eligible = len(passing) >= minimum and len(model_ids) >= minimum return { "eligible": eligible, "independent_case_count": len(model_ids), "passing_case_count": len(passing), "required_case_count": minimum, "next_state": "promoted" if eligible else "candidate", } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) validate = commands.add_parser("validate") validate.add_argument("designir", type=Path) guard = commands.add_parser("anti-cheat") guard.add_argument("designir", type=Path) guard.add_argument("--generator", type=Path) compile_command = commands.add_parser("compile") compile_command.add_argument("designir", type=Path) compile_command.add_argument("--output", type=Path, required=True) compile_command.add_argument( "--backend", choices=("build123d", "simplecadapi") ) isolated = commands.add_parser("isolated-rebuild") isolated.add_argument("designir", type=Path) isolated.add_argument("--output", type=Path, required=True) isolated.add_argument("--backend", choices=("build123d", "simplecadapi")) accept = commands.add_parser("accept") accept.add_argument("--teacher", type=Path, required=True) accept.add_argument("--designir", type=Path, required=True) accept.add_argument("--rebuilt", type=Path, required=True) accept.add_argument("--report", type=Path, required=True) promote = commands.add_parser("promotion-check") promote.add_argument("reports", type=Path, nargs="+") promote.add_argument("--policy", type=Path, required=True) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: if args.command == "validate": payload = validate_designir(read_json(args.designir)) result = { "valid": True, "model_id": payload["model_id"], "supported_operations": sorted(SUPPORTED_OPERATIONS), } elif args.command == "anti-cheat": result = anti_cheat(read_json(args.designir), args.generator) elif args.command == "compile": result = compile_designir( read_json(args.designir), args.output, backend=args.backend ) elif args.command == "isolated-rebuild": result = isolated_rebuild( args.designir, args.output, backend=args.backend ) elif args.command == "accept": result = acceptance_report( args.teacher, args.rebuilt, read_json(args.designir) ) write_json(args.report, result) else: result = promotion_check(args.reports, args.policy) except (DesignIRError, OSError) as exc: print(json.dumps({"valid": False, "error": str(exc)}, ensure_ascii=False, indent=2)) return 2 print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result.get("valid", result.get("eligible", True)) is not False else 1 if __name__ == "__main__": raise SystemExit(main())