Jerry
1237 lines
48 KiB
Python
1237 lines
48 KiB
Python
#!/usr/bin/env python3
|
|
"""Compile and validate semantic DesignIR 3.0 without teacher geometry.
|
|
|
|
Legacy DesignIR 2.0 documents remain accepted and can be migrated explicitly.
|
|
Surface-parametric DesignIR 3.0 documents are handled by surfaceir_pipeline.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import math
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
if str(SCRIPT_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPT_DIR))
|
|
|
|
from designir_codec import (
|
|
DesignIRCodecError,
|
|
read_designir,
|
|
write_designir,
|
|
)
|
|
|
|
from build123d import (
|
|
Align,
|
|
Box,
|
|
CenterOf,
|
|
Cylinder,
|
|
Part,
|
|
Plane,
|
|
Pos,
|
|
export_step,
|
|
import_step,
|
|
)
|
|
|
|
|
|
SCHEMA_VERSION = "3.0"
|
|
DESIGNIR_KIND = "independent_parametric_cad"
|
|
LEGACY_SCHEMA_VERSION = "2.0"
|
|
LEGACY_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 = read_designir(path)
|
|
except (OSError, json.JSONDecodeError, DesignIRCodecError) 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:
|
|
if path.name.endswith(".designir.json") and "surface_layer" in value:
|
|
write_designir(path, value)
|
|
return
|
|
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_compiler_payload(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"] != LEGACY_SCHEMA_VERSION:
|
|
raise DesignIRError("DesignIR schema_version must be 2.0")
|
|
if payload["designir_kind"] != LEGACY_DESIGNIR_KIND:
|
|
raise DesignIRError(f"designir_kind must be {LEGACY_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))
|
|
disconnected = sorted(
|
|
name
|
|
for name in editable
|
|
if not parameters[name].get("editable")
|
|
or _parameter_reference_count(features, name) == 0
|
|
)
|
|
if disconnected:
|
|
raise DesignIRError(
|
|
"Editable parameters must drive at least one feature: "
|
|
+ ", ".join(disconnected)
|
|
)
|
|
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')}"
|
|
)
|
|
if not str(test.get("expected_change") or "").strip():
|
|
raise DesignIRError("Every perturbation requires expected_change")
|
|
return payload
|
|
|
|
|
|
def migrate_designir_2_to_3(payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""Convert the legacy flat semantic contract into unified DesignIR 3.0."""
|
|
legacy = _validate_compiler_payload(copy.deepcopy(payload))
|
|
editable = list(legacy["edit_interface"].get("editable_parameters", []))
|
|
validation = legacy["validation_contract"]
|
|
semantic_layer = {
|
|
"reconstruction_status": legacy["reconstruction_status"],
|
|
"coordinate_system": legacy["coordinate_system"],
|
|
"datums": legacy["datums"],
|
|
"parameters": legacy["parameters"],
|
|
"expressions": legacy["expressions"],
|
|
"sketches": legacy["sketches"],
|
|
"constraints": legacy["constraints"],
|
|
"features": legacy["features"],
|
|
"patterns": legacy["patterns"],
|
|
"attachments": legacy["attachments"],
|
|
"construction_stages": legacy["construction_stages"],
|
|
"missing_capabilities": legacy.get("missing_capabilities", []),
|
|
"applied_experience_ids": legacy.get("applied_experience_ids", []),
|
|
}
|
|
return {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"designir_kind": DESIGNIR_KIND,
|
|
"model_id": legacy["model_id"],
|
|
"family": legacy["family"],
|
|
"units": legacy["units"],
|
|
"document_status": "geometry_present",
|
|
"reconstruction_mode": "fully_semantic_parametric",
|
|
"authoring_mode": "semantic_feature_program",
|
|
"backend_hint": legacy.get("backend_hint", "build123d"),
|
|
"semantic_layer": semantic_layer,
|
|
"edit_interface": {
|
|
"semantic_parameters": editable,
|
|
"surface_parameter_groups": [],
|
|
"modification_levels": ["semantic_feature"],
|
|
"preserved_interfaces": legacy["edit_interface"].get(
|
|
"preserved_interfaces", []
|
|
),
|
|
},
|
|
"validation_contract": {
|
|
"source_independence": True,
|
|
"geometry_checks": validation.get(
|
|
"geometry_checks",
|
|
["solid_count", "bounding_box", "volume", "center_of_mass"],
|
|
),
|
|
"edit_checks": validation.get(
|
|
"edit_checks",
|
|
["parameter_perturbation", "constraint_preservation"],
|
|
),
|
|
"invariants": validation.get("invariants", []),
|
|
"perturbations": validation.get("perturbations", []),
|
|
"thresholds": validation.get("thresholds", {}),
|
|
},
|
|
}
|
|
|
|
|
|
def compiler_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""Return the stable flat representation consumed by both CAD backends."""
|
|
version = payload.get("schema_version")
|
|
if version == LEGACY_SCHEMA_VERSION:
|
|
return _validate_compiler_payload(copy.deepcopy(payload))
|
|
if version != SCHEMA_VERSION or payload.get("designir_kind") != DESIGNIR_KIND:
|
|
raise DesignIRError(
|
|
"DesignIR must be legacy 2.0 or independent_parametric_cad 3.0"
|
|
)
|
|
mode = payload.get("reconstruction_mode")
|
|
if mode == "surface_parametric":
|
|
raise DesignIRError(
|
|
"Surface-parametric DesignIR must be rebuilt with surfaceir_pipeline.py"
|
|
)
|
|
if mode not in {
|
|
"fully_semantic_parametric",
|
|
"hybrid_semantic_surface_parametric",
|
|
}:
|
|
raise DesignIRError(f"Unsupported DesignIR 3.0 reconstruction_mode: {mode}")
|
|
if payload.get("units") != "mm":
|
|
raise DesignIRError("Only millimetres are supported")
|
|
forbidden = find_forbidden(payload.get("semantic_layer", {}))
|
|
if forbidden:
|
|
raise DesignIRError(f"Teacher geometry dependency is forbidden at {forbidden}")
|
|
semantic = payload.get("semantic_layer")
|
|
if not isinstance(semantic, dict):
|
|
raise DesignIRError("semantic_layer must be an object")
|
|
edit_interface = payload.get("edit_interface", {})
|
|
validation = payload.get("validation_contract", {})
|
|
flat = {
|
|
"schema_version": LEGACY_SCHEMA_VERSION,
|
|
"designir_kind": LEGACY_DESIGNIR_KIND,
|
|
"model_id": payload.get("model_id"),
|
|
"family": payload.get("family", "general_mechanical_part"),
|
|
"units": payload.get("units"),
|
|
"reconstruction_status": semantic.get("reconstruction_status", "ready"),
|
|
"coordinate_system": semantic.get("coordinate_system"),
|
|
"datums": semantic.get("datums", {}),
|
|
"parameters": semantic.get("parameters", {}),
|
|
"expressions": semantic.get("expressions", {}),
|
|
"sketches": semantic.get("sketches", []),
|
|
"constraints": semantic.get("constraints", []),
|
|
"features": semantic.get("features", []),
|
|
"patterns": semantic.get("patterns", []),
|
|
"attachments": semantic.get("attachments", []),
|
|
"construction_stages": semantic.get("construction_stages", []),
|
|
"edit_interface": {
|
|
"editable_parameters": edit_interface.get("semantic_parameters", []),
|
|
"preserved_interfaces": edit_interface.get("preserved_interfaces", []),
|
|
},
|
|
"validation_contract": {
|
|
"invariants": validation.get("invariants", []),
|
|
"perturbations": validation.get("perturbations", []),
|
|
"thresholds": validation.get("thresholds", {}),
|
|
},
|
|
"backend_hint": payload.get("backend_hint", "build123d"),
|
|
"missing_capabilities": semantic.get("missing_capabilities", []),
|
|
"applied_experience_ids": semantic.get("applied_experience_ids", []),
|
|
}
|
|
return _validate_compiler_payload(flat)
|
|
|
|
|
|
def validate_designir(payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""Validate either supported version and return its compiler representation."""
|
|
return compiler_payload(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, "<designir-expression>", "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 resolve_center(
|
|
value: Any, values: dict[str, float], label: str
|
|
) -> list[float]:
|
|
if not isinstance(value, list) or len(value) != 3:
|
|
raise DesignIRError(f"{label} must contain three values")
|
|
return [
|
|
resolve_value(component, values, f"{label}[{index}]")
|
|
for index, component in enumerate(value)
|
|
]
|
|
|
|
|
|
def feature_axis(payload: dict[str, Any], feature: dict[str, Any]) -> list[float]:
|
|
reference = feature.get("axis", "primary_axis")
|
|
if isinstance(reference, list):
|
|
axis = reference
|
|
elif reference in {"x", "y", "z"}:
|
|
axis = {
|
|
"x": [1.0, 0.0, 0.0],
|
|
"y": [0.0, 1.0, 0.0],
|
|
"z": [0.0, 0.0, 1.0],
|
|
}[reference]
|
|
else:
|
|
datum = payload.get("datums", {}).get(str(reference), {})
|
|
axis = datum.get("axis")
|
|
if isinstance(axis, str) and axis in {"x", "y", "z"}:
|
|
axis = {
|
|
"x": [1.0, 0.0, 0.0],
|
|
"y": [0.0, 1.0, 0.0],
|
|
"z": [0.0, 0.0, 1.0],
|
|
}[axis]
|
|
_vector3(axis, f"{feature.get('id', 'feature')}.axis")
|
|
length = math.sqrt(sum(float(value) ** 2 for value in axis))
|
|
if length <= 1e-12:
|
|
raise DesignIRError(f"{feature.get('id', 'feature')}.axis is zero")
|
|
return [float(value) / length for value in axis]
|
|
|
|
|
|
def _axis_is_z(axis: list[float]) -> bool:
|
|
return abs(axis[0]) <= 1e-9 and abs(axis[1]) <= 1e-9 and axis[2] > 0.0
|
|
|
|
|
|
def feature_base_center(
|
|
feature: dict[str, Any],
|
|
center: list[float],
|
|
axis: list[float],
|
|
height: float,
|
|
) -> list[float]:
|
|
"""Resolve the primitive's bottom-face center from an explicit anchor."""
|
|
anchor = str(feature.get("anchor", "base_center"))
|
|
if anchor == "base_center":
|
|
return center
|
|
if anchor == "geometric_center":
|
|
return [
|
|
float(center[index]) - float(axis[index]) * height / 2.0
|
|
for index in range(3)
|
|
]
|
|
raise DesignIRError(
|
|
f"{feature.get('id', 'feature')}.anchor must be base_center or geometric_center"
|
|
)
|
|
|
|
|
|
def build_shape(payload: dict[str, Any]) -> Part:
|
|
payload = 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 = feature_axis(payload, 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 = resolve_center(
|
|
feature.get("center", [0, 0, 0]),
|
|
values,
|
|
f"{feature['id']}.center",
|
|
)
|
|
base_center = feature_base_center(feature, center, axis, height)
|
|
tool = Plane(origin=base_center, z_dir=axis) * 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 = resolve_center(
|
|
feature.get("center", [0, 0, 0]),
|
|
values,
|
|
f"{feature['id']}.center",
|
|
)
|
|
base_center = feature_base_center(feature, center, axis, height)
|
|
tool = Plane(origin=base_center, z_dir=axis) * 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 not _axis_is_z(axis):
|
|
raise DesignIRError(
|
|
f"{feature['id']} hole cutting currently requires positive Z axis"
|
|
)
|
|
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 = resolve_center(
|
|
feature.get("center", [0, 0, 0]),
|
|
values,
|
|
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]:
|
|
payload = 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)
|
|
|
|
def semantic_tag(value: object, *, prefix: str = "feature") -> str:
|
|
token = re.sub(r"[^a-z0-9]+", ".", str(value).lower()).strip(".")
|
|
return f"{prefix}.{token or 'unnamed'}"
|
|
|
|
graph_id = re.sub(
|
|
r"[^a-zA-Z0-9_.-]+",
|
|
"_",
|
|
str(payload.get("model_id") or payload.get("family") or "designir_model"),
|
|
).strip("._-") or "designir_model"
|
|
|
|
@scad.model(graph_id=graph_id)
|
|
def compile_model() -> Any:
|
|
shape: Any = None
|
|
z_min: float | None = None
|
|
z_max: float | None = None
|
|
for feature in payload["features"]:
|
|
operation = feature["operation"]
|
|
axis = feature_axis(payload, feature)
|
|
feature_tag = semantic_tag(feature["id"])
|
|
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 = resolve_center(
|
|
feature.get("center", [0, 0, 0]),
|
|
values,
|
|
f"{feature['id']}.center",
|
|
)
|
|
point = tuple(feature_base_center(feature, center, axis, height))
|
|
tool = scad.make_cylinder_rsolid(
|
|
radius=radius,
|
|
height=height,
|
|
bottom_face_center=point,
|
|
axis=tuple(axis),
|
|
tag_prefix=feature_tag,
|
|
result_tag=f"{feature_tag}.primitive",
|
|
)
|
|
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":
|
|
if not _axis_is_z(axis):
|
|
raise DesignIRError(
|
|
f"{feature['id']} SimpleCAD box currently requires positive Z axis"
|
|
)
|
|
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 = resolve_center(
|
|
feature.get("center", [0, 0, 0]),
|
|
values,
|
|
f"{feature['id']}.center",
|
|
)
|
|
point = tuple(feature_base_center(feature, center, axis, height))
|
|
tool = scad.make_box_rsolid(
|
|
width=width,
|
|
height=depth,
|
|
depth=height,
|
|
bottom_face_center=point,
|
|
tag_prefix=feature_tag,
|
|
result_tag=f"{feature_tag}.primitive",
|
|
)
|
|
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 not _axis_is_z(axis):
|
|
raise DesignIRError(
|
|
f"{feature['id']} SimpleCAD hole currently requires positive Z axis"
|
|
)
|
|
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 = resolve_center(
|
|
feature.get("center", [0, 0, 0]),
|
|
values,
|
|
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),
|
|
tag_prefix=f"{feature_tag}.tool.item{index + 1}",
|
|
result_tag=f"{feature_tag}.tool.item{index + 1}.solid",
|
|
)
|
|
for index, (x, y) in enumerate(centers)
|
|
]
|
|
shape = scad.cut_rsolid(
|
|
shape, *cutters, skip_non_intersecting=False
|
|
)
|
|
shape = scad.apply_tag(shape=shape, tag=f"{feature_tag}.result")
|
|
print(
|
|
f"simplecad grounding {feature['id']}: volume={shape.get_volume():.6f}",
|
|
file=sys.stderr,
|
|
)
|
|
return scad.capture_result(value=shape)
|
|
|
|
result = compile_model()
|
|
shape = result.value
|
|
model_json = result.model_json
|
|
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 materialize_hybrid_designir(
|
|
payload: dict[str, Any], generated_step: Path
|
|
) -> dict[str, Any]:
|
|
"""Attach a compiled SurfaceIR snapshot to a semantic DesignIR 3.0 model.
|
|
|
|
The semantic feature program remains authoritative for future edits. The
|
|
SurfaceIR snapshot is extracted only from the newly generated STEP, never
|
|
from teacher geometry.
|
|
"""
|
|
semantic = (
|
|
migrate_designir_2_to_3(payload)
|
|
if payload.get("schema_version") == LEGACY_SCHEMA_VERSION
|
|
else copy.deepcopy(payload)
|
|
)
|
|
compiler_payload(semantic)
|
|
surface_script = Path(__file__).with_name("surfaceir_pipeline.py")
|
|
spec = importlib.util.spec_from_file_location(
|
|
"_designir_surfaceir_pipeline", surface_script
|
|
)
|
|
if spec is None or spec.loader is None:
|
|
raise DesignIRError("Cannot load SurfaceIR compiler")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
extracted = module.extract_surfaceir(generated_step.expanduser().resolve())
|
|
semantic["surface_layer"] = extracted["surface_layer"]
|
|
semantic["reconstruction_strategy"] = extracted.get(
|
|
"reconstruction_strategy",
|
|
{
|
|
"boundary_strategy": "exact_3d",
|
|
"selection_status": "default",
|
|
},
|
|
)
|
|
semantic["document_status"] = extracted.get(
|
|
"document_status", "geometry_present"
|
|
)
|
|
semantic["reconstruction_mode"] = "hybrid_semantic_surface_parametric"
|
|
edit_interface = semantic["edit_interface"]
|
|
edit_interface["surface_parameter_groups"] = extracted.get(
|
|
"edit_interface", {}
|
|
).get("surface_parameter_groups", [])
|
|
edit_interface["modification_levels"] = list(
|
|
dict.fromkeys(
|
|
list(edit_interface.get("modification_levels", []))
|
|
+ ["semantic_feature", "analytic_surface", "spline_control"]
|
|
)
|
|
)
|
|
semantic["compiled_surface_provenance"] = {
|
|
"source": "generated_step",
|
|
"teacher_geometry_used": False,
|
|
"semantic_layer_authoritative": True,
|
|
}
|
|
return semantic
|
|
|
|
|
|
def anti_cheat(payload: dict[str, Any], generator: Path | None = None) -> dict[str, Any]:
|
|
validate_designir(payload)
|
|
violations: list[str] = []
|
|
inspected = (
|
|
payload.get("semantic_layer", {})
|
|
if payload.get("schema_version") == SCHEMA_VERSION
|
|
else payload
|
|
)
|
|
forbidden = find_forbidden(inspected)
|
|
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 _parameter_reference_count(value: Any, parameter: str) -> int:
|
|
if isinstance(value, dict):
|
|
direct = 1 if value.get("parameter") == parameter else 0
|
|
return direct + sum(
|
|
_parameter_reference_count(child, parameter)
|
|
for child in value.values()
|
|
)
|
|
if isinstance(value, list):
|
|
return sum(_parameter_reference_count(child, parameter) for child in value)
|
|
return 0
|
|
|
|
|
|
def acceptance_report(
|
|
teacher_path: Path,
|
|
rebuilt_path: Path,
|
|
payload: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
payload = validate_designir(payload)
|
|
teacher_resolved = teacher_path.expanduser().resolve()
|
|
rebuilt_resolved = rebuilt_path.expanduser().resolve()
|
|
if teacher_resolved == rebuilt_resolved:
|
|
raise DesignIRError("Teacher and rebuilt STEP must be different files")
|
|
teacher = import_step(teacher_resolved)
|
|
rebuilt = import_step(rebuilt_resolved)
|
|
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)
|
|
modified_shape = import_step(output)
|
|
changed_volume = _symmetric_difference_volume(
|
|
rebuilt, modified_shape
|
|
)
|
|
change_ratio = changed_volume / max(
|
|
rebuilt_facts["volume_mm3"], 1e-12
|
|
)
|
|
reference_count = _parameter_reference_count(
|
|
payload.get("features", []), name
|
|
)
|
|
preserved = {
|
|
preserved_name: (
|
|
preserved_name in payload["parameters"]
|
|
and preserved_name in modified["parameters"]
|
|
and payload["parameters"][preserved_name]["value"]
|
|
== modified["parameters"][preserved_name]["value"]
|
|
)
|
|
for preserved_name in test.get("preserve", [])
|
|
}
|
|
expected_change = str(test.get("expected_change") or "")
|
|
expected_change_observed = change_ratio > 1e-9
|
|
if expected_change == "overall_height_increases":
|
|
expected_change_observed = (
|
|
result["facts"]["size_mm"][2]
|
|
> rebuilt_facts["size_mm"][2] + 1e-9
|
|
)
|
|
success = (
|
|
result["facts"]["solid_count"] > 0
|
|
and reference_count > 0
|
|
and expected_change_observed
|
|
and all(preserved.values())
|
|
)
|
|
perturbations.append(
|
|
{
|
|
"parameter": name,
|
|
"original_value": original,
|
|
"modified_value": modified["parameters"][name]["value"],
|
|
"expected_change": test.get("expected_change"),
|
|
"preserve": test.get("preserve", []),
|
|
"parameter_reference_count": reference_count,
|
|
"geometry_change_ratio": change_ratio,
|
|
"expected_change_observed": expected_change_observed,
|
|
"preserved_parameter_checks": preserved,
|
|
"success": success,
|
|
"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"],
|
|
"teacher_sha256": hashlib.sha256(teacher_resolved.read_bytes()).hexdigest(),
|
|
"applied_experience_ids": sorted(
|
|
{
|
|
str(item)
|
|
for item in payload.get("applied_experience_ids", [])
|
|
if str(item).strip()
|
|
}
|
|
),
|
|
"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]
|
|
passing = [item for item in loaded if item.get("next_state") == "replay_validated"]
|
|
minimum = int(policy["minimum_independent_cases"])
|
|
methods: dict[str, set[str]] = {}
|
|
for item in passing:
|
|
source_hash = str(item.get("teacher_sha256") or "")
|
|
if not source_hash:
|
|
continue
|
|
for experience_id in item.get("applied_experience_ids", []):
|
|
methods.setdefault(str(experience_id), set()).add(source_hash)
|
|
validated_methods = sorted(
|
|
experience_id
|
|
for experience_id, source_hashes in methods.items()
|
|
if len(source_hashes) >= minimum
|
|
)
|
|
independent_sources = len(
|
|
{
|
|
source_hash
|
|
for source_hashes in methods.values()
|
|
for source_hash in source_hashes
|
|
}
|
|
)
|
|
eligible = bool(validated_methods)
|
|
return {
|
|
"eligible": eligible,
|
|
"independent_case_count": independent_sources,
|
|
"passing_case_count": len(passing),
|
|
"required_case_count": minimum,
|
|
"validated_experience_ids": validated_methods,
|
|
"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)
|
|
|
|
migrate = commands.add_parser("migrate")
|
|
migrate.add_argument("designir", type=Path)
|
|
migrate.add_argument("--output", type=Path, required=True)
|
|
|
|
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"))
|
|
|
|
materialize = commands.add_parser("materialize-hybrid")
|
|
materialize.add_argument("designir", type=Path)
|
|
materialize.add_argument("--step", type=Path, required=True)
|
|
materialize.add_argument("--output", type=Path, required=True)
|
|
|
|
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":
|
|
original = read_json(args.designir)
|
|
payload = validate_designir(original)
|
|
result = {
|
|
"valid": True,
|
|
"model_id": payload["model_id"],
|
|
"input_schema_version": original.get("schema_version"),
|
|
"supported_operations": sorted(SUPPORTED_OPERATIONS),
|
|
}
|
|
elif args.command == "migrate":
|
|
original = read_json(args.designir)
|
|
migrated = (
|
|
migrate_designir_2_to_3(original)
|
|
if original.get("schema_version") == LEGACY_SCHEMA_VERSION
|
|
else original
|
|
)
|
|
compiler_payload(migrated)
|
|
write_json(args.output, migrated)
|
|
result = {
|
|
"valid": True,
|
|
"output": str(args.output.expanduser().resolve()),
|
|
"schema_version": migrated["schema_version"],
|
|
"reconstruction_mode": migrated["reconstruction_mode"],
|
|
}
|
|
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 == "materialize-hybrid":
|
|
materialized = materialize_hybrid_designir(
|
|
read_json(args.designir), args.step
|
|
)
|
|
write_json(args.output, materialized)
|
|
result = {
|
|
"valid": True,
|
|
"output": str(args.output.expanduser().resolve()),
|
|
"schema_version": materialized["schema_version"],
|
|
"reconstruction_mode": materialized["reconstruction_mode"],
|
|
"semantic_layer_authoritative": True,
|
|
}
|
|
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())
|