feat: unify CAD workflows on DesignIR 3.0

This commit is contained in:
Jerry
2026-07-28 14:41:26 +08:00
parent 6e33bc0e11
commit b5738e9109
36 changed files with 9690 additions and 1305 deletions
+461 -48
View File
@@ -1,11 +1,17 @@
#!/usr/bin/env python3
"""Compile and validate DesignIR 2.0 without depending on teacher geometry."""
"""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
@@ -15,11 +21,23 @@ import tempfile
from pathlib import Path
from typing import Any
from build123d import Align, Box, CenterOf, Cylinder, Part, Pos, export_step, import_step
from build123d import (
Align,
Box,
CenterOf,
Cylinder,
Part,
Plane,
Pos,
export_step,
import_step,
)
SCHEMA_VERSION = "2.0"
DESIGNIR_KIND = "executable_parametric_design"
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",
@@ -138,14 +156,14 @@ def _vector3(value: Any, label: str) -> None:
raise DesignIRError(f"{label} must be a three-number vector")
def validate_designir(payload: dict[str, Any]) -> dict[str, Any]:
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"] != SCHEMA_VERSION:
if payload["schema_version"] != LEGACY_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["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 {
@@ -196,14 +214,150 @@ def validate_designir(payload: dict[str, Any]) -> dict[str, Any]:
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")
@@ -261,13 +415,69 @@ def resolve_value(value: Any, values: dict[str, float], label: str) -> float:
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 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:
validate_designir(payload)
payload = validate_designir(payload)
if payload["reconstruction_status"] != "ready":
missing = ", ".join(payload.get("missing_capabilities", [])) or "unspecified"
raise DesignIRError(
@@ -278,32 +488,39 @@ def build_shape(payload: dict[str, Any]) -> Part:
for feature in payload["features"]:
operation = feature["operation"]
_axis_is_z(feature)
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 = 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),
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 = 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),
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(
@@ -314,8 +531,11 @@ def build_shape(payload: dict[str, Any]) -> Part:
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")
center = resolve_center(
feature.get("center", [0, 0, 0]),
values,
f"{feature['id']}.center",
)
centers = [(float(center[0]), float(center[1]))]
else:
count = int(
@@ -352,7 +572,7 @@ def build_shape(payload: dict[str, Any]) -> Part:
def build_simplecad_shape(payload: dict[str, Any]) -> tuple[Any, str]:
validate_designir(payload)
payload = validate_designir(payload)
if payload["reconstruction_status"] != "ready":
missing = ", ".join(payload.get("missing_capabilities", [])) or "unspecified"
raise DesignIRError(
@@ -373,7 +593,7 @@ def build_simplecad_shape(payload: dict[str, Any]) -> tuple[Any, str]:
with scad.GraphSession() as session:
for feature in payload["features"]:
operation = feature["operation"]
_axis_is_z(feature)
axis = feature_axis(payload, feature)
if operation in {"extrude_circle", "add_cylinder"}:
radius = resolve_value(
feature["radius"], values, f"{feature['id']}.radius"
@@ -381,14 +601,17 @@ def build_simplecad_shape(payload: dict[str, Any]) -> tuple[Any, str]:
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))
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=(0.0, 0.0, 1.0),
axis=tuple(axis),
)
shape = (
tool
@@ -402,6 +625,10 @@ def build_simplecad_shape(payload: dict[str, Any]) -> tuple[Any, str]:
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"
)
@@ -411,9 +638,12 @@ def build_simplecad_shape(payload: dict[str, Any]) -> tuple[Any, str]:
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))
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,
@@ -432,14 +662,21 @@ def build_simplecad_shape(payload: dict[str, Any]) -> tuple[Any, str]:
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 = feature.get("center", [0, 0, 0])
_vector3(center, f"{feature['id']}.center")
center = resolve_center(
feature.get("center", [0, 0, 0]),
values,
f"{feature['id']}.center",
)
centers = [(float(center[0]), float(center[1]))]
else:
count = int(
@@ -532,10 +769,69 @@ def compile_designir(
}
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] = []
forbidden = find_forbidden(payload)
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:
@@ -594,14 +890,30 @@ def _symmetric_difference_volume(first: Part, second: Part) -> float:
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]:
validate_designir(payload)
teacher = import_step(teacher_path.expanduser().resolve())
rebuilt = import_step(rebuilt_path.expanduser().resolve())
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)
@@ -640,6 +952,38 @@ def acceptance_report(
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,
@@ -647,7 +991,11 @@ def acceptance_report(
"modified_value": modified["parameters"][name]["value"],
"expected_change": test.get("expected_change"),
"preserve": test.get("preserve", []),
"success": result["facts"]["solid_count"] > 0,
"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"],
}
)
@@ -678,6 +1026,14 @@ def acceptance_report(
"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"]),
@@ -700,15 +1056,34 @@ def acceptance_report(
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
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": len(model_ids),
"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",
}
@@ -720,6 +1095,10 @@ def build_parser() -> argparse.ArgumentParser:
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)
@@ -736,6 +1115,11 @@ def build_parser() -> argparse.ArgumentParser:
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)
@@ -752,12 +1136,29 @@ 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))
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":
@@ -768,6 +1169,18 @@ def main(argv: list[str] | None = None) -> int:
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)