Files
cadSet/text-to-cad/skills/cad-router/scripts/model_spec.py
T

633 lines
23 KiB
Python

#!/usr/bin/env python3
"""Create and execute per-part JSON model specifications.
Two reconstruction modes are supported:
* ``native_generator`` loads a task-local Python generator. The generator reads
the same model-spec.json, so named parameter edits remain the source of truth.
* ``exact_step_base`` imports an immutable task-local STEP file and applies a
parametric modification layer. With no modifications, the imported B-Rep is
preserved geometrically even though the exported STEP text may differ.
"""
from __future__ import annotations
import argparse
import hashlib
import importlib.util
import json
import math
import re
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any
from build123d import (
Align,
Box,
Cylinder,
Part,
Pos,
export_step,
export_stl,
import_step,
)
SCHEMA_VERSION = "1.0"
MODEL_SPEC_KIND = "parametric_cad_model"
MODEL_ID_RE = re.compile(r"[^a-zA-Z0-9._-]+")
SUPPORTED_OPERATIONS = {
"add_box",
"cut_box",
"add_cylinder",
"cut_cylinder",
}
class ModelSpecError(ValueError):
"""Raised when a model specification is invalid or unsafe to execute."""
def _read_json(path: Path) -> dict[str, Any]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ModelSpecError(f"Cannot read model spec {path}: {exc}") from exc
if not isinstance(value, dict):
raise ModelSpecError("Model spec must be a JSON object")
return value
def _write_json(path: Path, value: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(
json.dumps(value, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
temporary.replace(path)
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _safe_model_id(value: str) -> str:
result = MODEL_ID_RE.sub("_", value.strip()).strip("._-")
if not result:
raise ModelSpecError("model_id must contain a letter or number")
return result
def _resolve_relative(spec_path: Path, value: str, field: str) -> Path:
candidate = Path(value)
if candidate.is_absolute():
raise ModelSpecError(f"{field} must be task-relative, not absolute")
return (spec_path.parent / candidate).resolve()
def _parameter_value(spec: dict[str, Any], name: str) -> float:
parameters = spec.get("parameters", {})
entry = parameters.get(name) if isinstance(parameters, dict) else None
if not isinstance(entry, dict) or "value" not in entry:
raise ModelSpecError(f"Unknown parameter reference: {name}")
value = entry["value"]
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ModelSpecError(f"Parameter {name} must contain a numeric value")
if not math.isfinite(float(value)):
raise ModelSpecError(f"Parameter {name} must be finite")
return float(value)
def resolve_number(spec: dict[str, Any], value: Any, field: str) -> float:
if isinstance(value, bool):
raise ModelSpecError(f"{field} must be numeric")
if isinstance(value, (int, float)):
result = float(value)
elif isinstance(value, dict) and "parameter" in value:
unexpected = set(value) - {"parameter", "scale", "offset"}
if unexpected:
raise ModelSpecError(
f"{field} parameter expression has unsupported keys: "
+ ", ".join(sorted(unexpected))
)
scale = value.get("scale", 1.0)
offset = value.get("offset", 0.0)
if (
isinstance(scale, bool)
or not isinstance(scale, (int, float))
or isinstance(offset, bool)
or not isinstance(offset, (int, float))
):
raise ModelSpecError(f"{field} scale and offset must be numeric")
result = (
_parameter_value(spec, str(value["parameter"])) * float(scale)
+ float(offset)
)
else:
raise ModelSpecError(
f"{field} must be a number or a parameter expression"
)
if not math.isfinite(result):
raise ModelSpecError(f"{field} must be finite")
return result
def _vector3(spec: dict[str, Any], value: Any, field: str) -> tuple[float, float, float]:
if not isinstance(value, list) or len(value) != 3:
raise ModelSpecError(f"{field} must contain three values")
return tuple(
resolve_number(spec, item, f"{field}[{index}]")
for index, item in enumerate(value)
)
def validate_model_spec(spec: dict[str, Any]) -> None:
if spec.get("schema_version") != SCHEMA_VERSION:
raise ModelSpecError(f"Unsupported model spec schema: {spec.get('schema_version')}")
if spec.get("model_spec_kind") != MODEL_SPEC_KIND:
raise ModelSpecError("Not a parametric CAD model specification")
_safe_model_id(str(spec.get("model_id", "")))
if spec.get("units") != "mm":
raise ModelSpecError("V1 model specs use millimetres")
reconstruction = spec.get("reconstruction")
if not isinstance(reconstruction, dict):
raise ModelSpecError("reconstruction must be an object")
mode = reconstruction.get("mode")
if mode not in {"native_generator", "exact_step_base"}:
raise ModelSpecError(f"Unsupported reconstruction mode: {mode}")
if mode == "native_generator":
generator = reconstruction.get("generator")
if not isinstance(generator, dict) or not generator.get("path"):
raise ModelSpecError("native_generator requires generator.path")
else:
source = reconstruction.get("source")
if (
not isinstance(source, dict)
or not source.get("path")
or not source.get("sha256")
):
raise ModelSpecError("exact_step_base requires source.path and source.sha256")
parameters = spec.get("parameters", {})
if not isinstance(parameters, dict):
raise ModelSpecError("parameters must be an object")
for name, entry in parameters.items():
if not isinstance(entry, dict) or "value" not in entry:
raise ModelSpecError(f"Parameter {name} must be an object with value")
_parameter_value(spec, name)
modifications = spec.get("modifications", [])
if not isinstance(modifications, list):
raise ModelSpecError("modifications must be an array")
seen_ids: set[str] = set()
for index, modification in enumerate(modifications):
if not isinstance(modification, dict):
raise ModelSpecError(f"modifications[{index}] must be an object")
feature_id = str(modification.get("id", "")).strip()
if not feature_id or feature_id in seen_ids:
raise ModelSpecError("Every modification requires a unique id")
seen_ids.add(feature_id)
operation = modification.get("operation")
if operation not in SUPPORTED_OPERATIONS:
raise ModelSpecError(f"Unsupported modification operation: {operation}")
outputs = spec.get("outputs")
if not isinstance(outputs, dict) or not outputs.get("step"):
raise ModelSpecError("outputs.step is required")
def geometry_facts(shape: Any) -> dict[str, Any]:
box = shape.bounding_box()
solids = list(shape.solids())
volume = sum(float(solid.volume) for solid in solids)
return {
"size_mm": [
round(float(box.size.X), 9),
round(float(box.size.Y), 9),
round(float(box.size.Z), 9),
],
"center_mm": [
round(float(box.center().X), 9),
round(float(box.center().Y), 9),
round(float(box.center().Z), 9),
],
"solid_count": len(solids),
"face_count": len(shape.faces()),
"edge_count": len(shape.edges()),
"volume_mm3": round(volume, 6),
}
def _rotation_for_axis(axis: str) -> tuple[float, float, float]:
normalized = axis.lower()
if normalized == "z":
return (0.0, 0.0, 0.0)
if normalized == "x":
return (0.0, 90.0, 0.0)
if normalized == "y":
return (-90.0, 0.0, 0.0)
raise ModelSpecError(f"Unsupported cylinder axis: {axis}")
def _make_tool(spec: dict[str, Any], modification: dict[str, Any]) -> Any:
operation = str(modification["operation"])
center = _vector3(spec, modification.get("center", [0, 0, 0]), "center")
if operation.endswith("_cylinder"):
diameter = resolve_number(spec, modification.get("diameter"), "diameter")
length = resolve_number(spec, modification.get("length"), "length")
if diameter <= 0 or length <= 0:
raise ModelSpecError("Cylinder diameter and length must be positive")
tool = Cylinder(
diameter / 2.0,
length,
align=(Align.CENTER, Align.CENTER, Align.CENTER),
rotation=_rotation_for_axis(str(modification.get("axis", "z"))),
)
else:
size = _vector3(spec, modification.get("size"), "size")
if any(item <= 0 for item in size):
raise ModelSpecError("Box dimensions must be positive")
tool = Box(
*size,
align=(Align.CENTER, Align.CENTER, Align.CENTER),
)
return Pos(*center) * tool
def _apply_modifications(shape: Any, spec: dict[str, Any]) -> Any:
result = shape
for modification in spec.get("modifications", []):
if modification.get("enabled", True) is False:
continue
tool = _make_tool(spec, modification)
operation = modification["operation"]
result = result + tool if operation.startswith("add_") else result - tool
return result
def _load_generator(spec_path: Path, reconstruction: dict[str, Any]) -> Any:
generator = reconstruction["generator"]
source_path = _resolve_relative(spec_path, str(generator["path"]), "generator.path")
if not source_path.is_file():
raise ModelSpecError(f"Generator does not exist: {source_path}")
module_spec = importlib.util.spec_from_file_location(
f"cad_model_{hash(source_path)}",
source_path,
)
if module_spec is None or module_spec.loader is None:
raise ModelSpecError(f"Cannot load generator: {source_path}")
module = importlib.util.module_from_spec(module_spec)
module_spec.loader.exec_module(module)
entrypoint = str(generator.get("entrypoint", "gen_step"))
function = getattr(module, entrypoint, None)
if not callable(function):
raise ModelSpecError(f"Generator has no callable {entrypoint}: {source_path}")
return function()
def build_shape(spec_path: Path, spec: dict[str, Any] | None = None) -> Any:
resolved_spec = spec_path.expanduser().resolve()
payload = spec or _read_json(resolved_spec)
validate_model_spec(payload)
reconstruction = payload["reconstruction"]
if reconstruction["mode"] == "native_generator":
shape = _load_generator(resolved_spec, reconstruction)
else:
source = reconstruction["source"]
source_path = _resolve_relative(resolved_spec, str(source["path"]), "source.path")
if not source_path.is_file():
raise ModelSpecError(f"Exact STEP base does not exist: {source_path}")
actual_hash = _sha256(source_path)
if actual_hash != source["sha256"]:
raise ModelSpecError(
"Exact STEP base checksum changed; import it again instead of "
"silently rebuilding from a different source"
)
imported = import_step(source_path)
# build123d's STEP importer returns a topology wrapper that is readable
# and boolean-capable but is not always directly accepted by its STEP
# exporter. Normalize it to a Part while preserving the wrapped B-Rep.
shape = Part(imported.wrapped)
return _apply_modifications(shape, payload)
def build_model(spec_path: Path) -> dict[str, Any]:
resolved_spec = spec_path.expanduser().resolve()
payload = _read_json(resolved_spec)
validate_model_spec(payload)
outputs = payload["outputs"]
step_path = _resolve_relative(resolved_spec, str(outputs["step"]), "outputs.step")
step_path.parent.mkdir(parents=True, exist_ok=True)
reconstruction = payload["reconstruction"]
stl_value = outputs.get("stl")
stl_path = (
_resolve_relative(resolved_spec, str(stl_value), "outputs.stl")
if stl_value
else None
)
if reconstruction["mode"] == "native_generator":
generator_path = _resolve_relative(
resolved_spec,
str(reconstruction["generator"]["path"]),
"generator.path",
)
if generator_path.with_suffix(".step").resolve() != step_path:
raise ModelSpecError(
"V1 native generator output must use the generator basename"
)
cad_step = Path(__file__).resolve().parents[2] / "cad" / "scripts" / "step"
command = [sys.executable, str(cad_step), str(generator_path), "--force"]
if stl_path is not None:
if stl_path.parent != generator_path.parent:
raise ModelSpecError(
"V1 native generator STL must remain in the task directory"
)
command.extend(["--stl", stl_path.name])
completed = subprocess.run(
command,
cwd=resolved_spec.parent,
text=True,
capture_output=True,
)
if completed.returncode != 0:
raise ModelSpecError(
"CAD generator failed: "
+ (completed.stderr.strip() or completed.stdout.strip())
)
if not step_path.is_file():
raise ModelSpecError(f"CAD generator did not write {step_path}")
else:
shape = build_shape(resolved_spec, payload)
source_path = _resolve_relative(
resolved_spec,
str(reconstruction["source"]["path"]),
"source.path",
)
if not payload.get("modifications"):
# Empty exact-base rebuilds preserve the STEP byte-for-byte.
if source_path != step_path:
shutil.copy2(source_path, step_path)
else:
export_step(shape, step_path)
if stl_path is not None:
stl_path.parent.mkdir(parents=True, exist_ok=True)
export_stl(shape, stl_path)
exported_shape = Part(import_step(step_path).wrapped)
result: dict[str, Any] = {
"model_spec": str(resolved_spec),
"step": str(step_path),
"facts": geometry_facts(exported_shape),
}
if stl_path is not None:
result["stl"] = str(stl_path)
return result
def import_step_model(
source_step: Path,
task_dir: Path,
model_id: str | None = None,
) -> dict[str, Any]:
source = source_step.expanduser().resolve()
if not source.is_file() or source.suffix.lower() not in {".step", ".stp"}:
raise ModelSpecError(f"Input must be an existing STEP/STP file: {source}")
target_dir = task_dir.expanduser().resolve()
target_dir.mkdir(parents=True, exist_ok=True)
normalized_id = _safe_model_id(model_id or source.stem)
copied_source = target_dir / "source.step"
if copied_source.resolve() != source:
shutil.copy2(source, copied_source)
source_shape = import_step(copied_source)
source_facts = geometry_facts(source_shape)
spec_path = target_dir / "model-spec.json"
payload: dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"model_spec_kind": MODEL_SPEC_KIND,
"model_id": normalized_id,
"units": "mm",
"reconstruction": {
"mode": "exact_step_base",
"source": {
"path": "source.step",
"sha256": _sha256(copied_source),
"geometry_signature": source_facts,
},
"history_recovery": {
"status": "not_present_in_step",
"contract": (
"Preserve the imported B-Rep exactly as the immutable base; "
"represent later edits as named parametric modifications."
),
},
},
"parameters": {},
"modifications": [],
"outputs": {
"step": f"{normalized_id}.step",
"stl": f"{normalized_id}.stl",
},
"validation": {
"exact_base_required": True,
"geometry_signature_tolerance_mm": 1e-7,
},
}
_write_json(spec_path, payload)
result = build_model(spec_path)
result["source"] = str(copied_source)
result["source_sha256"] = payload["reconstruction"]["source"]["sha256"]
return result
def _numbers_close(first: Any, second: Any, tolerance: float) -> bool:
if isinstance(first, list) and isinstance(second, list):
return len(first) == len(second) and all(
_numbers_close(left, right, tolerance)
for left, right in zip(first, second)
)
if isinstance(first, (int, float)) and isinstance(second, (int, float)):
return math.isclose(
float(first),
float(second),
rel_tol=tolerance,
abs_tol=tolerance,
)
return first == second
def verify_model(spec_path: Path) -> dict[str, Any]:
resolved_spec = spec_path.expanduser().resolve()
payload = _read_json(resolved_spec)
shape = build_shape(resolved_spec, payload)
facts = geometry_facts(shape)
reconstruction = payload["reconstruction"]
result: dict[str, Any] = {
"model_spec": str(resolved_spec),
"valid": True,
"facts": facts,
"checks": [],
}
if reconstruction["mode"] == "exact_step_base" and not payload.get("modifications"):
expected = reconstruction["source"].get("geometry_signature", {})
source_path = _resolve_relative(
resolved_spec,
str(reconstruction["source"]["path"]),
"source.path",
)
output_path = _resolve_relative(
resolved_spec,
str(payload["outputs"]["step"]),
"outputs.step",
)
output_shape = (
Part(import_step(output_path).wrapped)
if output_path.is_file()
else None
)
output_facts = geometry_facts(output_shape) if output_shape is not None else {}
tolerance = float(
payload.get("validation", {}).get(
"geometry_signature_tolerance_mm",
1e-7,
)
)
signature_exact = all(
_numbers_close(expected.get(key), facts.get(key), tolerance)
for key in (
"size_mm",
"center_mm",
"solid_count",
"face_count",
"edge_count",
"volume_mm3",
)
)
exported_exact = (
output_path.is_file()
and _sha256(output_path) == _sha256(source_path)
and all(
_numbers_close(expected.get(key), output_facts.get(key), tolerance)
for key in (
"size_mm",
"center_mm",
"solid_count",
"face_count",
"edge_count",
"volume_mm3",
)
)
)
exact = signature_exact and exported_exact
result["checks"].append(
{
"check": "exact_base_geometry_signature",
"passed": exact,
"expected": expected,
"actual": facts,
"exported": output_facts,
"byte_identical_step": (
output_path.is_file()
and _sha256(output_path) == _sha256(source_path)
),
}
)
result["valid"] = exact
else:
result["checks"].append(
{
"check": "model_spec_execution",
"passed": True,
"detail": "The parameterized model spec executed successfully.",
}
)
return result
def set_parameter(spec_path: Path, name: str, value: float) -> dict[str, Any]:
resolved_spec = spec_path.expanduser().resolve()
payload = _read_json(resolved_spec)
validate_model_spec(payload)
parameters = payload["parameters"]
if name not in parameters:
raise ModelSpecError(
f"Unknown parameter {name}; add a named parameter and feature binding first"
)
parameters[name]["value"] = value
validate_model_spec(payload)
_write_json(resolved_spec, payload)
return {
"model_spec": str(resolved_spec),
"parameter": name,
"value": value,
}
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Create, build, modify, and verify per-part model-spec.json files."
)
subparsers = parser.add_subparsers(dest="command", required=True)
import_parser = subparsers.add_parser(
"import-step",
help="Create an exact STEP-base model spec in a task-owned directory.",
)
import_parser.add_argument("source_step", type=Path)
import_parser.add_argument("--task-dir", type=Path, required=True)
import_parser.add_argument("--model-id")
build_command = subparsers.add_parser(
"build",
help="Regenerate STEP/STL from a model specification.",
)
build_command.add_argument("model_spec", type=Path)
verify_command = subparsers.add_parser(
"verify",
help="Validate a model specification and exact-base signature.",
)
verify_command.add_argument("model_spec", type=Path)
set_command = subparsers.add_parser(
"set",
help="Change one existing named parameter in a model specification.",
)
set_command.add_argument("model_spec", type=Path)
set_command.add_argument("name")
set_command.add_argument("value", type=float)
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
if args.command == "import-step":
result = import_step_model(args.source_step, args.task_dir, args.model_id)
elif args.command == "build":
result = build_model(args.model_spec)
elif args.command == "verify":
result = verify_model(args.model_spec)
else:
result = set_parameter(args.model_spec, args.name, args.value)
except ModelSpecError as exc:
print(json.dumps({"error": str(exc)}, ensure_ascii=False, indent=2))
return 2
print(json.dumps(result, ensure_ascii=False, indent=2))
if args.command == "verify" and result.get("valid") is False:
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())