651 lines
29 KiB
Python
651 lines
29 KiB
Python
"""Declarative, headless neutral-CAD adapter backed by pinned CadQuery/OCP."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from collections import Counter
|
|
from datetime import datetime, timezone
|
|
from importlib.metadata import version
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
ADAPTER_VERSION = "1.0.0"
|
|
RECIPE_VERSION = 1
|
|
MAX_SOLIDS = 256
|
|
OUTPUT_MEDIA = {
|
|
"geometry.step": ("geometry", "model/step"),
|
|
"geometry-recipe.json": ("geometry_recipe", "application/json"),
|
|
"geometry-manifest.json": ("geometry_manifest", "application/json"),
|
|
"geometry-preview.png": ("geometry_preview", "image/png"),
|
|
"provenance.json": ("provenance", "application/json"),
|
|
}
|
|
|
|
|
|
def _atomic_json(path: Path, value: Any) -> None:
|
|
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 _artifact(path: Path) -> dict[str, Any]:
|
|
artifact_id, media_type = OUTPUT_MEDIA[path.name]
|
|
return {
|
|
"artifact_id": artifact_id,
|
|
"filename": path.name,
|
|
"media_type": media_type,
|
|
"size_bytes": path.stat().st_size,
|
|
"sha256": _sha256(path),
|
|
}
|
|
|
|
|
|
def _terminal(record: dict[str, Any], status: str, artifacts: list[dict[str, Any]], error: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"job_id": record.get("job_id"),
|
|
"lease_id": record.get("lease_id"),
|
|
"request_digest": record.get("request_digest"),
|
|
"status": status,
|
|
"artifacts": artifacts,
|
|
"error": error,
|
|
"finished_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
def _input_file(job_dir: Path, key: str) -> Path:
|
|
input_root = (job_dir / "input").resolve(strict=True)
|
|
if not input_root.is_relative_to(job_dir):
|
|
raise ValueError("INPUT_ROOT_ESCAPES_JOB")
|
|
directory = (input_root / key).resolve(strict=True)
|
|
if not directory.is_relative_to(input_root):
|
|
raise ValueError("INPUT_DIRECTORY_ESCAPES_JOB")
|
|
files = [item for item in directory.iterdir() if item.is_file() and not item.name.startswith(".")]
|
|
if len(files) != 1:
|
|
raise ValueError(f"INPUT_FILE_COUNT_INVALID:{key}")
|
|
path = files[0].resolve(strict=True)
|
|
if not path.is_relative_to(input_root):
|
|
raise ValueError("INPUT_PATH_ESCAPES_JOB")
|
|
if path.suffix.lower() not in {".step", ".stp"}:
|
|
raise ValueError("STEP_INPUT_REQUIRED")
|
|
if path.stat().st_size > 256 * 1024 * 1024:
|
|
raise ValueError("STEP_INPUT_TOO_LARGE")
|
|
return path
|
|
|
|
|
|
def _require_step_unit(path: Path) -> None:
|
|
from OCP.IFSelect import IFSelect_RetDone
|
|
from OCP.STEPControl import STEPControl_Reader
|
|
from OCP.TColStd import TColStd_SequenceOfAsciiString
|
|
|
|
reader = STEPControl_Reader()
|
|
if reader.ReadFile(str(path)) != IFSelect_RetDone:
|
|
raise ValueError("STEP_INPUT_INVALID")
|
|
lengths = TColStd_SequenceOfAsciiString()
|
|
angles = TColStd_SequenceOfAsciiString()
|
|
solid_angles = TColStd_SequenceOfAsciiString()
|
|
reader.FileUnits(lengths, angles, solid_angles)
|
|
units = {
|
|
lengths.Value(index).ToCString().strip().casefold()
|
|
for index in range(1, lengths.Length() + 1)
|
|
if lengths.Value(index).ToCString().strip()
|
|
}
|
|
if len(units) != 1:
|
|
raise ValueError("STEP_UNIT_MISSING_OR_AMBIGUOUS")
|
|
|
|
|
|
def _vector(value: Iterable[float]):
|
|
import cadquery as cq
|
|
|
|
return cq.Vector(*(float(item) for item in value))
|
|
|
|
|
|
def _primitive(spec: dict[str, Any]):
|
|
import cadquery as cq
|
|
|
|
shape = spec["shape"]
|
|
required = {
|
|
"box": {"length", "width", "height"},
|
|
"cylinder": {"radius", "height"},
|
|
"tube": {"radius", "inner_radius", "height"},
|
|
"cone": {"radius", "height"},
|
|
"frustum": {"radius1", "radius2", "height"},
|
|
"sphere": {"radius"},
|
|
"wedge": {"x_min", "x_max", "y_min", "y_max", "z_min", "z_max"},
|
|
"torus": {"major_radius", "minor_radius"},
|
|
}[shape]
|
|
allowed = required | {"shape"}
|
|
if shape in {"cylinder", "tube", "cone", "frustum", "sphere", "torus"}:
|
|
allowed.add("angle_degrees")
|
|
if set(spec) - allowed or not required.issubset(spec):
|
|
raise ValueError(f"PRIMITIVE_PARAMETERS_INVALID:{shape}")
|
|
angle = float(spec.get("angle_degrees", 360.0))
|
|
if shape == "box":
|
|
result = cq.Solid.makeBox(spec["length"], spec["width"], spec["height"])
|
|
elif shape == "cylinder":
|
|
result = cq.Solid.makeCylinder(spec["radius"], spec["height"], angleDegrees=angle)
|
|
elif shape == "tube":
|
|
if spec["inner_radius"] >= spec["radius"]:
|
|
raise ValueError("TUBE_INNER_RADIUS_MUST_BE_SMALLER")
|
|
outer = cq.Solid.makeCylinder(spec["radius"], spec["height"], angleDegrees=angle)
|
|
inner = cq.Solid.makeCylinder(spec["inner_radius"], spec["height"], angleDegrees=angle)
|
|
result = outer.cut(inner)
|
|
elif shape in {"cone", "frustum"}:
|
|
radius1 = spec["radius"] if shape == "cone" else spec["radius1"]
|
|
radius2 = 0.0 if shape == "cone" else spec["radius2"]
|
|
if radius1 == 0 and radius2 == 0:
|
|
raise ValueError("FRUSTUM_REQUIRES_NONZERO_RADIUS")
|
|
result = cq.Solid.makeCone(radius1, radius2, spec["height"], angleDegrees=angle)
|
|
elif shape == "sphere":
|
|
result = cq.Solid.makeSphere(spec["radius"], angleDegrees1=-90, angleDegrees2=90, angleDegrees3=angle)
|
|
elif shape == "wedge":
|
|
if not (spec["x_min"] < spec["x_max"] and spec["y_min"] < spec["y_max"] and spec["z_min"] < spec["z_max"]):
|
|
raise ValueError("WEDGE_BOUNDS_INVALID")
|
|
dx = spec["x_max"] - spec["x_min"]
|
|
dy = spec["y_max"] - spec["y_min"]
|
|
dz = spec["z_max"] - spec["z_min"]
|
|
result = cq.Solid.makeWedge(dx, dy, dz, 0, 0, 0, dz)
|
|
result = result.translate((spec["x_min"], spec["y_min"], spec["z_min"]))
|
|
else:
|
|
if spec["minor_radius"] >= spec["major_radius"]:
|
|
raise ValueError("TORUS_MINOR_RADIUS_MUST_BE_SMALLER")
|
|
result = cq.Solid.makeTorus(
|
|
spec["major_radius"], spec["minor_radius"], angleDegrees2=angle
|
|
)
|
|
return _checked(result, f"primitive:{shape}")
|
|
|
|
|
|
def _checked(shape: Any, role: str):
|
|
if shape is None or not shape.isValid():
|
|
raise RuntimeError(f"CAD_SHAPE_INVALID:{role}")
|
|
solids = list(shape.Solids())
|
|
if not solids:
|
|
raise RuntimeError(f"CAD_SHAPE_HAS_NO_SOLIDS:{role}")
|
|
if len(solids) > MAX_SOLIDS:
|
|
raise RuntimeError(f"CAD_SOLID_LIMIT_EXCEEDED:{len(solids)}")
|
|
return shape
|
|
|
|
|
|
def _profiles(items: list[dict[str, Any]]) -> dict[str, Any]:
|
|
import cadquery as cq
|
|
|
|
result: dict[str, Any] = {}
|
|
for item in items:
|
|
name = item["id"]
|
|
if name in result:
|
|
raise ValueError(f"DUPLICATE_PROFILE_ID:{name}")
|
|
points = [cq.Vector(*map(float, point)) for point in item["points"]]
|
|
if points[0].sub(points[-1]).Length < 1e-12:
|
|
points.pop()
|
|
if len(points) < 3:
|
|
raise ValueError(f"PROFILE_TOO_SHORT:{name}")
|
|
normal = None
|
|
for index in range(1, len(points) - 1):
|
|
candidate = points[index].sub(points[0]).cross(points[index + 1].sub(points[0]))
|
|
if candidate.Length > 1e-10:
|
|
normal = candidate.normalized()
|
|
break
|
|
if normal is None:
|
|
raise ValueError(f"PROFILE_COLLINEAR:{name}")
|
|
scale = max(1.0, max(point.sub(points[0]).Length for point in points))
|
|
if any(abs(point.sub(points[0]).dot(normal)) > scale * 1e-9 for point in points):
|
|
raise ValueError(f"PROFILE_NOT_PLANAR:{name}")
|
|
wire = cq.Wire.makePolygon(points + [points[0]])
|
|
if not wire.IsClosed():
|
|
raise ValueError(f"PROFILE_NOT_CLOSED:{name}")
|
|
face = cq.Face.makeFromWires(wire)
|
|
if not face.isValid() or face.Area() <= scale * scale * 1e-12:
|
|
raise ValueError(f"PROFILE_INVALID:{name}")
|
|
result[name] = face
|
|
return result
|
|
|
|
|
|
def _lookup(values: dict[str, Any], name: str, role: str):
|
|
try:
|
|
return values[name]
|
|
except KeyError as exc:
|
|
raise ValueError(f"UNKNOWN_{role.upper()}:{name}") from exc
|
|
|
|
|
|
def _require_keys(step: dict[str, Any], required: set[str], optional: set[str] = set()) -> None:
|
|
allowed = {"id", "op"} | required | optional
|
|
if not required.issubset(step) or set(step) - allowed:
|
|
raise ValueError(f"STEP_PARAMETERS_INVALID:{step['id']}:{step['op']}")
|
|
|
|
|
|
def _combine(shapes: list[Any]):
|
|
import cadquery as cq
|
|
|
|
flattened = [solid for shape in shapes for solid in shape.Solids()]
|
|
return cq.Compound.makeCompound(flattened)
|
|
|
|
|
|
def _recipe(spec: dict[str, Any], input_path: Path | None, unit: str):
|
|
import cadquery as cq
|
|
|
|
profiles = _profiles(spec["profiles"])
|
|
values: dict[str, Any] = {}
|
|
for step in spec["steps"]:
|
|
name, op = step["id"], step["op"]
|
|
if name in values or name in profiles:
|
|
raise ValueError(f"DUPLICATE_RECIPE_ID:{name}")
|
|
if op in {"box", "cylinder", "tube", "cone", "frustum", "sphere", "wedge", "torus"}:
|
|
_require_keys(step, {"primitive"})
|
|
if step["primitive"]["shape"] != op:
|
|
raise ValueError(f"STEP_PRIMITIVE_MISMATCH:{name}")
|
|
shape = _primitive(step["primitive"])
|
|
elif op == "import_step":
|
|
_require_keys(step, {"input"})
|
|
if step["input"] != "geometry":
|
|
raise ValueError("IMPORT_STEP_INPUT_MUST_BE_GEOMETRY")
|
|
if input_path is None:
|
|
raise ValueError("IMPORT_STEP_INPUT_NOT_BOUND")
|
|
shape = cq.importers.importStep(str(input_path), unit=unit.upper()).val()
|
|
elif op == "extrude":
|
|
_require_keys(step, {"profile", "vector"})
|
|
face = _lookup(profiles, step["profile"], "profile")
|
|
if _vector(step["vector"]).Length <= 1e-12:
|
|
raise ValueError("EXTRUDE_VECTOR_MUST_BE_NONZERO")
|
|
shape = cq.Solid.extrudeLinear(
|
|
face.outerWire(), list(face.innerWires()), _vector(step["vector"])
|
|
)
|
|
elif op == "revolve":
|
|
_require_keys(step, {"profile", "axis_start", "axis_end", "angle_degrees"})
|
|
start, end = _vector(step["axis_start"]), _vector(step["axis_end"])
|
|
if end.sub(start).Length <= 1e-12:
|
|
raise ValueError("REVOLVE_AXIS_MUST_BE_NONZERO")
|
|
if abs(float(step["angle_degrees"])) <= 1e-12:
|
|
raise ValueError("REVOLVE_ANGLE_MUST_BE_NONZERO")
|
|
face = _lookup(profiles, step["profile"], "profile")
|
|
shape = cq.Solid.revolve(
|
|
face.outerWire(), list(face.innerWires()),
|
|
step["angle_degrees"], start, end
|
|
)
|
|
elif op == "sweep":
|
|
_require_keys(step, {"profile", "path"})
|
|
path = cq.Wire.makePolygon([_vector(point) for point in step["path"]])
|
|
face = _lookup(profiles, step["profile"], "profile")
|
|
shape = cq.Solid.sweep(
|
|
face.outerWire(), list(face.innerWires()), path
|
|
)
|
|
elif op == "loft":
|
|
_require_keys(step, {"profiles"})
|
|
wires = [_lookup(profiles, item, "profile").outerWire() for item in step["profiles"]]
|
|
shape = cq.Solid.makeLoft(wires)
|
|
elif op in {"union", "cut", "intersect"}:
|
|
_require_keys(step, {"target", "tools"})
|
|
shape = _lookup(values, step["target"], "step")
|
|
tools = [_lookup(values, item, "step") for item in step["tools"]]
|
|
if op == "union":
|
|
for tool in tools:
|
|
shape = shape.fuse(tool)
|
|
elif op == "cut":
|
|
for tool in tools:
|
|
shape = shape.cut(tool)
|
|
else:
|
|
for tool in tools:
|
|
shape = shape.intersect(tool)
|
|
elif op in {"fillet", "chamfer", "shell"}:
|
|
amount_key = "thickness" if op == "shell" else "radius" if op == "fillet" else "distance"
|
|
_require_keys(step, {"target", "selector", amount_key})
|
|
base = _lookup(values, step["target"], "step")
|
|
selected = _select(base, "face" if op == "shell" else "edge", step["selector"])
|
|
if op == "fillet":
|
|
shape = base.fillet(step[amount_key], selected)
|
|
elif op == "chamfer":
|
|
shape = base.chamfer(step[amount_key], None, selected)
|
|
else:
|
|
shape = cq.Workplane(obj=base).newObject(selected).shell(
|
|
-step[amount_key]
|
|
).val()
|
|
elif op == "translate":
|
|
_require_keys(step, {"target", "vector"})
|
|
shape = _lookup(values, step["target"], "step").translate(step["vector"])
|
|
elif op == "rotate":
|
|
_require_keys(step, {"target", "axis_start", "axis_end", "angle_degrees"})
|
|
shape = _lookup(values, step["target"], "step").rotate(step["axis_start"], step["axis_end"], step["angle_degrees"])
|
|
elif op == "mirror":
|
|
_require_keys(step, {"target", "plane"})
|
|
shape = _lookup(values, step["target"], "step").mirror(step["plane"])
|
|
elif op == "linear_pattern":
|
|
_require_keys(step, {"target", "vector", "count", "spacing"})
|
|
direction = _vector(step["vector"])
|
|
if direction.Length <= 1e-12:
|
|
raise ValueError("LINEAR_PATTERN_VECTOR_MUST_BE_NONZERO")
|
|
direction = direction.normalized().multiply(step["spacing"])
|
|
base = _lookup(values, step["target"], "step")
|
|
shape = _combine([base.translate(direction.multiply(index).toTuple()) for index in range(step["count"])])
|
|
elif op == "circular_pattern":
|
|
_require_keys(step, {"target", "axis_start", "axis_end", "count", "angle_degrees"})
|
|
if _vector(step["axis_end"]).sub(_vector(step["axis_start"])).Length <= 1e-12:
|
|
raise ValueError("CIRCULAR_PATTERN_AXIS_MUST_BE_NONZERO")
|
|
base = _lookup(values, step["target"], "step")
|
|
shape = _combine([base.rotate(step["axis_start"], step["axis_end"], index * step["angle_degrees"] / step["count"]) for index in range(step["count"])])
|
|
elif op == "combine":
|
|
_require_keys(step, {"targets"})
|
|
shape = _combine([_lookup(values, item, "step") for item in step["targets"]])
|
|
else:
|
|
raise ValueError(f"STEP_OPERATION_UNSUPPORTED:{op}")
|
|
values[name] = _checked(shape, f"step:{name}")
|
|
return _lookup(values, spec["result"], "result")
|
|
|
|
|
|
def _surface_type(face: Any) -> str:
|
|
value = str(face.geomType()).casefold()
|
|
return next((item for item in ("plane", "cylinder", "cone", "sphere", "torus") if item in value), "other")
|
|
|
|
|
|
def _curve_type(edge: Any) -> str:
|
|
value = str(edge.geomType()).casefold()
|
|
return next((item for item in ("line", "circle", "ellipse", "spline") if item in value), "other")
|
|
|
|
|
|
def _face_data(face: Any) -> dict[str, Any]:
|
|
center = face.Center()
|
|
normal = None
|
|
if _surface_type(face) == "plane":
|
|
try:
|
|
normal = list(face.normalAt(center).toTuple())
|
|
except Exception: # noqa: BLE001 - OCCT surface query varies by face kind
|
|
normal = None
|
|
return {
|
|
"surface": _surface_type(face),
|
|
"centroid": list(center.toTuple()),
|
|
"normal": normal,
|
|
"area": float(face.Area()),
|
|
}
|
|
|
|
|
|
def _assert_count(selector: dict[str, Any], selected: list[Any]) -> None:
|
|
expected = selector.get("expected_count")
|
|
if expected is not None and len(selected) != expected:
|
|
raise RuntimeError(f"GEOMETRY_SCOPE_COUNT_MISMATCH:expected={expected},actual={len(selected)}")
|
|
|
|
|
|
def _select(shape: Any, topology: str, selector: dict[str, Any]) -> list[Any]:
|
|
if topology == "solid":
|
|
if selector["type"] != "all_solids":
|
|
raise ValueError("SOLID_SELECTOR_TYPE_UNSUPPORTED")
|
|
selected = list(shape.Solids())
|
|
elif topology == "edge":
|
|
if selector["type"] != "curve_edges":
|
|
raise ValueError("EDGE_SELECTOR_TYPE_UNSUPPORTED")
|
|
selected = [edge for edge in shape.Edges() if _curve_type(edge) == selector["curve"]]
|
|
elif topology == "face":
|
|
faces = list(shape.Faces())
|
|
if selector["type"] == "surface_faces":
|
|
selected = [face for face in faces if _surface_type(face) == selector["surface"]]
|
|
else:
|
|
planar = [(face, _face_data(face)) for face in faces if _surface_type(face) == "plane"]
|
|
if not planar:
|
|
raise RuntimeError("GEOMETRY_SCOPE_NO_PLANAR_FACES")
|
|
angle = math.radians(float(selector.get("angle_tolerance_degrees", 5.0)))
|
|
cosine = math.cos(angle)
|
|
if selector["type"] == "extreme_face":
|
|
axis = {"x": 0, "y": 1, "z": 2}[selector["axis"]]
|
|
aligned = [(face, data) for face, data in planar if abs(data["normal"][axis]) >= cosine]
|
|
if not aligned:
|
|
raise RuntimeError("GEOMETRY_SCOPE_NO_AXIS_ALIGNED_FACES")
|
|
coordinates = [data["centroid"][axis] for _, data in aligned]
|
|
extreme = min(coordinates) if selector["side"] == "min" else max(coordinates)
|
|
box = shape.BoundingBox()
|
|
span = (box.xlen, box.ylen, box.zlen)[axis]
|
|
tolerance = float(selector.get("tolerance", max(abs(span) * 1e-7, 1e-12)))
|
|
selected = [face for face, data in aligned if abs(data["centroid"][axis] - extreme) <= tolerance]
|
|
elif selector["type"] == "planar_faces":
|
|
normal = [float(item) for item in selector["normal"]]
|
|
magnitude = math.sqrt(sum(item * item for item in normal))
|
|
if magnitude <= 1e-12:
|
|
raise ValueError("PLANE_NORMAL_MUST_BE_NONZERO")
|
|
normal = [item / magnitude for item in normal]
|
|
selected = []
|
|
for face, data in planar:
|
|
alignment = abs(sum(a * b for a, b in zip(normal, data["normal"], strict=True)))
|
|
position = sum(a * b for a, b in zip(normal, data["centroid"], strict=True))
|
|
if alignment >= cosine and abs(position - selector["offset"]) <= selector["tolerance"]:
|
|
selected.append(face)
|
|
else:
|
|
raise ValueError("FACE_SELECTOR_TYPE_UNSUPPORTED")
|
|
else:
|
|
raise ValueError("REGION_TOPOLOGY_UNSUPPORTED")
|
|
if not selected:
|
|
raise RuntimeError(f"GEOMETRY_SCOPE_EMPTY:{selector['type']}")
|
|
_assert_count(selector, selected)
|
|
return selected
|
|
|
|
|
|
def _regions(shape: Any, specs: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
names: set[str] = set()
|
|
result = []
|
|
for spec in specs:
|
|
if spec["name"] in names:
|
|
raise ValueError(f"DUPLICATE_REGION_NAME:{spec['name']}")
|
|
names.add(spec["name"])
|
|
selected = _select(shape, spec["topology"], spec["selector"])
|
|
summaries = []
|
|
for item in selected:
|
|
if spec["topology"] == "face":
|
|
summaries.append(_face_data(item))
|
|
elif spec["topology"] == "edge":
|
|
summaries.append({"curve": _curve_type(item), "length": float(item.Length()), "centroid": list(item.Center().toTuple())})
|
|
else:
|
|
summaries.append({"volume": float(item.Volume()), "centroid": list(item.Center().toTuple())})
|
|
result.append({"name": spec["name"], "topology": spec["topology"], "selector": spec["selector"], "count": len(selected), "entities": summaries})
|
|
return result
|
|
|
|
|
|
def _manifest(shape: Any, prepare: dict[str, Any], resolved_regions: list[dict[str, Any]], source: dict[str, Any] | None) -> dict[str, Any]:
|
|
box = shape.BoundingBox()
|
|
center = shape.Center()
|
|
surfaces = Counter(_surface_type(face) for face in shape.Faces())
|
|
curves = Counter(_curve_type(edge) for edge in shape.Edges())
|
|
return {
|
|
"schema_version": 1,
|
|
"title": prepare["title"],
|
|
"canonical_unit": prepare["canonical_unit"],
|
|
"valid": bool(shape.isValid()),
|
|
"topology": {
|
|
"solids": len(shape.Solids()), "shells": len(shape.Shells()),
|
|
"faces": len(shape.Faces()), "wires": len(shape.Wires()),
|
|
"edges": len(shape.Edges()), "vertices": len(shape.Vertices()),
|
|
},
|
|
"bounding_box": {"min": [box.xmin, box.ymin, box.zmin], "max": [box.xmax, box.ymax, box.zmax], "size": [box.xlen, box.ylen, box.zlen]},
|
|
"area": float(shape.Area()),
|
|
"volume": float(shape.Volume()),
|
|
"center_of_mass": list(center.toTuple()),
|
|
"shape_description": {"kind": "multi_solid" if len(shape.Solids()) > 1 else "solid", "surface_types": dict(sorted(surfaces.items())), "curve_types": dict(sorted(curves.items()))},
|
|
"regions": resolved_regions,
|
|
"input_summary": source,
|
|
"output_summary": {"format": "STEP", "filename": "geometry.step", "solid_count": len(shape.Solids())},
|
|
}
|
|
|
|
|
|
def _preview(shape: Any, path: Path) -> None:
|
|
from PIL import Image, ImageDraw
|
|
|
|
box = shape.BoundingBox()
|
|
tolerance = max(max(box.xlen, box.ylen, box.zlen) * 1e-3, 1e-3)
|
|
vertices, triangles = shape.tessellate(tolerance, 0.15)
|
|
if not vertices or not triangles or len(triangles) > 500_000:
|
|
raise RuntimeError("PREVIEW_MESH_INVALID_OR_TOO_LARGE")
|
|
projected = []
|
|
for vertex in vertices:
|
|
x, y, z = vertex.toTuple()
|
|
projected.append((0.866 * (x - y), 0.5 * (x + y) - z, 0.408 * (x + y + z)))
|
|
xs, ys = [p[0] for p in projected], [p[1] for p in projected]
|
|
span = max(max(xs) - min(xs), max(ys) - min(ys), 1e-12)
|
|
scale = 1000.0 / span
|
|
points = [((x - (min(xs) + max(xs)) / 2) * scale + 600, (y - (min(ys) + max(ys)) / 2) * scale + 600, depth) for x, y, depth in projected]
|
|
image = Image.new("RGB", (1200, 1200), "white")
|
|
draw = ImageDraw.Draw(image)
|
|
ordered = sorted(triangles, key=lambda tri: sum(points[index][2] for index in tri) / 3)
|
|
light = (0.25, -0.5, 0.83)
|
|
for triangle in ordered:
|
|
a, b, c = (vertices[index] for index in triangle)
|
|
normal = b.sub(a).cross(c.sub(a))
|
|
magnitude = max(normal.Length, 1e-12)
|
|
shade = 0.35 + 0.55 * abs(normal.dot(_vector(light)) / magnitude)
|
|
color = tuple(int(channel * shade) for channel in (112, 160, 210))
|
|
draw.polygon([(points[index][0], points[index][1]) for index in triangle], fill=color, outline=(55, 75, 95))
|
|
temporary = path.with_name(path.name + ".tmp.png")
|
|
image.save(temporary, format="PNG", optimize=True)
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def _export_step(shape: Any, path: Path, unit: str) -> None:
|
|
temporary = path.with_name(path.name + ".tmp.step")
|
|
try:
|
|
shape.exportStep(str(temporary), unit=unit.upper())
|
|
if not temporary.exists() or temporary.stat().st_size < 256:
|
|
raise RuntimeError("STEP_EXPORT_EMPTY")
|
|
os.replace(temporary, path)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def _load_record(job_dir: Path) -> dict[str, Any]:
|
|
record = json.loads((job_dir / "request" / "request.json").read_text(encoding="utf-8"))
|
|
request = record.get("request")
|
|
if not isinstance(request, dict):
|
|
raise TypeError("JOB_REQUEST_MISSING")
|
|
return record
|
|
|
|
|
|
def run(job_dir: Path) -> list[dict[str, Any]]:
|
|
import cadquery as cq
|
|
|
|
job_dir = job_dir.resolve(strict=True)
|
|
record = _load_record(job_dir)
|
|
request = record["request"]
|
|
prepare = request["operation"]["prepare"]
|
|
bindings = {item["key"] for item in request["inputs"]}
|
|
feature = prepare["type"]
|
|
if request["outputs"]:
|
|
raise ValueError("CAD_OUTPUTS_MUST_BE_EMPTY")
|
|
input_path = _input_file(job_dir, "geometry") if "geometry" in bindings else None
|
|
if input_path is not None:
|
|
_require_step_unit(input_path)
|
|
if feature == "inspect":
|
|
if input_path is None or prepare["geometry"]["input"] != "geometry" or len(bindings) != 1:
|
|
raise ValueError("INSPECT_INPUT_NOT_BOUND")
|
|
shape = cq.importers.importStep(str(input_path), unit=prepare["canonical_unit"].upper()).val()
|
|
elif feature == "primitive":
|
|
if bindings:
|
|
raise ValueError("PRIMITIVE_INPUTS_MUST_BE_EMPTY")
|
|
shape = _primitive(prepare["primitive"])
|
|
elif feature == "recipe":
|
|
shape = _recipe(prepare["recipe"], input_path, prepare["canonical_unit"])
|
|
else:
|
|
raise ValueError("CAD_FEATURE_UNSUPPORTED")
|
|
shape = _checked(shape, "result")
|
|
source = None if input_path is None else {"key": "geometry", "filename": input_path.name, "size_bytes": input_path.stat().st_size, "sha256": _sha256(input_path)}
|
|
normalized_recipe = {
|
|
"schema_version": RECIPE_VERSION,
|
|
"feature": feature,
|
|
"title": prepare["title"],
|
|
"canonical_unit": prepare["canonical_unit"],
|
|
"source": source,
|
|
"definition": prepare.get("primitive") or prepare.get("recipe") or {"input": "geometry"},
|
|
"regions": prepare["regions"],
|
|
}
|
|
output = job_dir / "output"
|
|
output.mkdir(exist_ok=True)
|
|
output = output.resolve(strict=True)
|
|
if not output.is_relative_to(job_dir):
|
|
raise ValueError("OUTPUT_PATH_ESCAPES_JOB")
|
|
paths = {
|
|
"step": output / "geometry.step", "recipe": output / "geometry-recipe.json",
|
|
"manifest": output / "geometry-manifest.json", "preview": output / "geometry-preview.png",
|
|
"provenance": output / "provenance.json",
|
|
}
|
|
_export_step(shape, paths["step"], prepare["canonical_unit"])
|
|
reopened = cq.importers.importStep(str(paths["step"]), unit=prepare["canonical_unit"].upper()).val()
|
|
reopened = _checked(reopened, "step_roundtrip")
|
|
tolerance = max(1e-9, abs(float(shape.Volume())) * 1e-8)
|
|
if abs(float(reopened.Volume()) - float(shape.Volume())) > tolerance:
|
|
raise RuntimeError("STEP_ROUNDTRIP_VOLUME_MISMATCH")
|
|
resolved_regions = _regions(reopened, prepare["regions"])
|
|
_atomic_json(paths["recipe"], normalized_recipe)
|
|
_atomic_json(paths["manifest"], _manifest(reopened, prepare, resolved_regions, source))
|
|
_preview(reopened, paths["preview"])
|
|
_atomic_json(paths["provenance"], {
|
|
"adapter_version": ADAPTER_VERSION,
|
|
"cadquery_version": version("cadquery"),
|
|
"ocp_version": version("cadquery-ocp"),
|
|
"request_digest": record.get("request_digest"),
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"network_access": False,
|
|
"gui_started": False,
|
|
"step_sha256": _sha256(paths["step"]),
|
|
})
|
|
result_paths = [paths[key] for key in ("step", "recipe", "manifest", "preview", "provenance")]
|
|
return [_artifact(path) for path in result_paths]
|
|
|
|
|
|
def _probe() -> int:
|
|
payload: dict[str, Any] = {"adapter_version": ADAPTER_VERSION, "software": "CadQuery/OCP", "software_version": None, "health": "unavailable", "detail": ""}
|
|
try:
|
|
import cadquery as cq
|
|
|
|
with tempfile.TemporaryDirectory(prefix="zcbot-cad-probe-") as directory:
|
|
path = Path(directory) / "probe.step"
|
|
box = cq.Solid.makeBox(2, 3, 5)
|
|
box.exportStep(str(path), unit="MM")
|
|
reopened = cq.importers.importStep(str(path), unit="MM").val()
|
|
if not reopened.isValid() or abs(reopened.Volume() - 30.0) > 1e-8:
|
|
raise RuntimeError("CAD_PROBE_STEP_ROUNDTRIP_FAILED")
|
|
payload.update({"software_version": f"CadQuery {version('cadquery')} / OCP {version('cadquery-ocp')}", "health": "ready", "detail": "依赖、基本体与 STEP roundtrip 可用"})
|
|
except Exception as exc: # noqa: BLE001 - probe must always return structured health
|
|
payload["detail"] = f"{type(exc).__name__}: {exc}"[:500]
|
|
print(json.dumps(payload, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
if sys.argv[1:] == ["--probe"]:
|
|
return _probe()
|
|
if len(sys.argv) != 2:
|
|
print("[ERR] Usage: worker.py <job-directory>", file=sys.stderr)
|
|
return 2
|
|
job_dir = Path(sys.argv[1])
|
|
record: dict[str, Any] = {}
|
|
try:
|
|
record = _load_record(job_dir)
|
|
artifacts = run(job_dir)
|
|
_atomic_json(job_dir / "artifacts.json", artifacts)
|
|
_atomic_json(job_dir / "terminal.json", _terminal(record, "succeeded", artifacts, {}))
|
|
print("[OK] Neutral CAD geometry prepared.")
|
|
return 0
|
|
except Exception as exc: # noqa: BLE001 - terminal must capture kernel failures
|
|
_atomic_json(job_dir / "terminal.json", _terminal(record, "failed", [], {"code": type(exc).__name__, "detail": str(exc)[:500]}))
|
|
print(f"[ERR] {type(exc).__name__}: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit_code = main()
|
|
sys.stdout.flush()
|
|
sys.stderr.flush()
|
|
if sys.platform == "win32":
|
|
# OCP/VTK extension finalizers can fault during CPython shutdown on
|
|
# Windows after all requested artifacts have already been persisted.
|
|
os._exit(exit_code)
|
|
raise SystemExit(exit_code)
|