1394 lines
73 KiB
Python
1394 lines
73 KiB
Python
#!/usr/bin/env python3
|
|
"""Convert SolidWorks ``solidworks.cad_evidence.v2`` records to CDSL v1.1.
|
|
|
|
The converter deliberately produces semantic CDSL, not executable build123d
|
|
input. It preserves feature records which the current runtime cannot execute
|
|
yet, and records the evidence or STEP-derived references needed by a future
|
|
engine to implement them without re-reading the original SolidWorks model.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import concurrent.futures
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import re
|
|
import sys
|
|
from collections import Counter
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ENGINE_ROOT = ROOT / "backend" / "engine"
|
|
if str(ENGINE_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ENGINE_ROOT))
|
|
|
|
from cdsl_engine.semantic_validation import validate_semantic_cdsl # noqa: E402
|
|
|
|
|
|
METERS_TO_MM = 1000.0
|
|
EPSILON_MM = 1e-4
|
|
EVIDENCE_V2_SUFFIX = ".solidworks_evidence_v2.json"
|
|
SUPPORTED_ATOMS = {
|
|
"extrude_add_blind",
|
|
"extrude_add_two_sided",
|
|
"extrude_cut_blind",
|
|
"revolve_add",
|
|
"revolve_cut",
|
|
"hole_blind",
|
|
"hole_countersink",
|
|
"hole_counterbore",
|
|
}
|
|
SKETCH_TYPES = {"ProfileFeature", "OriginProfileFeature", "3DProfileFeature"}
|
|
MODELING_TYPES = {
|
|
"Boss", "Extrusion", "Cut", "Revolution", "RevCut", "HoleWzd", "Fillet",
|
|
"Chamfer", "LPattern", "MirrorPattern", "RefPlane", "RefAxis",
|
|
}
|
|
END_CONDITIONS = {
|
|
0: "blind", 1: "through_all", 2: "through_all_both", 3: "up_to_vertex",
|
|
4: "up_to_surface", 5: "offset_from_surface", 6: "through_all_and_blind",
|
|
7: "up_to_body", 8: "mid_plane", 9: "through_next",
|
|
}
|
|
|
|
|
|
def _number(value: Any, default: float = 0.0) -> float:
|
|
try:
|
|
number = float(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
return number if math.isfinite(number) else default
|
|
|
|
|
|
def _mm(value: Any) -> float:
|
|
return round(_number(value) * METERS_TO_MM, 9)
|
|
|
|
|
|
def _point2(value: Any) -> list[float] | None:
|
|
if not isinstance(value, (list, tuple)) or len(value) < 2:
|
|
return None
|
|
return [_mm(value[0]), _mm(value[1])]
|
|
|
|
|
|
def _point3(value: Any) -> list[float] | None:
|
|
if not isinstance(value, (list, tuple)) or len(value) < 3:
|
|
return None
|
|
return [_mm(value[0]), _mm(value[1]), _mm(value[2])]
|
|
|
|
|
|
def _safe_id(prefix: str, number: int) -> str:
|
|
return f"{prefix}_{number:03d}"
|
|
|
|
|
|
def _identity_key(value: dict[str, Any]) -> str | None:
|
|
if not isinstance(value, dict):
|
|
return None
|
|
return str(value.get("stable_id") or value.get("persist_reference_sha256") or "") or None
|
|
|
|
|
|
def _record_values(feature: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
"""Return typed properties/methods regardless of direct or history capture."""
|
|
history = feature.get("history_definition") or {}
|
|
props = ((history.get("definition_properties") or feature.get("definition_properties") or {}).get("values")) or {}
|
|
methods = ((history.get("definition_methods") or feature.get("definition_methods") or {}).get("values")) or {}
|
|
return props if isinstance(props, dict) else {}, methods if isinstance(methods, dict) else {}
|
|
|
|
|
|
def _record_parents(feature: dict[str, Any]) -> list[dict[str, Any]]:
|
|
history = feature.get("history_definition") or {}
|
|
value = history.get("parents") or feature.get("parents", [])
|
|
return [item for item in (value or []) if isinstance(item, dict)]
|
|
|
|
|
|
def _record_selections(feature: dict[str, Any]) -> list[dict[str, Any]]:
|
|
history = feature.get("history_definition") or {}
|
|
value = history.get("selections") or feature.get("selections", [])
|
|
return [item for item in (value or []) if isinstance(item, dict)]
|
|
|
|
|
|
def _selector(reference: dict[str, Any], *, source: str = "solidworks", confidence: float = 1.0,
|
|
owner_feature_id: str | None = None, kind_hint: str | None = None) -> dict[str, Any] | None:
|
|
if not isinstance(reference, dict):
|
|
return None
|
|
stable_id = _identity_key(reference)
|
|
kind = kind_hint or str(reference.get("kind") or "")
|
|
if not stable_id or kind not in {"face", "edge", "axis", "plane", "feature", "vertex", "body", "sketch_segment"}:
|
|
return None
|
|
if kind == "sketch_segment":
|
|
kind = "axis"
|
|
result: dict[str, Any] = {
|
|
"kind": kind,
|
|
"stable_id": stable_id,
|
|
"source": source,
|
|
"confidence": round(confidence, 3),
|
|
}
|
|
if owner_feature_id:
|
|
result["owner_feature_id"] = owner_feature_id
|
|
geometry = reference.get("geometry")
|
|
if isinstance(geometry, dict):
|
|
result["geometry"] = _geometry_signature(geometry)
|
|
return result
|
|
|
|
|
|
def _referenced_feature_id(reference: Any, feature_id_by_source: dict[str, str],
|
|
feature_id_by_name: dict[str, str]) -> str | None:
|
|
"""Resolve SolidWorks' persisted reference, falling back to its tree name."""
|
|
if not isinstance(reference, dict):
|
|
return None
|
|
return (
|
|
feature_id_by_source.get(_identity_key(reference) or "")
|
|
or feature_id_by_name.get(str(reference.get("name") or ""))
|
|
)
|
|
|
|
|
|
def _geometry_signature(geometry: dict[str, Any]) -> dict[str, Any]:
|
|
"""Retain a small, stable geometry signature instead of COM object payloads."""
|
|
result: dict[str, Any] = {}
|
|
if isinstance(geometry.get("surface"), dict):
|
|
result["surface"] = {
|
|
key: geometry["surface"][key]
|
|
for key in ("type", "parameters") if key in geometry["surface"]
|
|
}
|
|
if isinstance(geometry.get("curve"), dict):
|
|
result["curve"] = {
|
|
key: geometry["curve"][key]
|
|
for key in ("type", "parameters") if key in geometry["curve"]
|
|
}
|
|
for key in ("start", "end", "box", "area"):
|
|
if key in geometry and geometry[key] is not None:
|
|
result[key] = geometry[key]
|
|
return result
|
|
|
|
|
|
def _normalize(vector: list[float]) -> list[float]:
|
|
length = math.sqrt(sum(component * component for component in vector))
|
|
return [round(component / length, 9) for component in vector] if length > 1e-12 else [0.0, 0.0, 1.0]
|
|
|
|
|
|
def _cross(left: list[float], right: list[float]) -> list[float]:
|
|
return [
|
|
left[1] * right[2] - left[2] * right[1],
|
|
left[2] * right[0] - left[0] * right[2],
|
|
left[0] * right[1] - left[1] * right[0],
|
|
]
|
|
|
|
|
|
def _workplane(sketch: dict[str, Any]) -> dict[str, Any]:
|
|
"""Derive a world-space workplane from SolidWorks' model-to-sketch matrix.
|
|
|
|
SolidWorks stores the inverse transform (model -> sketch). For its rigid
|
|
4x4 matrix, transpose the rotational part and apply the inverse translation.
|
|
A conservative XY fallback keeps the record valid when the exporter did
|
|
not provide a usable matrix.
|
|
"""
|
|
workplane = _workplane_from_matrix(sketch)
|
|
if workplane is not None:
|
|
return workplane
|
|
return {"origin_mm": [0.0, 0.0, 0.0], "x_dir": [1.0, 0.0, 0.0], "y_dir": [0.0, 1.0, 0.0], "normal": [0.0, 0.0, 1.0]}
|
|
|
|
|
|
def _workplane_from_matrix(sketch: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""Parse the exporter's model-to-sketch matrix into a workplane.
|
|
|
|
Returns None when no usable rigid rotation can be extracted (missing
|
|
matrix or a non-orthonormal frame). The caller then decides between a
|
|
rebuilt workplane and the conservative XY fallback.
|
|
"""
|
|
matrix = sketch.get("model_to_sketch_transform")
|
|
if not isinstance(matrix, list) or len(matrix) != 16 or not all(isinstance(item, (int, float)) for item in matrix):
|
|
return None
|
|
# The exporter serializes MathTransform in row-major form.
|
|
rotation = [matrix[0:3], matrix[4:7], matrix[8:11]]
|
|
translation = [matrix[3], matrix[7], matrix[11]]
|
|
inverse_rotation = [[rotation[column][row] for column in range(3)] for row in range(3)]
|
|
origin_m = [-sum(inverse_rotation[row][column] * translation[column] for column in range(3)) for row in range(3)]
|
|
x_dir = _normalize([inverse_rotation[row][0] for row in range(3)])
|
|
y_dir = _normalize([inverse_rotation[row][1] for row in range(3)])
|
|
normal = _normalize(_cross(x_dir, y_dir))
|
|
# A rigid model-to-sketch rotation keeps the sketch axes orthogonal. The
|
|
# exporter sometimes emits a non-orthogonal matrix for profiles placed on
|
|
# a copied reference plane; only trust a frame that is truly orthonormal.
|
|
if abs(x_dir[0] * y_dir[0] + x_dir[1] * y_dir[1] + x_dir[2] * y_dir[2]) > 1e-9:
|
|
return None
|
|
return {"origin_mm": [_mm(value) for value in origin_m], "x_dir": x_dir, "y_dir": y_dir, "normal": normal}
|
|
|
|
|
|
def _workplane_is_orthonormal(workplane: dict[str, Any]) -> bool:
|
|
"""True when a workplane's sketch axes describe an orthonormal frame."""
|
|
x_dir = workplane.get("x_dir")
|
|
y_dir = workplane.get("y_dir")
|
|
if not x_dir or not y_dir:
|
|
return False
|
|
return abs(x_dir[0] * y_dir[0] + x_dir[1] * y_dir[1] + x_dir[2] * y_dir[2]) <= 1e-9
|
|
|
|
|
|
def _workplane_rebuilt(parent_plane: dict[str, Any] | None, axis_direction: list[float] | None) -> dict[str, Any] | None:
|
|
"""Rebuild a workplane from the parent reference plane and the revolve axis.
|
|
|
|
When the exporter's matrix is unusable, the profile still lies on its
|
|
parent reference plane and the revolve axis runs along the sketch V axis.
|
|
The captured axis direction is world space, so V maps directly onto it and
|
|
U is completed by the cross product to keep a right-handed frame.
|
|
"""
|
|
if not parent_plane or not axis_direction:
|
|
return None
|
|
normal = _normalize(list(parent_plane.get("normal") or []))
|
|
y_dir = _normalize(list(axis_direction))
|
|
if abs(normal[0] * y_dir[0] + normal[1] * y_dir[1] + normal[2] * y_dir[2]) > 0.9:
|
|
# The axis would lie parallel to the plane normal and the revolve
|
|
# would degenerate to zero volume; do not trust this rebuild.
|
|
return None
|
|
x_dir = _normalize(_cross(y_dir, normal))
|
|
origin_mm = list(parent_plane.get("origin_mm") or [0.0, 0.0, 0.0])
|
|
return {"origin_mm": origin_mm, "x_dir": x_dir, "y_dir": y_dir, "normal": normal}
|
|
|
|
|
|
def _axis_in_world(axis: dict[str, Any], workplane: dict[str, Any]) -> dict[str, Any]:
|
|
"""Transform a revolve axis captured in sketch-local coordinates to world space."""
|
|
origin = axis.get("origin_mm")
|
|
direction = axis.get("direction")
|
|
if not origin or not direction or not workplane.get("x_dir"):
|
|
return axis
|
|
origin_mm = workplane.get("origin_mm") or [0.0, 0.0, 0.0]
|
|
x_dir = workplane["x_dir"]
|
|
y_dir = workplane.get("y_dir") or [0.0, 1.0, 0.0]
|
|
normal = workplane.get("normal") or [0.0, 0.0, 1.0]
|
|
world_origin = [
|
|
origin_mm[index] + origin[0] * x_dir[index] + origin[1] * y_dir[index] + origin[2] * normal[index]
|
|
for index in range(3)
|
|
]
|
|
world_direction = _normalize([
|
|
direction[0] * x_dir[index] + direction[1] * y_dir[index] + direction[2] * normal[index]
|
|
for index in range(3)
|
|
])
|
|
return {
|
|
**axis,
|
|
"origin_mm": [round(value, 9) for value in world_origin],
|
|
"direction": world_direction if math.sqrt(sum(c * c for c in world_direction)) > 1e-12 else list(direction),
|
|
}
|
|
|
|
|
|
def _workplane_from_plane_reference(reference: Any) -> dict[str, Any] | None:
|
|
"""Build a workplane from an exporter-captured planar face signature."""
|
|
if not isinstance(reference, dict):
|
|
return None
|
|
geometry = reference.get("geometry") if isinstance(reference.get("geometry"), dict) else {}
|
|
surface = geometry.get("surface") if isinstance(geometry.get("surface"), dict) else {}
|
|
parameters = surface.get("parameters") if isinstance(surface.get("parameters"), list) else []
|
|
if surface.get("type") != "plane" or len(parameters) < 6:
|
|
return None
|
|
raw_normal = [_number(value) for value in parameters[0:3]]
|
|
origin = _point3(parameters[3:6])
|
|
if not origin or math.sqrt(sum(value * value for value in raw_normal)) <= 1e-12:
|
|
return None
|
|
normal = _normalize(raw_normal)
|
|
seed = [1.0, 0.0, 0.0] if abs(normal[0]) < 0.9 else [0.0, 1.0, 0.0]
|
|
x_dir = _normalize(_cross(seed, normal))
|
|
return {"origin_mm": origin, "x_dir": x_dir, "normal": normal}
|
|
|
|
|
|
def _segment(item: dict[str, Any]) -> dict[str, Any] | None:
|
|
geometry = item.get("geometry") if isinstance(item.get("geometry"), dict) else {}
|
|
segment_type = str(geometry.get("segment_type") or "")
|
|
curve = geometry.get("curve") if isinstance(geometry.get("curve"), dict) else {}
|
|
curve_type = str(curve.get("type") or "")
|
|
start = _point2(geometry.get("start"))
|
|
end = _point2(geometry.get("end"))
|
|
center = _point2(geometry.get("center"))
|
|
parameters = curve.get("parameters") if isinstance(curve.get("parameters"), list) else []
|
|
|
|
if segment_type == "swSketchLINE" or curve_type == "line":
|
|
return {"type": "line", "start": start, "end": end} if start and end else None
|
|
if segment_type == "swSketchARC" or curve_type == "circle":
|
|
if not center and len(parameters) >= 3:
|
|
center = _point2(parameters[0:3])
|
|
radius_m = geometry.get("radius")
|
|
if radius_m is None and len(parameters) >= 7:
|
|
radius_m = parameters[6]
|
|
radius_mm = _mm(radius_m)
|
|
if not center or radius_mm <= 0:
|
|
return None
|
|
if start and end and math.dist(start, end) > EPSILON_MM:
|
|
return {"type": "arc", "start": start, "end": end, "center": center, "radius_mm": radius_mm,
|
|
"clockwise": int(_number(geometry.get("direction"), 1)) < 0}
|
|
return {"type": "circle", "center": center, "radius_mm": radius_mm}
|
|
if segment_type == "swSketchSPLINE" or curve_type == "other":
|
|
spline = item.get("spline") if isinstance(item.get("spline"), dict) else {}
|
|
raw_points = spline.get("control_points") or []
|
|
dimension = int(_number(spline.get("dimension"), 3))
|
|
if dimension < 2 or len(raw_points) < dimension * 2:
|
|
return None
|
|
points = [_point2(raw_points[index:index + dimension]) for index in range(0, len(raw_points), dimension)]
|
|
points = [point for point in points if point]
|
|
if len(points) < 2:
|
|
return None
|
|
result: dict[str, Any] = {
|
|
"type": "bspline", "degree": int(_number(spline.get("degree"), 3)),
|
|
"control_points": points, "knots": [float(value) for value in spline.get("knots") or []],
|
|
"periodic": bool(spline.get("periodic")),
|
|
}
|
|
if spline.get("rational") and isinstance(spline.get("weights"), list):
|
|
result["weights"] = [float(value) for value in spline["weights"]]
|
|
return result
|
|
return None
|
|
|
|
|
|
def _segment_endpoints(segment: dict[str, Any]) -> tuple[list[float] | None, list[float] | None]:
|
|
if segment["type"] in {"line", "arc"}:
|
|
return segment.get("start"), segment.get("end")
|
|
if segment["type"] == "bspline":
|
|
points = segment.get("control_points") or []
|
|
return (points[0], points[-1]) if points else (None, None)
|
|
return None, None
|
|
|
|
|
|
def _reverse_segment(segment: dict[str, Any]) -> dict[str, Any]:
|
|
result = dict(segment)
|
|
if result["type"] in {"line", "arc"}:
|
|
result["start"], result["end"] = result["end"], result["start"]
|
|
elif result["type"] == "bspline":
|
|
result["control_points"] = list(reversed(result["control_points"]))
|
|
result["knots"] = list(reversed(result["knots"]))
|
|
return result
|
|
|
|
|
|
def _chain_segments(segments: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
|
|
"""Join line/arc/spline records by endpoints, preserving standalone circles."""
|
|
circles = [[segment] for segment in segments if segment["type"] == "circle"]
|
|
pending = [segment for segment in segments if segment["type"] != "circle"]
|
|
chains: list[list[dict[str, Any]]] = []
|
|
while pending:
|
|
chain = [pending.pop(0)]
|
|
while pending:
|
|
_, tail = _segment_endpoints(chain[-1])
|
|
if not tail:
|
|
break
|
|
found: tuple[int, dict[str, Any]] | None = None
|
|
for index, candidate in enumerate(pending):
|
|
start, end = _segment_endpoints(candidate)
|
|
if start and math.dist(tail, start) <= EPSILON_MM:
|
|
found = (index, candidate)
|
|
break
|
|
if end and math.dist(tail, end) <= EPSILON_MM:
|
|
found = (index, _reverse_segment(candidate))
|
|
break
|
|
if found is None:
|
|
break
|
|
index, candidate = found
|
|
pending.pop(index)
|
|
chain.append(candidate)
|
|
chains.append(chain)
|
|
return chains + circles
|
|
|
|
|
|
def _contour_area(contour: list[dict[str, Any]]) -> float:
|
|
points = [segment.get("start") for segment in contour if segment.get("start")]
|
|
if len(points) < 3:
|
|
return 0.0
|
|
return abs(sum(points[index][0] * points[(index + 1) % len(points)][1] - points[(index + 1) % len(points)][0] * points[index][1]
|
|
for index in range(len(points))) / 2)
|
|
|
|
|
|
def _analytic_profile(sketch: dict[str, Any]) -> dict[str, Any]:
|
|
raw_segments = sketch.get("segments") or []
|
|
drawable: list[dict[str, Any]] = []
|
|
construction: list[dict[str, Any]] = []
|
|
for item in raw_segments:
|
|
geometry = item.get("geometry") if isinstance(item, dict) else None
|
|
if not isinstance(geometry, dict):
|
|
continue
|
|
converted = _segment(item)
|
|
if converted is None:
|
|
continue
|
|
(construction if geometry.get("construction") else drawable).append(converted)
|
|
|
|
if len(drawable) == 1 and drawable[0]["type"] == "circle":
|
|
profile: dict[str, Any] = {"type": "circle", "center": drawable[0]["center"], "radius_mm": drawable[0]["radius_mm"]}
|
|
return profile
|
|
if len(drawable) == 2 and all(item["type"] == "circle" for item in drawable):
|
|
left, right = drawable
|
|
if math.dist(left["center"], right["center"]) <= EPSILON_MM:
|
|
smaller, larger = sorted(drawable, key=lambda item: item["radius_mm"])
|
|
# Annulus is emitted as analytic_contours (outer + inner) so the
|
|
# generic runtime profile contract accepts the record; ring_revolve
|
|
# can then revolve the two concentric circles into a hollow solid.
|
|
return {
|
|
"type": "analytic_contours",
|
|
"contours": [
|
|
{"role": "outer", "closed": True, "segments": [larger]},
|
|
{"role": "inner", "closed": True, "segments": [smaller]},
|
|
],
|
|
}
|
|
if drawable and all(item["type"] == "circle" for item in drawable):
|
|
# Multiple independent circles share the "extrude every closed loop" intent;
|
|
# emit them as analytic_contours (one outer contour per circle) so the
|
|
# generic runtime profile contract accepts the record.
|
|
return {
|
|
"type": "analytic_contours",
|
|
"contours": [
|
|
{"role": "outer", "closed": True, "segments": [item]}
|
|
for item in drawable
|
|
],
|
|
}
|
|
|
|
contours = _chain_segments(drawable)
|
|
areas = [_contour_area(contour) for contour in contours]
|
|
outer_index = max(range(len(contours)), key=lambda index: areas[index], default=-1)
|
|
values: list[dict[str, Any]] = []
|
|
for index, contour in enumerate(contours):
|
|
start, end = _segment_endpoints(contour[0]) if contour else (None, None)
|
|
_, tail = _segment_endpoints(contour[-1]) if contour else (None, None)
|
|
closed = bool(len(contour) == 1 and contour[0]["type"] == "circle") or bool(start and tail and math.dist(start, tail) <= EPSILON_MM)
|
|
values.append({"role": "outer" if index == outer_index else "inner", "closed": closed, "segments": contour})
|
|
profile = {"type": "analytic_contours", "contours": values}
|
|
if construction:
|
|
profile["construction"] = construction
|
|
return profile
|
|
|
|
|
|
def _feature_family(feature: dict[str, Any]) -> str | None:
|
|
feature_type = str(feature.get("effective_type") or feature.get("type_name1") or "")
|
|
return feature_type if feature_type in MODELING_TYPES else None
|
|
|
|
|
|
def _parent_feature_ids(feature: dict[str, Any], feature_id_by_source: dict[str, str], feature_id_by_name: dict[str, str],
|
|
previous_id: str | None) -> list[str]:
|
|
dependencies: list[str] = []
|
|
for parent in _record_parents(feature):
|
|
source_id = _identity_key(parent)
|
|
output_id = feature_id_by_source.get(source_id or "") or feature_id_by_name.get(str(parent.get("name") or ""))
|
|
if output_id and output_id not in dependencies:
|
|
dependencies.append(output_id)
|
|
if not dependencies and previous_id:
|
|
dependencies.append(previous_id)
|
|
return dependencies
|
|
|
|
|
|
def _parent_sketch_id(feature: dict[str, Any], sketch_id_by_source: dict[str, str], sketch_id_by_name: dict[str, str]) -> str | None:
|
|
references = [*_record_parents(feature), *[item for item in feature.get("subfeatures") or [] if isinstance(item, dict)]]
|
|
for parent in references:
|
|
sketch_id = sketch_id_by_source.get(_identity_key(parent) or "") or sketch_id_by_name.get(str(parent.get("name") or ""))
|
|
if sketch_id:
|
|
return sketch_id
|
|
return None
|
|
|
|
|
|
def _end_condition(methods: dict[str, Any], forward: bool = True) -> dict[str, Any]:
|
|
suffix = "true" if forward else "false"
|
|
code = int(_number(methods.get(f"GetEndCondition({suffix})"), 0))
|
|
result = {"type": END_CONDITIONS.get(code, f"solidworks_{code}"), "solidworks_code": code}
|
|
reference = methods.get(f"GetEndConditionReference({suffix},0)")
|
|
if isinstance(reference, dict) and isinstance(reference.get("return"), dict):
|
|
selected = _selector(reference["return"])
|
|
if selected:
|
|
result["reference"] = selected
|
|
return result
|
|
|
|
|
|
def _extrude_params(props: dict[str, Any], methods: dict[str, Any]) -> dict[str, Any]:
|
|
distance = _mm(methods.get("GetDepth(true)"))
|
|
reverse_distance = _mm(methods.get("GetDepth(false)"))
|
|
params: dict[str, Any] = {
|
|
"distance_mm": distance,
|
|
"reverse": bool(props.get("ReverseDirection")),
|
|
"end_condition": _end_condition(methods, True),
|
|
}
|
|
if bool(props.get("BothDirections")) or reverse_distance > 0:
|
|
params["reverse_distance_mm"] = reverse_distance
|
|
params["reverse_end_condition"] = _end_condition(methods, False)
|
|
return params
|
|
|
|
|
|
def _axis_from_reference(reference: Any) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
|
|
if not isinstance(reference, dict):
|
|
return None, None
|
|
selected = _selector(reference)
|
|
geometry = reference.get("geometry") if isinstance(reference.get("geometry"), dict) else {}
|
|
start = _point3(geometry.get("start"))
|
|
end = _point3(geometry.get("end"))
|
|
if start and end:
|
|
direction = _normalize([end[index] - start[index] for index in range(3)])
|
|
return {"origin_mm": start, "direction": direction}, selected
|
|
surface = geometry.get("surface") if isinstance(geometry.get("surface"), dict) else {}
|
|
parameters = surface.get("parameters") if isinstance(surface.get("parameters"), list) else []
|
|
if surface.get("type") == "cylinder" and len(parameters) >= 6:
|
|
origin = _point3(parameters[0:3])
|
|
raw_direction = [_number(value) for value in parameters[3:6]]
|
|
direction = _normalize(raw_direction)
|
|
if origin and math.sqrt(sum(component * component for component in raw_direction)) > 1e-12:
|
|
return {"origin_mm": origin, "direction": direction, "selector": selected} if selected else {"origin_mm": origin, "direction": direction}, selected
|
|
if selected:
|
|
return {"selector": selected}, selected
|
|
return None, None
|
|
|
|
|
|
def _captured_selection_values(methods: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""Unwrap the exporter result for COM methods with an out-array return."""
|
|
value = methods.get("GetSelections(null)")
|
|
if isinstance(value, dict):
|
|
value = value.get("return")
|
|
return [item for item in (value or []) if isinstance(item, dict)]
|
|
|
|
|
|
def _revolve_params(props: dict[str, Any], methods: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
|
|
axis, selector = _axis_from_reference(props.get("Axis"))
|
|
unresolved: list[str] = []
|
|
if axis is None:
|
|
axis = {"unresolved": "SolidWorks revolve axis was not captured"}
|
|
unresolved.append("revolve axis was not captured")
|
|
angle = math.degrees(_number(methods.get("GetRevolutionAngle(true)"), 2 * math.pi))
|
|
params: dict[str, Any] = {
|
|
"angle_deg": round(angle, 9), "axis": axis,
|
|
"reverse": bool(props.get("ReverseDirection")),
|
|
"end_condition": _end_condition(methods, True),
|
|
}
|
|
if selector:
|
|
params["axis_selector"] = selector
|
|
return params, unresolved
|
|
|
|
|
|
def _hole_params(props: dict[str, Any], methods: dict[str, Any]) -> tuple[dict[str, Any], list[list[float]]]:
|
|
diameter = next((_mm(props.get(key)) for key in ("ThreadDiameter", "ThruHoleDiameter", "HoleDiameter", "Diameter") if _mm(props.get(key)) > 0), 0.0)
|
|
depth = next((_mm(props.get(key)) for key in ("ThreadDepth", "TapDrillDepth", "ThruHoleDepth", "HoleDepth", "Depth") if _mm(props.get(key)) > 0), 0.0)
|
|
positions: list[list[float]] = []
|
|
for point in methods.get("GetSketchPoints") or []:
|
|
if isinstance(point, dict):
|
|
value = _point3((point.get("geometry") or {}).get("point"))
|
|
if value:
|
|
positions.append(value)
|
|
params: dict[str, Any] = {
|
|
"hole_type": str(props.get("FastenerType") or props.get("Type") or "hole_wizard"),
|
|
"diameter_mm": diameter,
|
|
"depth_mm": depth,
|
|
"end_condition": {"solidworks_code": int(_number(props.get("EndCondition"), 0)), "type": END_CONDITIONS.get(int(_number(props.get("EndCondition"), 0)), "blind")},
|
|
}
|
|
if positions:
|
|
params["positions"] = [{"mm": position} for position in positions]
|
|
thread_diameter = _mm(props.get("ThreadDiameter"))
|
|
if thread_diameter > 0:
|
|
params["thread"] = {"diameter_mm": thread_diameter, "depth_mm": _mm(props.get("ThreadDepth")), "class": props.get("ThreadClass")}
|
|
if _mm(props.get("CounterSinkDiameter")) > 0:
|
|
params["countersink"] = {"diameter_mm": _mm(props.get("CounterSinkDiameter")), "angle_rad": _number(props.get("CounterSinkAngle"))}
|
|
if _mm(props.get("CounterBoreDiameter")) > 0:
|
|
params["counterbore"] = {"diameter_mm": _mm(props.get("CounterBoreDiameter")), "depth_mm": _mm(props.get("CounterBoreDepth"))}
|
|
return params, positions
|
|
|
|
|
|
def _dimension_values(feature: dict[str, Any]) -> list[float]:
|
|
result: list[float] = []
|
|
history = feature.get("history_definition") or {}
|
|
values = history.get("dimensions", feature.get("dimensions", [])) or []
|
|
for value in values:
|
|
if isinstance(value, dict):
|
|
result.append(_mm(value.get("system_value")))
|
|
return result
|
|
|
|
|
|
def _raw_dimension_values(feature: dict[str, Any]) -> list[float]:
|
|
"""Return SolidWorks system values without applying a length conversion.
|
|
|
|
A feature's dimensions may mix metres and radians; callers must choose
|
|
the appropriate unit for each semantic field.
|
|
"""
|
|
history = feature.get("history_definition") or {}
|
|
values = history.get("dimensions", feature.get("dimensions", [])) or []
|
|
return [_number(value.get("system_value")) for value in values if isinstance(value, dict)]
|
|
|
|
|
|
def _deferred_params(family: str, feature: dict[str, Any], props: dict[str, Any], methods: dict[str, Any],
|
|
source_feature_ids: list[str], feature_id_by_source: dict[str, str],
|
|
feature_id_by_name: dict[str, str]) -> tuple[dict[str, Any], list[str]]:
|
|
if family == "Fillet":
|
|
radius = _mm(props.get("Radius")) or (_dimension_values(feature) or [0.0])[0]
|
|
return {"radius_mm": radius, "tangent_propagation": bool(props.get("TangentPropagation"))}, []
|
|
if family == "Chamfer":
|
|
dimensions = _dimension_values(feature)
|
|
raw_dimensions = _raw_dimension_values(feature)
|
|
distance = _mm(props.get("Distance")) or (dimensions or [0.0])[0]
|
|
params: dict[str, Any] = {"distance_mm": distance}
|
|
angle = _number(props.get("EdgeChamferAngle"))
|
|
if angle <= 0 and len(raw_dimensions) > 1:
|
|
# 兜底：EdgeChamferAngle 缺失时，第二个 raw 值按角度（弧度）解释。
|
|
# 注意：SolidWorks 角度 system value 为弧度；Distance-Distance 倒角的
|
|
# 第二距离（米）也会落入该兜底，但真实数据中倒角均为 Distance-Angle
|
|
# （第二值为 π/4 等弧度值），runtime 通过 distance_2_mm/angle_rad
|
|
# 契约消费第二参数，不会把角度静默当等距。
|
|
angle = raw_dimensions[1]
|
|
if angle > 0:
|
|
# Distance-Angle 倒角：角度单位为弧度（build123d 侧由 runtime 换算）。
|
|
params["angle_rad"] = angle
|
|
return params, []
|
|
if family == "LPattern":
|
|
property_sources = [
|
|
_referenced_feature_id(item, feature_id_by_source, feature_id_by_name)
|
|
for item in props.get("PatternFeatureArray") or [] if isinstance(item, dict)
|
|
]
|
|
source_feature_ids = list(dict.fromkeys([*source_feature_ids, *[item for item in property_sources if item]]))
|
|
params = {
|
|
"source_feature_ids": source_feature_ids,
|
|
"direction_1": _pattern_direction(props.get("D1Axis")),
|
|
"spacing_1_mm": _mm(props.get("D1Spacing")),
|
|
"pattern_count_1": max(1, int(_number(props.get("D1TotalInstances"), 1))),
|
|
}
|
|
if int(_number(props.get("D2TotalInstances"), 1)) > 1:
|
|
params.update({"direction_2": _pattern_direction(props.get("D2Axis")), "spacing_2_mm": _mm(props.get("D2Spacing")), "pattern_count_2": int(_number(props.get("D2TotalInstances"), 1))})
|
|
missing = [] if source_feature_ids and params["spacing_1_mm"] > 0 else ["linear pattern source features or primary spacing were not captured"]
|
|
return params, missing
|
|
if family == "MirrorPattern":
|
|
property_sources = [
|
|
_referenced_feature_id(item, feature_id_by_source, feature_id_by_name)
|
|
for item in props.get("PatternFeatureArray") or [] if isinstance(item, dict)
|
|
]
|
|
source_feature_ids = list(dict.fromkeys([*source_feature_ids, *[item for item in property_sources if item]]))
|
|
plane_reference = props.get("Plane")
|
|
plane = (
|
|
_selector(
|
|
plane_reference,
|
|
kind_hint="plane",
|
|
owner_feature_id=_referenced_feature_id(plane_reference, feature_id_by_source, feature_id_by_name),
|
|
)
|
|
if isinstance(plane_reference, dict) else None
|
|
)
|
|
if plane is None:
|
|
plane = next(
|
|
(selector for selector in (_selector(item, kind_hint="plane") for item in _record_selections(feature)) if selector),
|
|
None,
|
|
)
|
|
return {"source_feature_ids": source_feature_ids, "mirror_plane": plane or {"unresolved": "mirror plane was not captured"}}, ([] if source_feature_ids and plane else ["mirror pattern source features or plane were not captured"])
|
|
if family == "RefPlane":
|
|
references: list[dict[str, Any]] = []
|
|
reference_items = [*_record_parents(feature), *[item for item in props.get("Selections") or [] if isinstance(item, dict)]]
|
|
for reference in reference_items:
|
|
selector = _selector(
|
|
reference,
|
|
kind_hint="plane" if _is_named_coordinate_plane(reference.get("name")) else None,
|
|
owner_feature_id=_referenced_feature_id(reference, feature_id_by_source, feature_id_by_name),
|
|
)
|
|
if selector and selector not in references:
|
|
references.append(selector)
|
|
base_name = next((str(item.get("name")) for item in _record_parents(feature) if _is_named_coordinate_plane(item.get("name"))), None)
|
|
if base_name:
|
|
plane = _named_plane(base_name)
|
|
elif _is_named_coordinate_plane(feature.get("name")):
|
|
plane = _named_plane(feature.get("name"))
|
|
elif (plane_reference := next((item for item in reference_items if _workplane_from_plane_reference(item)), None)):
|
|
plane = _workplane_from_plane_reference(plane_reference) or {}
|
|
else:
|
|
plane = {"unresolved": "reference plane orientation was not captured"}
|
|
if "unresolved" not in plane:
|
|
plane.pop("name", None)
|
|
offset = _mm(props.get("Distance"))
|
|
if offset:
|
|
sign = -1.0 if bool(props.get("ReverseDirection")) else 1.0
|
|
plane["origin_mm"] = [round(sign * offset * value, 9) for value in plane["normal"]]
|
|
params: dict[str, Any] = {
|
|
"plane": plane,
|
|
"offset_mm": _mm(props.get("Distance")),
|
|
"angle_rad": _number(props.get("Angle")),
|
|
"reverse": bool(props.get("ReverseDirection")),
|
|
"solidworks_type": int(_number(props.get("Type"), -1)),
|
|
}
|
|
if references:
|
|
params["references"] = references
|
|
missing = ["reference plane orientation was not captured"] if "unresolved" in plane else []
|
|
return params, missing
|
|
if family == "RefAxis":
|
|
axis, _ = _axis_from_reference(props.get("Axis"))
|
|
if axis is None:
|
|
for reference in [*_captured_selection_values(methods), *_record_selections(feature)]:
|
|
axis, _ = _axis_from_reference(reference)
|
|
if axis is not None:
|
|
break
|
|
if axis is None:
|
|
parent_planes = [
|
|
_named_plane(item.get("name"))
|
|
for item in _record_parents(feature)
|
|
if _is_named_coordinate_plane(item.get("name"))
|
|
]
|
|
if len(parent_planes) == 2:
|
|
direction = _cross(parent_planes[0]["normal"], parent_planes[1]["normal"])
|
|
if math.sqrt(sum(component * component for component in direction)) > 1e-12:
|
|
axis = {"origin_mm": [0.0, 0.0, 0.0], "direction": _normalize(direction)}
|
|
return {"axis": axis or {"unresolved": "reference axis was not captured"}}, ([] if axis else ["reference axis was not captured"])
|
|
return {}, []
|
|
|
|
|
|
def _pattern_direction(value: Any) -> list[float]:
|
|
if isinstance(value, dict):
|
|
geometry = value.get("geometry") if isinstance(value.get("geometry"), dict) else value
|
|
start = _point3(geometry.get("start"))
|
|
end = _point3(geometry.get("end"))
|
|
if start and end:
|
|
return _normalize([end[index] - start[index] for index in range(3)])
|
|
vector = geometry.get("vector")
|
|
if isinstance(vector, list) and len(vector) >= 3:
|
|
return _normalize([_number(item) for item in vector[0:3]])
|
|
return [1.0, 0.0, 0.0]
|
|
|
|
|
|
def _named_plane(name: Any) -> dict[str, Any]:
|
|
text = str(name or "").lower()
|
|
# Named coordinate planes appear in both SolidWorks' localized origin and
|
|
# as internal "ip_N XY/XZ/YZ" references. Check the pair names first so
|
|
# an axis such as "ip_1 X" is never mistaken for an arbitrary plane.
|
|
if re.search(r"(?:^|\s)xy(?:$|\s)", text):
|
|
normal, x_dir = [0.0, 0.0, 1.0], [1.0, 0.0, 0.0]
|
|
elif re.search(r"(?:^|\s)xz(?:$|\s)", text):
|
|
normal, x_dir = [0.0, 1.0, 0.0], [1.0, 0.0, 0.0]
|
|
elif re.search(r"(?:^|\s)yz(?:$|\s)", text):
|
|
normal, x_dir = [1.0, 0.0, 0.0], [0.0, 1.0, 0.0]
|
|
elif "上" in text or "top" in text:
|
|
normal, x_dir = [0.0, 0.0, 1.0], [1.0, 0.0, 0.0]
|
|
elif "右" in text or "right" in text:
|
|
normal, x_dir = [1.0, 0.0, 0.0], [0.0, 1.0, 0.0]
|
|
elif "前" in text or "front" in text:
|
|
normal, x_dir = [0.0, 1.0, 0.0], [1.0, 0.0, 0.0]
|
|
else:
|
|
normal, x_dir = [0.0, 0.0, 1.0], [1.0, 0.0, 0.0]
|
|
return {"origin_mm": [0.0, 0.0, 0.0], "normal": normal, "x_dir": x_dir, "name": str(name or "reference plane")}
|
|
|
|
|
|
def _is_named_coordinate_plane(name: Any) -> bool:
|
|
text = str(name or "").lower()
|
|
return bool(
|
|
re.search(r"(?:^|\s)(?:xy|xz|yz)(?:$|\s)", text)
|
|
or any(token in text for token in ("top", "front", "right", "上", "前", "右"))
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class StepInspector:
|
|
path: Path | None
|
|
faces: list[dict[str, Any]] = field(default_factory=list)
|
|
axes: list[dict[str, Any]] = field(default_factory=list)
|
|
repeat_spacings_mm: list[dict[str, Any]] = field(default_factory=list)
|
|
metrics: dict[str, Any] = field(default_factory=dict)
|
|
error: str | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.path or not self.path.exists():
|
|
return
|
|
try:
|
|
from build123d import import_step
|
|
|
|
solid = import_step(str(self.path))
|
|
for index, face in enumerate(solid.faces()):
|
|
box = face.bounding_box()
|
|
center = face.center()
|
|
normal = face.normal_at()
|
|
values = [box.min.X, box.min.Y, box.min.Z, box.max.X, box.max.Y, box.max.Z]
|
|
surface_type = str(getattr(face, "geom_type", "unknown")).rsplit(".", 1)[-1].lower()
|
|
geometry: dict[str, Any] = {
|
|
"bbox_mm": [round(value, 9) for value in values],
|
|
"center_mm": [round(center.X, 9), round(center.Y, 9), round(center.Z, 9)],
|
|
"normal": [round(normal.X, 9), round(normal.Y, 9), round(normal.Z, 9)],
|
|
"surface_type": surface_type,
|
|
}
|
|
axis = getattr(face, "axis_of_rotation", None)
|
|
if surface_type in {"cylinder", "cone"} and axis is not None:
|
|
axis_record = {
|
|
"origin_mm": [round(axis.position.X, 9), round(axis.position.Y, 9), round(axis.position.Z, 9)],
|
|
"direction": [round(axis.direction.X, 9), round(axis.direction.Y, 9), round(axis.direction.Z, 9)],
|
|
"surface_type": surface_type,
|
|
}
|
|
radius = getattr(face, "radius", None)
|
|
if isinstance(radius, (int, float)) and math.isfinite(radius):
|
|
axis_record["radius_mm"] = round(radius, 9)
|
|
geometry["axis"] = axis_record
|
|
stable_payload = json.dumps({"index": index, "geometry": geometry}, sort_keys=True, separators=(",", ":"))
|
|
stable = hashlib.sha256(stable_payload.encode()).hexdigest()
|
|
record = {
|
|
"kind": "face", "stable_id": f"step-face-{stable[:24]}", "source": "inferred_from_step", "confidence": 0.0,
|
|
"geometry": geometry,
|
|
}
|
|
self.faces.append(record)
|
|
if "axis" in geometry:
|
|
self.axes.append({"stable_id": record["stable_id"], **geometry["axis"]})
|
|
self.repeat_spacings_mm = self._repeated_axis_spacings()
|
|
overall_box = solid.bounding_box()
|
|
self.metrics = {
|
|
"bounding_box_mm": [
|
|
round(value, 9)
|
|
for value in (overall_box.min.X, overall_box.min.Y, overall_box.min.Z, overall_box.max.X, overall_box.max.Y, overall_box.max.Z)
|
|
],
|
|
"volume_mm3": round(float(solid.volume), 9),
|
|
"surface_area_mm2": round(float(solid.area), 9),
|
|
"solid_count": len(solid.solids()),
|
|
"face_count": len(solid.faces()),
|
|
"edge_count": len(solid.edges()),
|
|
"vertex_count": len(solid.vertices()),
|
|
}
|
|
except Exception as error: # STEP inference is optional and must not abort conversion.
|
|
self.error = str(error)
|
|
|
|
def _repeated_axis_spacings(self) -> list[dict[str, Any]]:
|
|
"""Report only repeatable topology distances, never a guessed pattern."""
|
|
origins = list(dict.fromkeys(tuple(axis["origin_mm"]) for axis in self.axes))[:160]
|
|
distances: Counter[float] = Counter()
|
|
for index, first in enumerate(origins):
|
|
for second in origins[index + 1:]:
|
|
distance = math.dist(first, second)
|
|
if distance > EPSILON_MM:
|
|
distances[round(distance, 4)] += 1
|
|
return [
|
|
{"spacing_mm": distance, "pair_count": count}
|
|
for distance, count in distances.most_common(12) if count > 1
|
|
]
|
|
|
|
def infer_host_face(self, positions: list[list[float]]) -> dict[str, Any] | None:
|
|
if not positions or not self.faces:
|
|
return None
|
|
candidates = []
|
|
for face in self.faces:
|
|
box = face["geometry"]["bbox_mm"]
|
|
geometry = face["geometry"]
|
|
if not all(box[0] - 0.05 <= point[0] <= box[3] + 0.05 and box[1] - 0.05 <= point[1] <= box[4] + 0.05 and box[2] - 0.05 <= point[2] <= box[5] + 0.05 for point in positions):
|
|
continue
|
|
if geometry["surface_type"] == "plane":
|
|
normal = geometry["normal"]
|
|
center = geometry["center_mm"]
|
|
if not all(abs(sum((point[index] - center[index]) * normal[index] for index in range(3))) <= 0.05 for point in positions):
|
|
continue
|
|
elif geometry["surface_type"] == "cylinder" and isinstance(geometry.get("axis"), dict):
|
|
axis = geometry["axis"]
|
|
radius = axis.get("radius_mm")
|
|
if not isinstance(radius, (int, float)):
|
|
continue
|
|
origin, direction = axis["origin_mm"], axis["direction"]
|
|
if not all(abs(math.dist(point, [origin[index] + sum((point[j] - origin[j]) * direction[j] for j in range(3)) * direction[index] for index in range(3)]) - radius) <= 0.05 for point in positions):
|
|
continue
|
|
else:
|
|
continue
|
|
candidates.append(face)
|
|
if len(candidates) != 1:
|
|
return None
|
|
candidate = dict(candidates[0])
|
|
candidate["confidence"] = 0.8
|
|
return candidate
|
|
|
|
def compare_truth(self, truth: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""Compare the STEP result against evidence's documented model truth.
|
|
|
|
Metrics are diagnostic evidence only. STEP topology can be split or
|
|
healed by the importer, so exact topology counts are reported but do
|
|
not turn an otherwise sound geometric match into a conversion failure.
|
|
"""
|
|
if not self.metrics or not isinstance(truth, dict):
|
|
return None
|
|
geometry = truth.get("geometry") if isinstance(truth.get("geometry"), dict) else {}
|
|
mass = truth.get("mass_properties") if isinstance(truth.get("mass_properties"), dict) else {}
|
|
expected_bbox = geometry.get("bounding_box")
|
|
expected: dict[str, Any] = {}
|
|
if isinstance(expected_bbox, list) and len(expected_bbox) == 6:
|
|
expected["bounding_box_mm"] = [_mm(value) for value in expected_bbox]
|
|
if _number(mass.get("volume")) > 0:
|
|
expected["volume_mm3"] = round(_number(mass["volume"]) * METERS_TO_MM ** 3, 9)
|
|
if _number(mass.get("surface_area")) > 0:
|
|
expected["surface_area_mm2"] = round(_number(mass["surface_area"]) * METERS_TO_MM ** 2, 9)
|
|
bodies = geometry.get("bodies") if isinstance(geometry.get("bodies"), list) else []
|
|
if bodies:
|
|
body_metrics = [item.get("geometry") for item in bodies if isinstance(item, dict) and isinstance(item.get("geometry"), dict)]
|
|
if body_metrics:
|
|
expected.update({
|
|
"solid_count": int(_number(geometry.get("solid_body_count"), len(body_metrics))),
|
|
"face_count": sum(int(_number(item.get("face_count"))) for item in body_metrics),
|
|
"edge_count": sum(int(_number(item.get("edge_count"))) for item in body_metrics),
|
|
"vertex_count": sum(int(_number(item.get("vertex_count"))) for item in body_metrics),
|
|
})
|
|
comparisons: dict[str, Any] = {}
|
|
for key in ("volume_mm3", "surface_area_mm2"):
|
|
if key in expected:
|
|
actual = self.metrics[key]
|
|
delta = abs(actual - expected[key])
|
|
comparisons[key] = {"expected": expected[key], "actual": actual, "absolute_delta": delta, "within_tolerance": delta <= max(1e-4, abs(expected[key]) * 1e-4)}
|
|
if "bounding_box_mm" in expected:
|
|
deltas = [abs(actual - target) for actual, target in zip(self.metrics["bounding_box_mm"], expected["bounding_box_mm"])]
|
|
comparisons["bounding_box_mm"] = {"expected": expected["bounding_box_mm"], "actual": self.metrics["bounding_box_mm"], "max_absolute_delta": max(deltas), "within_tolerance": max(deltas) <= 1e-3}
|
|
for key in ("solid_count", "face_count", "edge_count", "vertex_count"):
|
|
if key in expected:
|
|
comparisons[key] = {"expected": expected[key], "actual": self.metrics[key], "matches": expected[key] == self.metrics[key]}
|
|
numeric = [value["within_tolerance"] for value in comparisons.values() if "within_tolerance" in value]
|
|
return {
|
|
"expected_from_evidence": expected,
|
|
"comparisons": comparisons,
|
|
"numeric_geometry_match": all(numeric) if numeric else None,
|
|
"topology_counts_match": all(value["matches"] for value in comparisons.values() if "matches" in value),
|
|
}
|
|
|
|
def summary(self, truth: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
surface_counts = Counter(str(face["geometry"].get("surface_type") or "unknown") for face in self.faces)
|
|
result = {
|
|
"available": bool(self.path and self.path.exists()), "face_count": len(self.faces),
|
|
"surface_counts": dict(sorted(surface_counts.items())), "axis_count": len(self.axes),
|
|
"repeat_spacings_mm": self.repeat_spacings_mm, "metrics": self.metrics, "error": self.error,
|
|
}
|
|
if comparison := self.compare_truth(truth or {}):
|
|
result["truth_comparison"] = comparison
|
|
return result
|
|
|
|
|
|
def _blockers_by_feature(evidence: dict[str, Any]) -> dict[str, list[str]]:
|
|
result: dict[str, list[str]] = {}
|
|
for contract in (evidence.get("self_validation") or {}).get("feature_contracts") or []:
|
|
if isinstance(contract, dict) and contract.get("blockers"):
|
|
result[str(contract.get("feature") or "")] = [str(item) for item in contract["blockers"]]
|
|
return result
|
|
|
|
|
|
def _topologically_order_features(features: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Order source features by their explicit CDSL dependencies.
|
|
|
|
SolidWorks' tree order can place a reference object after its consumer. A
|
|
CDSL dependency must be emitted first, so normal forward references are
|
|
sorted here. A genuine cycle cannot be executed by a future engine either;
|
|
retain the feature and surface the unsupported edge as an unresolved item.
|
|
"""
|
|
remaining = list(features)
|
|
ordered: list[dict[str, Any]] = []
|
|
emitted: set[str] = set()
|
|
while remaining:
|
|
ready = [feature for feature in remaining if set(feature.get("depends_on") or []).issubset(emitted)]
|
|
if ready:
|
|
for feature in ready:
|
|
ordered.append(feature)
|
|
emitted.add(feature["id"])
|
|
remaining.remove(feature)
|
|
continue
|
|
feature = remaining.pop(0)
|
|
blocked = [dependency for dependency in feature.get("depends_on") or [] if dependency not in emitted]
|
|
feature["depends_on"] = [dependency for dependency in feature.get("depends_on") or [] if dependency in emitted]
|
|
feature.setdefault("unresolved", []).append(
|
|
"cyclic or unresolved feature dependency: " + ", ".join(blocked)
|
|
)
|
|
ordered.append(feature)
|
|
emitted.add(feature["id"])
|
|
return ordered
|
|
|
|
|
|
def _part_id_from_source_name(source_name: str) -> str:
|
|
"""Build a schema-valid ID from the entire evidence file name."""
|
|
stem = Path(source_name).name.removesuffix(EVIDENCE_V2_SUFFIX)
|
|
part_id = re.sub(r"[^A-Za-z0-9_-]+", "-", stem).strip("-")
|
|
if len(part_id) < 3:
|
|
part_id = f"part-{part_id}".rstrip("-")
|
|
return part_id[:80]
|
|
|
|
|
|
def convert_evidence(evidence: dict[str, Any], *, source_name: str, step_path: Path | None = None,
|
|
part_id: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
if evidence.get("schema") != "solidworks.cad_evidence.v2":
|
|
raise ValueError("Expected schema solidworks.cad_evidence.v2")
|
|
features = sorted((item for item in evidence.get("features") or [] if isinstance(item, dict)), key=lambda item: int(_number(item.get("sequence"))))
|
|
source_by_stable = {_identity_key(feature): feature for feature in features if _identity_key(feature)}
|
|
source_by_name = {str(feature.get("name")): feature for feature in features if feature.get("name") is not None}
|
|
sketch_features = [feature for feature in features if str(feature.get("effective_type") or "") in SKETCH_TYPES and isinstance(feature.get("sketch"), dict)]
|
|
sketch_id_by_source = {_identity_key(feature): _safe_id("sk", index + 1) for index, feature in enumerate(sketch_features) if _identity_key(feature)}
|
|
sketch_id_by_name = {str(feature.get("name")): sketch_id_by_source[source_id] for feature in sketch_features if (source_id := _identity_key(feature)) and feature.get("name") is not None}
|
|
sketch_workplanes_by_parent_source: dict[str, tuple[str, dict[str, Any]]] = {}
|
|
sketch_workplanes_by_parent_name: dict[str, tuple[str, dict[str, Any]]] = {}
|
|
# sketch source id -> the parent reference-plane source id it is drawn on
|
|
sketch_parent_plane_by_source: dict[str, str] = {}
|
|
# sketch source id -> the world-space direction of its revolve axis
|
|
sketch_axis_direction_by_source: dict[str, list[float]] = {}
|
|
# sketch source ids whose exporter matrix did not form an orthonormal frame
|
|
sketch_matrix_unreliable: set[str] = set()
|
|
sketch_record_by_id: dict[str, dict[str, Any]] = {}
|
|
# Prescan the revolve features so each profile's workplane can be rebuilt
|
|
# from the axis direction captured on its own revolve feature.
|
|
for feature in features:
|
|
if _feature_family(feature) not in {"Revolution", "RevCut"}:
|
|
continue
|
|
props, _ = _record_values(feature)
|
|
axis, _ = _axis_from_reference(props.get("Axis"))
|
|
if not axis or not axis.get("direction"):
|
|
continue
|
|
for parent in _record_parents(feature):
|
|
if str(parent.get("effective_type") or "") in SKETCH_TYPES and (sketch_source_id := _identity_key(parent)):
|
|
sketch_axis_direction_by_source.setdefault(sketch_source_id, list(axis["direction"]))
|
|
sketches = []
|
|
for feature in sketch_features:
|
|
source_id = _identity_key(feature)
|
|
if not source_id:
|
|
continue
|
|
sketch = feature["sketch"]
|
|
workplane = _workplane(sketch)
|
|
record = {"id": sketch_id_by_source[source_id], "name": str(feature.get("name") or sketch_id_by_source[source_id]), "role": "profile", "workplane": workplane, "profile": _analytic_profile(sketch)}
|
|
sketches.append(record)
|
|
sketch_record_by_id[record["id"]] = record
|
|
if not _workplane_is_orthonormal(workplane):
|
|
sketch_matrix_unreliable.add(source_id)
|
|
for parent in _record_parents(feature):
|
|
parent_source_id = _identity_key(parent)
|
|
if parent_source_id:
|
|
sketch_workplanes_by_parent_source.setdefault(parent_source_id, (record["id"], workplane))
|
|
if _feature_family(parent) == "RefPlane":
|
|
sketch_parent_plane_by_source[source_id] = parent_source_id
|
|
if parent.get("name") is not None:
|
|
sketch_workplanes_by_parent_name.setdefault(str(parent["name"]), (record["id"], workplane))
|
|
|
|
model_features = [feature for feature in features if _feature_family(feature)]
|
|
feature_id_by_source = {_identity_key(feature): _safe_id("f", index + 1) for index, feature in enumerate(model_features) if _identity_key(feature)}
|
|
feature_id_by_name = {str(feature.get("name")): feature_id_by_source[source_id] for feature in model_features if (source_id := _identity_key(feature)) and feature.get("name") is not None}
|
|
source_feature_ids = {
|
|
feature_id_by_source[source_id]
|
|
for feature in model_features
|
|
if (source_id := _identity_key(feature)) and _feature_family(feature) not in {"RefPlane", "RefAxis"}
|
|
}
|
|
# Type 10 is SolidWorks' document-origin reference plane. Some localized
|
|
# exports lose the front/top/right names but retain their fixed tree order.
|
|
origin_plane_by_source: dict[str, dict[str, Any]] = {}
|
|
origin_plane_by_name: dict[str, dict[str, Any]] = {}
|
|
origin_names = ("front", "top", "right")
|
|
origin_candidates = []
|
|
for item in model_features:
|
|
if _feature_family(item) != "RefPlane" or _record_parents(item):
|
|
continue
|
|
item_props, _ = _record_values(item)
|
|
if int(_number(item_props.get("Type"), -1)) == 10:
|
|
origin_candidates.append(item)
|
|
for index, item in enumerate(origin_candidates[:3]):
|
|
plane = _named_plane(origin_names[index])
|
|
plane.pop("name", None)
|
|
if (item_id := _identity_key(item)):
|
|
origin_plane_by_source[item_id] = plane
|
|
if item.get("name") is not None:
|
|
origin_plane_by_name[str(item["name"])] = plane
|
|
blockers = _blockers_by_feature(evidence)
|
|
# Conversion can discover a missing selector even when the exporter did
|
|
# not classify it as a self-validation blocker, so inspect supplied STEP
|
|
# truth for every record. The inspector remains a no-op when no file was
|
|
# requested or found.
|
|
step = StepInspector(step_path)
|
|
cdsl_features: list[dict[str, Any]] = []
|
|
previous_id: str | None = None
|
|
diagnostics: list[dict[str, Any]] = []
|
|
# Sketch source ids whose workplane was successfully rebuilt once their
|
|
# parent reference plane was resolved. Their revolve axis is expressed in
|
|
# sketch-local coordinates and must be transformed to world space too.
|
|
sketch_fix_applied: set[str] = set()
|
|
sketch_source_by_id = {sketch_id: source_id for source_id, sketch_id in sketch_id_by_source.items()}
|
|
|
|
for feature in model_features:
|
|
source_id = _identity_key(feature)
|
|
if not source_id:
|
|
continue
|
|
feature_id = feature_id_by_source[source_id]
|
|
family = _feature_family(feature)
|
|
props, methods = _record_values(feature)
|
|
sketch_id = _parent_sketch_id(feature, sketch_id_by_source, sketch_id_by_name)
|
|
dependencies = _parent_feature_ids(feature, feature_id_by_source, feature_id_by_name, previous_id)
|
|
source_feature_ids_for_current: list[str] = []
|
|
for parent in _record_parents(feature):
|
|
source_feature_id = feature_id_by_source.get(_identity_key(parent) or "") or feature_id_by_name.get(str(parent.get("name") or ""))
|
|
if source_feature_id and source_feature_id in source_feature_ids and source_feature_id not in source_feature_ids_for_current:
|
|
source_feature_ids_for_current.append(source_feature_id)
|
|
unresolved = list(blockers.get(str(feature.get("name") or ""), []))
|
|
selectors = [item for item in (_selector(value, owner_feature_id=feature_id_by_source.get(_identity_key(value) or "")) for value in _record_selections(feature)) if item]
|
|
|
|
if family in {"Boss", "Extrusion", "Cut", "Revolution", "RevCut"} and not sketch_id:
|
|
unresolved.append("missing source sketch parent")
|
|
|
|
if family in {"Boss", "Extrusion", "Cut"}:
|
|
atomic_id = "extrude_cut_blind" if family == "Cut" else ("extrude_add_two_sided" if bool(props.get("BothDirections")) else "extrude_add_blind")
|
|
params = _extrude_params(props, methods)
|
|
# A through-all, up-to-surface, up-to-vertex, or up-to-body
|
|
# extrusion intentionally has no blind depth. Its termination is
|
|
# represented by end_condition, so reporting a zero distance as a
|
|
# missing capture would make valid SolidWorks evidence look broken.
|
|
if params["distance_mm"] <= 0 and params["end_condition"]["type"] in {"blind", "offset_from_surface", "mid_plane", "through_all_and_blind"}:
|
|
unresolved.append("extrude depth was not captured")
|
|
elif family in {"Revolution", "RevCut"}:
|
|
atomic_id = "revolve_cut" if family == "RevCut" else "revolve_add"
|
|
params, more_unresolved = _revolve_params(props, methods)
|
|
unresolved.extend(more_unresolved)
|
|
if params["angle_deg"] <= 0:
|
|
unresolved.append("revolve angle was not captured")
|
|
axis_selector = params.pop("axis_selector", None)
|
|
if axis_selector:
|
|
selectors.append(axis_selector)
|
|
if props.get("Axis") and any("no captured semantic selections" in item for item in unresolved):
|
|
unresolved = [item for item in unresolved if "no captured semantic selections" not in item]
|
|
# When the profile workplane was rebuilt, the captured revolve axis
|
|
# is still in sketch-local coordinates; map it to world space so
|
|
# the cut lands on the resolved parent reference plane.
|
|
if sketch_id and (axis_source_id := sketch_source_by_id.get(sketch_id)) in sketch_fix_applied:
|
|
sketch_workplane = sketch_record_by_id.get(sketch_id)
|
|
axis = params.get("axis")
|
|
if isinstance(axis, dict) and sketch_workplane is not None:
|
|
params["axis"] = _axis_in_world(axis, sketch_workplane["workplane"])
|
|
elif family == "HoleWzd":
|
|
atomic_id = "hole_wizard"
|
|
params, positions = _hole_params(props, methods)
|
|
if params["diameter_mm"] <= 0 or params["depth_mm"] <= 0:
|
|
unresolved.append("hole diameter or depth was not captured")
|
|
host_face = _selector(props.get("Face"))
|
|
if not host_face:
|
|
host_face = step.infer_host_face(positions)
|
|
if host_face:
|
|
params["host_face"] = host_face
|
|
selectors.append(host_face)
|
|
unresolved = [item for item in unresolved if "hole has no captured semantic selections" not in item]
|
|
elif family == "Fillet":
|
|
atomic_id = "fillet"
|
|
params, more_unresolved = _deferred_params(family, feature, props, methods, source_feature_ids_for_current, feature_id_by_source, feature_id_by_name)
|
|
unresolved.extend(more_unresolved)
|
|
elif family == "Chamfer":
|
|
atomic_id = "chamfer"
|
|
params, more_unresolved = _deferred_params(family, feature, props, methods, source_feature_ids_for_current, feature_id_by_source, feature_id_by_name)
|
|
unresolved.extend(more_unresolved)
|
|
elif family == "LPattern":
|
|
atomic_id = "pattern_linear"
|
|
params, more_unresolved = _deferred_params(family, feature, props, methods, source_feature_ids_for_current, feature_id_by_source, feature_id_by_name)
|
|
unresolved.extend(more_unresolved)
|
|
dependencies = list(dict.fromkeys([*dependencies, *params["source_feature_ids"]]))
|
|
if not more_unresolved:
|
|
unresolved = [item for item in unresolved if "pattern has no captured semantic selections" not in item]
|
|
elif family == "MirrorPattern":
|
|
atomic_id = "pattern_mirror"
|
|
params, more_unresolved = _deferred_params(family, feature, props, methods, source_feature_ids_for_current, feature_id_by_source, feature_id_by_name)
|
|
unresolved.extend(more_unresolved)
|
|
dependencies = list(dict.fromkeys([*dependencies, *params["source_feature_ids"]]))
|
|
mirror_plane = params.get("mirror_plane")
|
|
if isinstance(mirror_plane, dict) and mirror_plane.get("kind"):
|
|
selectors.append(mirror_plane)
|
|
if mirror_plane.get("owner_feature_id"):
|
|
dependencies = list(dict.fromkeys([*dependencies, mirror_plane["owner_feature_id"]]))
|
|
if not more_unresolved:
|
|
unresolved = [item for item in unresolved if "pattern has no captured semantic selections" not in item]
|
|
elif family == "RefPlane":
|
|
atomic_id = "reference_plane"
|
|
params, more_unresolved = _deferred_params(family, feature, props, methods, source_feature_ids_for_current, feature_id_by_source, feature_id_by_name)
|
|
unresolved.extend(more_unresolved)
|
|
if isinstance(params.get("plane"), dict) and "unresolved" in params["plane"]:
|
|
inferred = (
|
|
sketch_workplanes_by_parent_source.get(source_id)
|
|
or sketch_workplanes_by_parent_name.get(str(feature.get("name") or ""))
|
|
)
|
|
source = "model_to_sketch_transform"
|
|
if inferred:
|
|
sketch_source_id, plane = inferred
|
|
params["plane"] = plane
|
|
params["derived_from_sketch_id"] = sketch_source_id
|
|
else:
|
|
plane = origin_plane_by_source.get(source_id) or origin_plane_by_name.get(str(feature.get("name") or ""))
|
|
source = "solidworks_origin_plane_order"
|
|
if plane:
|
|
params["plane"] = plane
|
|
params["plane_inference"] = source
|
|
if "unresolved" in params["plane"]:
|
|
for parent in _record_parents(feature):
|
|
parent_id = _referenced_feature_id(parent, feature_id_by_source, feature_id_by_name)
|
|
parent_feature = next((item for item in cdsl_features if item["id"] == parent_id), None)
|
|
parent_plane = (parent_feature or {}).get("params", {}).get("plane")
|
|
if not isinstance(parent_plane, dict) or "origin_mm" not in parent_plane:
|
|
continue
|
|
offset = params["offset_mm"] * (-1.0 if params["reverse"] else 1.0)
|
|
params["plane"] = {
|
|
**parent_plane,
|
|
"origin_mm": [
|
|
round(parent_plane["origin_mm"][index] + offset * parent_plane["normal"][index], 9)
|
|
for index in range(3)
|
|
],
|
|
}
|
|
params["derived_from_feature_id"] = parent_id
|
|
params["plane_inference"] = "parent_reference_plane"
|
|
break
|
|
if "unresolved" not in params["plane"]:
|
|
more_unresolved = [item for item in more_unresolved if item != "reference plane orientation was not captured"]
|
|
unresolved = [item for item in unresolved if item != "reference plane orientation was not captured"]
|
|
plane = params.get("plane")
|
|
if isinstance(plane, dict) and "origin_mm" in plane and "unresolved" not in plane:
|
|
# Profiles whose exporter matrix was unusable are rebuilt from
|
|
# this resolved reference plane (normal + origin) and the axis
|
|
# direction of the profile's own revolve feature.
|
|
for sketch_source_id, ref_source_id in sketch_parent_plane_by_source.items():
|
|
if ref_source_id != source_id or sketch_source_id not in sketch_matrix_unreliable:
|
|
continue
|
|
rebuilt = _workplane_rebuilt(plane, sketch_axis_direction_by_source.get(sketch_source_id))
|
|
if rebuilt is None:
|
|
continue
|
|
sketch_record = sketch_record_by_id.get(sketch_id_by_source.get(sketch_source_id) or "")
|
|
if sketch_record is not None:
|
|
sketch_record["workplane"] = rebuilt
|
|
sketch_fix_applied.add(sketch_source_id)
|
|
for reference in params.get("references", []):
|
|
if reference not in selectors:
|
|
selectors.append(reference)
|
|
elif family == "RefAxis":
|
|
atomic_id = "reference_axis"
|
|
params, more_unresolved = _deferred_params(family, feature, props, methods, source_feature_ids_for_current, feature_id_by_source, feature_id_by_name)
|
|
unresolved.extend(more_unresolved)
|
|
if "unresolved" not in params["axis"]:
|
|
axis_selector = params["axis"].get("selector")
|
|
if isinstance(axis_selector, dict):
|
|
selectors.append(axis_selector)
|
|
for parent in _record_parents(feature):
|
|
parent_id = _referenced_feature_id(parent, feature_id_by_source, feature_id_by_name)
|
|
source_parent = source_by_stable.get(_identity_key(parent) or "") or source_by_name.get(str(parent.get("name") or ""))
|
|
is_plane = bool(source_parent and _feature_family(source_parent) == "RefPlane") or _is_named_coordinate_plane(parent.get("name"))
|
|
selector = _selector(parent, kind_hint="plane", owner_feature_id=parent_id) if is_plane else None
|
|
if selector and selector not in selectors:
|
|
selectors.append(selector)
|
|
else:
|
|
continue
|
|
|
|
execution_status = "supported" if atomic_id in SUPPORTED_ATOMS and sketch_id and not unresolved and sketches and next((item for item in sketches if item["id"] == sketch_id), {}).get("profile", {}).get("type") != "analytic_contours" else "deferred"
|
|
output: dict[str, Any] = {
|
|
"id": feature_id, "name": str(feature.get("name") or feature_id), "atomic_id": atomic_id,
|
|
"depends_on": dependencies, "params": params, "execution_status": execution_status,
|
|
}
|
|
if sketch_id:
|
|
output["sketch_id"] = sketch_id
|
|
if selectors:
|
|
output["selectors"] = selectors
|
|
if unresolved:
|
|
output["unresolved"] = sorted(set(unresolved))
|
|
cdsl_features.append(output)
|
|
diagnostics.append({"feature_id": feature_id, "source_name": output["name"], "atomic_id": atomic_id, "execution_status": execution_status, "unresolved": output.get("unresolved", []), "step_inferred_selector_count": sum(1 for selector in selectors if selector["source"] == "inferred_from_step")})
|
|
previous_id = feature_id
|
|
|
|
if not sketches:
|
|
# The v1 envelope requires one sketch. This explicit empty reference sketch lets
|
|
# reference-only evidence be represented without fabricating a profile.
|
|
default_workplane = _named_plane("default")
|
|
default_workplane.pop("name", None)
|
|
sketches.append({"id": "sk_001", "name": "missing-source-sketch", "role": "reference", "workplane": default_workplane, "profile": {"type": "analytic_contours", "contours": []}})
|
|
if not cdsl_features:
|
|
origin_plane = _named_plane("document origin")
|
|
origin_plane.pop("name", None)
|
|
cdsl_features.append({
|
|
"id": "f_001", "name": "document-origin", "atomic_id": "reference_plane", "depends_on": [],
|
|
"params": {"plane": origin_plane}, "execution_status": "deferred",
|
|
})
|
|
cdsl_features = _topologically_order_features(cdsl_features)
|
|
|
|
part_id = part_id or _part_id_from_source_name(source_name)
|
|
truth = evidence.get("document_truth") or {}
|
|
step_summary = step.summary(truth)
|
|
cdsl = {
|
|
"schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", "part_id": part_id,
|
|
"geometry": {"sketches": sketches}, "features": cdsl_features,
|
|
"meta": {
|
|
"unit": "mm", "source": source_name, "source_schema": "solidworks.cad_evidence.v2",
|
|
"source_status": evidence.get("status"), "document_truth": {"mass_properties": truth.get("mass_properties"), "geometry": truth.get("geometry")},
|
|
"step_inference": step_summary, "source_has_modeling_features": bool(model_features),
|
|
},
|
|
}
|
|
validation = validate_semantic_cdsl(cdsl)
|
|
diagnostic = {
|
|
"source": source_name, "part_id": part_id, "source_status": evidence.get("status"),
|
|
"source_self_validation": evidence.get("self_validation"), "step_inference": step_summary,
|
|
"features": diagnostics, "semantic_validation": validation,
|
|
}
|
|
if not cdsl_features:
|
|
diagnostic["note"] = "Evidence contains no modeling features; emitted a valid empty semantic CDSL document."
|
|
return cdsl, diagnostic
|
|
|
|
|
|
def _write_json(path: Path, value: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def _convert_path(source: Path, output_dir: Path, truth_dir: Path | None, overwrite: bool,
|
|
source_name: str, truth_relative_path: Path, part_id: str) -> dict[str, Any]:
|
|
cdsl_path = output_dir / f"{part_id}.cdsl.json"
|
|
diagnostic_path = output_dir / f"{part_id}.diagnostic.json"
|
|
if not overwrite and cdsl_path.exists() and diagnostic_path.exists():
|
|
return {
|
|
"part_id": part_id, "source": source_name, "status": "skipped",
|
|
"cdsl": cdsl_path.name, "diagnostic": diagnostic_path.name,
|
|
}
|
|
try:
|
|
evidence = json.loads(source.read_text(encoding="utf-8"))
|
|
step_path = (truth_dir / truth_relative_path) if truth_dir else None
|
|
cdsl, diagnostic = convert_evidence(evidence, source_name=source_name, step_path=step_path, part_id=part_id)
|
|
_write_json(cdsl_path, cdsl)
|
|
_write_json(diagnostic_path, diagnostic)
|
|
return {"part_id": part_id, "source": source_name, "status": "converted", "cdsl": cdsl_path.name, "diagnostic": diagnostic_path.name,
|
|
"future_rebuild_ready": diagnostic["semantic_validation"]["future_rebuild_ready"],
|
|
"deferred_feature_count": len(diagnostic["semantic_validation"]["deferred_feature_ids"]),
|
|
"unresolved_feature_count": len(diagnostic["semantic_validation"]["unresolved"])}
|
|
except Exception as error:
|
|
_write_json(diagnostic_path, {"source": source_name, "part_id": part_id, "error": str(error)})
|
|
return {"part_id": part_id, "source": source_name, "status": "failed", "diagnostic": diagnostic_path.name, "error": str(error)}
|
|
|
|
|
|
def batch_convert(evidence_dir: Path, output_dir: Path, *, truth_dir: Path | None = None, workers: int = 1,
|
|
fail_fast: bool = False, overwrite: bool = False) -> dict[str, Any]:
|
|
sources = sorted(evidence_dir.rglob(f"*{EVIDENCE_V2_SUFFIX}"))
|
|
if not sources:
|
|
raise ValueError(f"No *{EVIDENCE_V2_SUFFIX} files found in {evidence_dir}")
|
|
sources_by_base_id: dict[str, list[Path]] = {}
|
|
for source in sources:
|
|
base_id = _part_id_from_source_name(source.name)
|
|
sources_by_base_id.setdefault(base_id, []).append(source)
|
|
part_id_by_source: dict[Path, str] = {}
|
|
for base_id, matching_sources in sources_by_base_id.items():
|
|
if len(matching_sources) == 1:
|
|
part_id_by_source[matching_sources[0]] = base_id
|
|
continue
|
|
for source in matching_sources:
|
|
relative_name = source.relative_to(evidence_dir).as_posix()
|
|
digest = hashlib.sha256(relative_name.encode("utf-8")).hexdigest()[:10]
|
|
part_id_by_source[source] = f"{base_id[:69]}-{digest}"
|
|
if len(set(part_id_by_source.values())) != len(part_id_by_source):
|
|
raise ValueError("Unable to create unique output part IDs from evidence source paths")
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
def convert(source: Path) -> dict[str, Any]:
|
|
relative = source.relative_to(evidence_dir)
|
|
step_relative = Path(str(relative).removesuffix(EVIDENCE_V2_SUFFIX) + ".step")
|
|
return _convert_path(source, output_dir, truth_dir, overwrite, relative.as_posix(), step_relative,
|
|
part_id_by_source[source])
|
|
results: list[dict[str, Any]] = []
|
|
if workers == 1:
|
|
for source in sources:
|
|
item = convert(source)
|
|
results.append(item)
|
|
if fail_fast and item["status"] == "failed":
|
|
break
|
|
else:
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
|
|
for item in executor.map(convert, sources):
|
|
results.append(item)
|
|
if fail_fast and item["status"] == "failed":
|
|
break
|
|
manifest = {
|
|
"schema": "cdsl.evidence-v2.batch-manifest.v1", "input_directory": str(evidence_dir),
|
|
"truth_directory": str(truth_dir) if truth_dir else None, "input_count": len(sources),
|
|
"converted_count": sum(item["status"] == "converted" for item in results),
|
|
"skipped_count": sum(item["status"] == "skipped" for item in results),
|
|
"failed_count": sum(item["status"] == "failed" for item in results),
|
|
"future_rebuild_ready_count": sum(bool(item.get("future_rebuild_ready")) for item in results),
|
|
"unresolved_feature_count": sum(int(item.get("unresolved_feature_count") or 0) for item in results),
|
|
"deferred_feature_count": sum(int(item.get("deferred_feature_count") or 0) for item in results),
|
|
"results": results,
|
|
}
|
|
_write_json(output_dir / "manifest.json", manifest)
|
|
if manifest["failed_count"]:
|
|
raise RuntimeError(f"{manifest['failed_count']} conversion(s) failed; see {output_dir / 'manifest.json'}")
|
|
return manifest
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("evidence_dir", type=Path, help="directory containing *.solidworks_evidence_v2.json")
|
|
parser.add_argument("--out", type=Path, default=ROOT / "json_to_cdsl" / "output", help="output CDSL directory")
|
|
parser.add_argument("--truth-dir", type=Path, default=None, help="optional directory containing <part_id>.step truth files")
|
|
parser.add_argument("--workers", type=int, default=1, help="parallel conversion workers; STEP inspection remains conservative")
|
|
parser.add_argument("--fail-fast", action="store_true", help="stop after the first conversion failure")
|
|
parser.add_argument("--overwrite", action="store_true", help="overwrite existing per-part outputs")
|
|
args = parser.parse_args()
|
|
if args.workers < 1:
|
|
parser.error("--workers must be >= 1")
|
|
manifest = batch_convert(args.evidence_dir, args.out, truth_dir=args.truth_dir, workers=args.workers,
|
|
fail_fast=args.fail_fast, overwrite=args.overwrite)
|
|
print(json.dumps({key: manifest[key] for key in manifest if key != "results"}, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|