308 lines
16 KiB
Python
308 lines
16 KiB
Python
"""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
|
|
if feature.intent is not None:
|
|
# Semantic annotation only: carried through for training data,
|
|
# never consumed by runtime geometry or validation semantics.
|
|
output["intent"] = feature.intent.model_dump(mode="json", exclude_none=True)
|
|
features.append(output)
|
|
document_meta: dict[str, Any] = {"unit": "mm"}
|
|
if doc.meta is not None:
|
|
document_meta.update(doc.meta.model_dump(mode="json", exclude_none=True))
|
|
runtime = {
|
|
"schema": "cad.runtime.v1", "schema_version": "1.0.0", "kind": "part",
|
|
"part_id": "compiled", "meta": document_meta,
|
|
"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")
|