"""Compile model-facing Authoring CDSL into server-owned runtime CDSL.""" from __future__ import annotations from copy import deepcopy from hashlib import sha256 import json from typing import Any, Callable from jsonschema import Draft202012Validator from .authoring_contract import AuthoringDocument, validate_finite, validation_error_code class AuthoringCompileError(ValueError): def __init__(self, code: str, message: str, *, path: str = "") -> None: self.code, self.path = code, path super().__init__(message) class AuthoringCompiler: def __init__(self, operation_contract: Callable[[str], dict[str, Any]], *, version: str = "1") -> None: self.operation_contract = operation_contract self.version = version def compile(self, raw: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: try: validate_finite(raw) doc = AuthoringDocument.model_validate(raw) except Exception as error: raise AuthoringCompileError(validation_error_code(error), str(error)) from error source_features = [feature for body in doc.bodies for feature in body.features] dependencies, implicit_selector_dependencies = self._effective_dependencies(doc) ordered = self._topological(doc, dependencies) # Identity follows the source document, while execution follows the # dependency graph. A dependency reordering must never renumber IDs. feature_ids = {feature.name: f"feature_{index:03d}" for index, feature in enumerate(source_features, 1)} body_ids = {body.name: f"body_{index:03d}" for index, body in enumerate(doc.bodies, 1)} features: list[dict[str, Any]] = [] sketches: list[dict[str, Any]] = [] sketch_index = 0 source_positions = {feature.name: index for index, feature in enumerate(source_features, 1)} sketch_ids_by_feature = { feature.name: f"sketch_{source_positions[feature.name]:03d}" for feature in source_features if feature.sketch is not None } for feature in ordered: try: contract = self.operation_contract(feature.operation) except Exception as error: raise AuthoringCompileError("OPERATION_UNSUPPORTED", str(error), path=f"features.{feature.name}.operation") from error params = self._references( deepcopy(feature.params), contract, feature_ids, feature.name, sketch_ids_by_feature=sketch_ids_by_feature, ) self._validate_params(contract, params, feature.name) selectors = [self._selector(item, feature_ids, source_features) for item in feature.selectors] selector_policy = contract.get("selector_policy") or {"slot": None, "token_kind": None, "min_items": 0, "max_items": 0} fragment_shape = contract.get("fragment_shape") or {"sketch": "forbidden", "selector_tokens": "forbidden"} required_selectors = fragment_shape["selector_tokens"] == "required" runtime_feature_selectors: list[dict[str, Any]] = [] if required_selectors and not selector_policy["min_items"] <= len(selectors) <= selector_policy["max_items"]: raise AuthoringCompileError("SELECTOR_NOT_FOUND", f"operation {feature.operation} requires {selector_policy['min_items']}..{selector_policy['max_items']} selectors", path=f"features.{feature.name}.selectors") if not required_selectors and selectors: raise AuthoringCompileError("SELECTOR_KIND_MISMATCH", f"operation {feature.operation} does not accept selectors", path=f"features.{feature.name}.selectors") if required_selectors and any(item["kind"] != selector_policy["token_kind"] for item in selectors): raise AuthoringCompileError("SELECTOR_KIND_MISMATCH", f"operation {feature.operation} requires {selector_policy['token_kind']} selectors", path=f"features.{feature.name}.selectors") if required_selectors: slot = str(selector_policy["slot"]) if slot == "feature.selectors": runtime_feature_selectors = selectors elif slot.startswith("params."): parameter = slot.removeprefix("params.") if "." in parameter: raise AuthoringCompileError( "OPERATION_UNSUPPORTED", f"runtime has no safe selector binding for {slot}", path=f"features.{feature.name}.selectors", ) if parameter in params: raise AuthoringCompileError( "AUTHOR_FORBIDDEN_FIELD", f"{parameter} is server-injected from selectors", path=f"features.{feature.name}.params.{parameter}", ) params[parameter] = selectors[0] if len(selectors) == 1 else selectors else: raise AuthoringCompileError( "OPERATION_UNSUPPORTED", f"runtime has no safe selector binding for {slot}", path=f"features.{feature.name}.selectors", ) output = { "id": feature_ids[feature.name], "atomic_id": feature.operation, "depends_on": [feature_ids[name] for name in dependencies[feature.name]], "params": params, **({"selectors": runtime_feature_selectors} if runtime_feature_selectors else {}), } needs_sketch = fragment_shape["sketch"] == "required" if needs_sketch and feature.sketch is None: raise AuthoringCompileError("AUTHOR_SCHEMA_INVALID", f"operation {feature.operation} requires sketch", path=f"features.{feature.name}.sketch") if not needs_sketch and feature.sketch is not None: raise AuthoringCompileError("AUTHOR_SCHEMA_INVALID", f"operation {feature.operation} does not accept sketch", path=f"features.{feature.name}.sketch") if feature.sketch is not None: sketch_index += 1 sketch_id = f"sketch_{source_positions[feature.name]:03d}" sketches.append({"id": sketch_id, **self._runtime_sketch(feature.sketch.model_dump(mode="json"))}) output["sketch_id"] = sketch_id features.append(output) runtime = { "schema": "cad.runtime.v1", "schema_version": "1.0.0", "kind": "part", "part_id": "compiled", "meta": {"unit": "mm"}, "bodies": [ {"id": body_ids[body.name], "name": body.name} for body in doc.bodies ], "geometry": {"sketches": sketches}, "features": features, } digest = sha256(self._canonical_json(raw)).hexdigest() return runtime, { "schema_version": "cad.author.compile-audit.v1", "compiler_version": self.version, "source_sha256": digest, "body_ids": body_ids, "feature_ids": feature_ids, "sketch_ids": [item["id"] for item in sketches], "implicit_selector_dependencies": implicit_selector_dependencies, } @staticmethod def _effective_dependencies(doc: AuthoringDocument) -> tuple[dict[str, list[str]], dict[str, list[str]]]: """Make every declared selector source an auditable graph dependency. A selector is already an explicit local source reference. Requiring the author to repeat that same edge in a second field only creates a formatting failure; it does not add CAD intent. The compiler therefore adds the direct source edge deterministically and records it in the audit. It never selects a substitute topology element. """ by_name = {item.name: item for body in doc.bodies for item in body.features} dependencies: dict[str, list[str]] = {} implicit: dict[str, list[str]] = {} for feature in by_name.values(): values = list(feature.depends_on) additions: list[str] = [] for selector in feature.selectors: source = selector.source.split(".", 1)[0] if source not in by_name: raise AuthoringCompileError( "AUTHOR_REFERENCE_INVALID", f"unknown selector source: {selector.source}", path=f"features.{feature.name}.selectors", ) if source not in values: values.append(source) additions.append(source) dependencies[feature.name] = values if additions: implicit[feature.name] = additions return dependencies, implicit @staticmethod def _topological(doc: AuthoringDocument, dependencies: dict[str, list[str]]) -> list[Any]: by_name = {item.name: item for body in doc.bodies for item in body.features} result: list[Any] = [] visiting: set[str] = set() done: set[str] = set() def visit(name: str) -> None: if name in visiting: raise AuthoringCompileError("AUTHOR_CYCLE", f"cyclic feature dependency: {name}") if name in done: return visiting.add(name) for dependency in dependencies[name]: visit(dependency) visiting.remove(name); done.add(name); result.append(by_name[name]) for body in doc.bodies: for feature in body.features: visit(feature.name) return result @staticmethod def _selector(selector: Any, feature_ids: dict[str, str], source_features: list[Any]) -> dict[str, Any]: source = selector.source.split(".", 1)[0] if source not in feature_ids: raise AuthoringCompileError("AUTHOR_REFERENCE_INVALID", f"unknown selector source: {selector.source}") source_feature = next(item for item in source_features if item.name == source) role = selector.source.split(".", 1)[1] role = AuthoringCompiler._runtime_role(role, source_feature.operation) return {"kind": selector.kind, "output_role": role, "owner_feature_id": feature_ids[source], "source": "runtime_snapshot", "confidence": 1.0, "match_mode": selector.match} @staticmethod def _runtime_sketch(sketch: dict[str, Any]) -> dict[str, Any]: """Lower the small Authoring sketch language to the generic runtime form.""" profile = sketch["profile"] if profile["type"] == "circle": profile = { "type": "circle", "center": profile["center_mm"], "radius_mm": float(profile["diameter_mm"]) / 2.0, } return {"workplane": sketch["workplane"], "profile": profile} @staticmethod def _runtime_role(role: str, operation: str) -> str: if role in {"extrude.start", "extrude.end", "sweep.start", "sweep.end", "loft.start", "loft.end", "cylinder.start", "cylinder.end", "shell.offset_face", "shell.closing_descendant", "shell.body_face"}: return role family = ( "extrude" if operation.startswith("extrude") else "sweep" if operation.startswith("sweep") else "loft" if operation.startswith("loft") else "cylinder" if operation == "cylinder_add" else "" ) if role in {"top_planar_face", "end_face"} and family: return family + ".end" if role in {"bottom_planar_face", "start_face"} and family: return family + ".start" raise AuthoringCompileError("AUTHOR_REFERENCE_INVALID", f"selector output role {role!r} is not available from {operation}") @staticmethod def _validate_params(contract: dict[str, Any], params: dict[str, Any], feature_name: str) -> None: schema = contract.get("author_params_schema") or {"type": "object"} errors = list(Draft202012Validator(schema).iter_errors(params)) if errors: first = errors[0] path = ".".join(str(item) for item in first.absolute_path) raise AuthoringCompileError("AUTHOR_SCHEMA_INVALID", first.message, path=f"features.{feature_name}.params.{path}".rstrip(".")) @staticmethod def _references( params: dict[str, Any], contract: dict[str, Any], feature_ids: dict[str, str], feature_name: str, *, sketch_ids_by_feature: dict[str, str], ) -> dict[str, Any]: params = AuthoringCompiler._rewrite_named_references( params, feature_ids, sketch_ids_by_feature, feature_name, ) policy = contract.get("reference_policy") or {"mode": "none"} if policy["mode"] != "snapshot_bound" or policy.get("slot") == "feature.selectors": return params key = str(policy["slot"]).removeprefix("params.") value = params.get(key) if isinstance(value, list): names = value elif value is None and policy["min_items"] == 0: names = [] elif isinstance(value, str): names = [value] else: raise AuthoringCompileError("AUTHOR_REFERENCE_INVALID", f"{key} must contain local feature names", path=f"features.{feature_name}.params.{key}") runtime_ids = set(feature_ids.values()) if not all(isinstance(item, str) and item in runtime_ids for item in names): raise AuthoringCompileError("AUTHOR_REFERENCE_INVALID", f"{key} contains an unknown local feature name", path=f"features.{feature_name}.params.{key}") if not policy["min_items"] <= len(names) <= policy["max_items"]: raise AuthoringCompileError("AUTHOR_REFERENCE_INVALID", f"{key} has invalid item count", path=f"features.{feature_name}.params.{key}") return params @staticmethod def _rewrite_named_references( value: Any, feature_ids: dict[str, str], sketch_ids_by_feature: dict[str, str], consuming_feature: str, key: str = "", ) -> Any: if isinstance(value, dict): return { item_key: AuthoringCompiler._rewrite_named_references( item_value, feature_ids, sketch_ids_by_feature, consuming_feature, item_key, ) for item_key, item_value in value.items() } if isinstance(value, list): return [ AuthoringCompiler._rewrite_named_references( item, feature_ids, sketch_ids_by_feature, consuming_feature, key, ) for item in value ] if not isinstance(value, str): return value if key == "profile_sketch_ids": if value not in sketch_ids_by_feature: raise AuthoringCompileError( "AUTHOR_REFERENCE_INVALID", f"unknown local sketch source: {value}", path=f"features.{consuming_feature}.params.{key}", ) return sketch_ids_by_feature[value] if key.endswith("_feature_id") or key.endswith("_feature_ids"): if value not in feature_ids: raise AuthoringCompileError( "AUTHOR_REFERENCE_INVALID", f"unknown local feature reference: {value}", path=f"features.{consuming_feature}.params.{key}", ) return feature_ids[value] return value @staticmethod def _canonical_json(value: Any) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")