354 lines
12 KiB
Python
354 lines
12 KiB
Python
"""Fixed FreeCAD-side model compiler. FreeCADCmd may import this file directly."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from collections import Counter
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import FreeCAD as App
|
|
import Import
|
|
import Part
|
|
|
|
MAX_SOLIDS = 256
|
|
MAX_TRIANGLES = 500_000
|
|
JOB_ENV = "ZCBOT_FREECAD_JOB_DIR"
|
|
|
|
|
|
class EngineError(RuntimeError):
|
|
def __init__(self, code: str, detail: str = "") -> None:
|
|
super().__init__(detail or code)
|
|
self.code = code
|
|
|
|
|
|
def _atomic_json(path: Path, value: object) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_name(path.name + ".tmp-" + os.urandom(8).hex())
|
|
try:
|
|
with temporary.open("w", encoding="utf-8", newline="\n") as handle:
|
|
json.dump(
|
|
value,
|
|
handle,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _checked(shape: object, role: str):
|
|
if shape is None or shape.isNull() or not shape.isValid():
|
|
raise EngineError("FREECAD_SHAPE_INVALID", role)
|
|
solids = list(shape.Solids)
|
|
if not solids:
|
|
raise EngineError("FREECAD_SHAPE_HAS_NO_SOLIDS", role)
|
|
if len(solids) > MAX_SOLIDS:
|
|
raise EngineError("FREECAD_ENTITY_LIMIT_EXCEEDED", role)
|
|
return shape
|
|
|
|
|
|
def _step_shape(path: Path):
|
|
document = App.newDocument("ZcbotStepImport")
|
|
try:
|
|
Import.insert(str(path), document.Name)
|
|
if document.recompute() is False:
|
|
raise EngineError("FREECAD_STEP_RECOMPUTE_FAILED")
|
|
shapes = [
|
|
obj.Shape.copy()
|
|
for obj in document.Objects
|
|
if hasattr(obj, "Shape") and not obj.Shape.isNull()
|
|
]
|
|
if not shapes:
|
|
raise EngineError("FREECAD_STEP_IMPORT_EMPTY")
|
|
shape = shapes[0] if len(shapes) == 1 else Part.makeCompound(shapes)
|
|
except Exception as exc:
|
|
if isinstance(exc, EngineError):
|
|
raise
|
|
raise EngineError("FREECAD_STEP_IMPORT_FAILED", str(exc)) from exc
|
|
finally:
|
|
App.closeDocument(document.Name)
|
|
return _checked(shape, "STEP input")
|
|
|
|
|
|
def _make(step: dict[str, object], values: dict[str, object], input_path: Path | None):
|
|
op = step["op"]
|
|
if op == "box":
|
|
shape = Part.makeBox(step["length"], step["width"], step["height"])
|
|
elif op == "cylinder":
|
|
shape = Part.makeCylinder(
|
|
step["radius"],
|
|
step["height"],
|
|
App.Vector(0, 0, 1),
|
|
step.get("angle_degrees", 360.0),
|
|
)
|
|
elif op == "cone":
|
|
shape = Part.makeCone(
|
|
step["radius1"],
|
|
step["radius2"],
|
|
step["height"],
|
|
App.Vector(0, 0, 1),
|
|
step.get("angle_degrees", 360.0),
|
|
)
|
|
elif op == "sphere":
|
|
shape = Part.makeSphere(
|
|
step["radius"],
|
|
App.Vector(),
|
|
App.Vector(0, 0, 1),
|
|
step.get("angle1_degrees", -90.0),
|
|
step.get("angle2_degrees", 90.0),
|
|
step.get("angle3_degrees", 360.0),
|
|
)
|
|
elif op == "torus":
|
|
shape = Part.makeTorus(
|
|
step["major_radius"],
|
|
step["minor_radius"],
|
|
App.Vector(),
|
|
App.Vector(0, 0, 1),
|
|
step.get("angle1_degrees", -180.0),
|
|
step.get("angle2_degrees", 180.0),
|
|
step.get("angle3_degrees", 360.0),
|
|
)
|
|
elif op == "import_step":
|
|
if input_path is None:
|
|
raise EngineError("FREECAD_STEP_INPUT_NOT_BOUND")
|
|
shape = _step_shape(input_path)
|
|
elif op in {"union", "cut", "intersect"}:
|
|
shape = values[step["target"]]
|
|
for name in step["tools"]:
|
|
tool = values[name]
|
|
if op == "union":
|
|
shape = shape.fuse(tool)
|
|
elif op == "cut":
|
|
shape = shape.cut(tool)
|
|
else:
|
|
shape = shape.common(tool)
|
|
elif op == "translate":
|
|
shape = values[step["target"]].copy()
|
|
shape.translate(App.Vector(*step["vector"]))
|
|
elif op == "rotate":
|
|
shape = values[step["target"]].copy()
|
|
shape.rotate(
|
|
App.Vector(*step["axis_point"]),
|
|
App.Vector(*step["axis_direction"]),
|
|
step["angle_degrees"],
|
|
)
|
|
elif op == "combine":
|
|
shape = Part.makeCompound([values[name] for name in step["targets"]])
|
|
else:
|
|
raise EngineError("FREECAD_OPERATION_UNSUPPORTED", str(op))
|
|
return _checked(shape, str(step["id"]))
|
|
|
|
|
|
def _metrics(shape: object) -> dict[str, object]:
|
|
box = shape.BoundBox
|
|
return {
|
|
"solids": len(shape.Solids),
|
|
"bounding_box": {
|
|
"min": [box.XMin, box.YMin, box.ZMin],
|
|
"max": [box.XMax, box.YMax, box.ZMax],
|
|
"size": [box.XLength, box.YLength, box.ZLength],
|
|
},
|
|
"volume": float(shape.Volume),
|
|
"area": float(shape.Area),
|
|
"center_of_mass": [
|
|
shape.CenterOfMass.x,
|
|
shape.CenterOfMass.y,
|
|
shape.CenterOfMass.z,
|
|
],
|
|
}
|
|
|
|
|
|
def _compare(expected: dict[str, object], actual: dict[str, object], role: str) -> None:
|
|
if expected["solids"] != actual["solids"]:
|
|
raise EngineError(f"FREECAD_{role}_SOLID_COUNT_MISMATCH")
|
|
expected_box = expected["bounding_box"]
|
|
actual_box = actual["bounding_box"]
|
|
scale = max([1.0, *expected_box["size"]])
|
|
box_tolerance = scale * 1e-7
|
|
for key in ("min", "max", "size"):
|
|
if any(
|
|
abs(a - b) > box_tolerance
|
|
for a, b in zip(expected_box[key], actual_box[key])
|
|
):
|
|
raise EngineError(f"FREECAD_{role}_BOUNDING_BOX_MISMATCH")
|
|
volume_tolerance = max(1e-8, abs(expected["volume"]) * 1e-7)
|
|
if abs(expected["volume"] - actual["volume"]) > volume_tolerance:
|
|
raise EngineError(f"FREECAD_{role}_VOLUME_MISMATCH")
|
|
|
|
|
|
def _mesh(shape: object) -> dict[str, object]:
|
|
box = shape.BoundBox
|
|
deflection = max(box.DiagonalLength * 0.002, 0.01)
|
|
vertices, triangles = shape.tessellate(deflection)
|
|
if not vertices or not triangles or len(triangles) > MAX_TRIANGLES:
|
|
raise EngineError("FREECAD_PREVIEW_LIMIT_EXCEEDED")
|
|
return {
|
|
"vertices": [
|
|
[float(vertex.x), float(vertex.y), float(vertex.z)] for vertex in vertices
|
|
],
|
|
"triangles": [[int(index) for index in triangle] for triangle in triangles],
|
|
}
|
|
|
|
|
|
def _surface_types(shape: object) -> dict[str, int]:
|
|
names = []
|
|
for face in shape.Faces:
|
|
names.append(type(face.Surface).__name__.removeprefix("Surface").casefold())
|
|
return dict(sorted(Counter(names).items()))
|
|
|
|
|
|
def run(job_dir: Path) -> None:
|
|
output = (job_dir / "output").resolve(strict=True)
|
|
metadata = (output / ".meta").resolve(strict=True)
|
|
if not output.is_relative_to(job_dir) or not metadata.is_relative_to(output):
|
|
raise EngineError("FREECAD_OUTPUT_PATH_ESCAPES_JOB")
|
|
payload = json.loads((metadata / "resolved-job.json").read_text(encoding="utf-8"))
|
|
model = payload["model"]
|
|
input_path = (
|
|
Path(payload["input_path"]).resolve(strict=True)
|
|
if payload.get("input_path")
|
|
else None
|
|
)
|
|
if input_path is not None and not input_path.is_relative_to(
|
|
(job_dir / "input").resolve(strict=True)
|
|
):
|
|
raise EngineError("FREECAD_INPUT_PATH_ESCAPES_JOB")
|
|
document = App.newDocument("ZcbotModel")
|
|
values: dict[str, object] = {}
|
|
objects: dict[str, object] = {}
|
|
try:
|
|
for step in model["steps"]:
|
|
shape = _make(step, values, input_path)
|
|
values[step["id"]] = shape
|
|
obj = document.addObject("Part::Feature", step["id"])
|
|
obj.Label = step["id"]
|
|
obj.Shape = shape
|
|
objects[step["id"]] = obj
|
|
result_shape = _checked(values[model["result"]], "result")
|
|
result_object_name = objects[model["result"]].Name
|
|
if document.recompute() is False:
|
|
raise EngineError("FREECAD_DOCUMENT_RECOMPUTE_FAILED")
|
|
for obj in document.Objects:
|
|
if (
|
|
hasattr(obj, "Shape")
|
|
and not obj.Shape.isNull()
|
|
and not obj.Shape.isValid()
|
|
):
|
|
raise EngineError("FREECAD_DOCUMENT_SHAPE_INVALID", obj.Name)
|
|
project_path = output / "model.FCStd"
|
|
document.saveAs(str(project_path))
|
|
baseline = _metrics(result_shape)
|
|
finally:
|
|
App.closeDocument(document.Name)
|
|
|
|
reopened = App.openDocument(str(project_path))
|
|
try:
|
|
if reopened.recompute() is False:
|
|
raise EngineError("FREECAD_FCSTD_RECOMPUTE_FAILED")
|
|
result_object = reopened.getObject(result_object_name)
|
|
if result_object is None:
|
|
raise EngineError("FREECAD_FCSTD_RESULT_MISSING")
|
|
fcstd_shape = _checked(result_object.Shape, "FCStd roundtrip")
|
|
_compare(baseline, _metrics(fcstd_shape), "FCSTD_ROUNDTRIP")
|
|
step_path = output / "model.step"
|
|
fcstd_shape.exportStep(str(step_path))
|
|
if not step_path.is_file() or step_path.stat().st_size < 256:
|
|
raise EngineError("FREECAD_STEP_EXPORT_EMPTY")
|
|
finally:
|
|
App.closeDocument(reopened.Name)
|
|
|
|
step_shape = _step_shape(step_path)
|
|
step_metrics = _metrics(step_shape)
|
|
_compare(baseline, step_metrics, "STEP_ROUNDTRIP")
|
|
_atomic_json(
|
|
output / "model-recipe.json",
|
|
{
|
|
"schema_version": 1,
|
|
"recipe": payload["recipe"],
|
|
"resolved_model": model,
|
|
},
|
|
)
|
|
_atomic_json(
|
|
output / "model-manifest.json",
|
|
{
|
|
"schema_version": 1,
|
|
"title": model["title"],
|
|
"unit": "mm",
|
|
"valid": True,
|
|
"result": model["result"],
|
|
"step_count": len(model["steps"]),
|
|
"operation_counts": dict(
|
|
sorted(Counter(step["op"] for step in model["steps"]).items())
|
|
),
|
|
"geometry": step_metrics,
|
|
"surface_types": _surface_types(step_shape),
|
|
"roundtrip": {"fcstd": "verified", "step": "verified"},
|
|
"input": None
|
|
if input_path is None
|
|
else {
|
|
"filename": input_path.name,
|
|
"size_bytes": input_path.stat().st_size,
|
|
"sha256": _sha256(input_path),
|
|
},
|
|
},
|
|
)
|
|
_atomic_json(metadata / "preview-mesh.json", _mesh(step_shape))
|
|
_atomic_json(
|
|
metadata / "provenance.json",
|
|
{
|
|
"adapter_version": payload["adapter_version"],
|
|
"freecad_version": payload["software_version"],
|
|
"request_digest": payload.get("request_digest"),
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"network_access": False,
|
|
"gui_started": False,
|
|
"safe_mode": True,
|
|
"fcstd_sha256": _sha256(project_path),
|
|
"step_sha256": _sha256(step_path),
|
|
},
|
|
)
|
|
|
|
|
|
def _entry() -> None:
|
|
raw_job_dir = os.environ.get(JOB_ENV)
|
|
if not raw_job_dir:
|
|
return
|
|
job_dir = Path(raw_job_dir).resolve(strict=True)
|
|
result_path = job_dir / "output" / ".meta" / "engine-result.json"
|
|
try:
|
|
run(job_dir)
|
|
_atomic_json(result_path, {"status": "succeeded"})
|
|
except Exception as exc: # noqa: BLE001 - fixed engine boundary must report all failures
|
|
code = exc.code if isinstance(exc, EngineError) else "FREECAD_ENGINE_FAILED"
|
|
_atomic_json(
|
|
result_path,
|
|
{
|
|
"status": "failed",
|
|
"code": code,
|
|
"detail": f"{type(exc).__name__}: {exc}"[:500],
|
|
},
|
|
)
|
|
print(f"[ERR] {code}: {exc}", file=sys.stderr)
|
|
|
|
|
|
# FreeCADCmd can execute Python files by importing them, so do not rely on __main__.
|
|
_entry()
|