#!/usr/bin/env python3 """Extract an independent, editable DesignIR 3.0 surface/topology program.""" from __future__ import annotations import argparse from collections import Counter from concurrent.futures import ThreadPoolExecutor, as_completed import copy import hashlib import json import math from pathlib import Path import re import subprocess import sys import tempfile from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from designir_codec import read_designir, write_designir from build123d import Shape, import_step from OCP.BRep import BRep_Builder, BRep_Tool from OCP.BRepBndLib import BRepBndLib from OCP.BRepCheck import BRepCheck_Analyzer from OCP.BRepClass import BRepClass_FaceClassifier from OCP.BRepExtrema import BRepExtrema_DistShapeShape from OCP.BRepAdaptor import BRepAdaptor_Curve, BRepAdaptor_Surface from OCP.BRepAlgoAPI import BRepAlgoAPI_Cut from OCP.BRepLib import BRepLib from OCP.BRepGProp import BRepGProp from OCP.BRepBuilderAPI import ( BRepBuilderAPI_GTransform, BRepBuilderAPI_MakeEdge, BRepBuilderAPI_MakeFace, BRepBuilderAPI_MakeSolid, BRepBuilderAPI_MakeVertex, BRepBuilderAPI_MakeWire, BRepBuilderAPI_Sewing, BRepBuilderAPI_Transform, ) from OCP.BRepTools import BRepTools, BRepTools_WireExplorer from OCP.BRepPrimAPI import BRepPrimAPI_MakeCylinder from OCP.Bnd import Bnd_Box from OCP.Geom import ( Geom_BSplineCurve, Geom_BSplineSurface, Geom_Circle, Geom_ConicalSurface, Geom_CylindricalSurface, Geom_Ellipse, Geom_Line, Geom_Plane, Geom_SphericalSurface, Geom_ToroidalSurface, ) from OCP.Geom2d import ( Geom2d_BSplineCurve, Geom2d_Circle, Geom2d_Ellipse, Geom2d_Line, ) from OCP.Geom2dAdaptor import Geom2dAdaptor_Curve from OCP.GeomAbs import ( GeomAbs_BSplineCurve, GeomAbs_BSplineSurface, GeomAbs_Circle, GeomAbs_Cone, GeomAbs_Cylinder, GeomAbs_Ellipse, GeomAbs_Line, GeomAbs_Plane, GeomAbs_Sphere, GeomAbs_Torus, ) from OCP.TColgp import TColgp_Array1OfPnt, TColgp_Array1OfPnt2d, TColgp_Array2OfPnt from OCP.TColStd import ( TColStd_Array1OfInteger, TColStd_Array1OfReal, TColStd_Array2OfReal, ) from OCP.IFSelect import IFSelect_RetDone from OCP.GProp import GProp_GProps from OCP.STEPControl import STEPControl_AsIs, STEPControl_Reader, STEPControl_Writer from OCP.ShapeFix import ShapeFix_Face, ShapeFix_Shape, ShapeFix_Shell from OCP.TopAbs import ( TopAbs_EDGE, TopAbs_FACE, TopAbs_FORWARD, TopAbs_IN, TopAbs_ON, TopAbs_SHELL, TopAbs_SOLID, TopAbs_VERTEX, TopAbs_WIRE, ) from OCP.TopExp import TopExp_Explorer from OCP.TopExp import TopExp from OCP.TopLoc import TopLoc_Location from OCP.TopTools import TopTools_IndexedMapOfShape from OCP.TopoDS import TopoDS, TopoDS_Compound, TopoDS_Shell, TopoDS_Vertex from OCP.gp import ( gp_Ax2, gp_Ax22d, gp_Ax3, gp_Circ, gp_Circ2d, gp_Dir, gp_Dir2d, gp_Elips, gp_Elips2d, gp_GTrsf, gp_Lin, gp_Lin2d, gp_Mat, gp_Pnt, gp_Pnt2d, gp_Trsf, gp_XYZ, ) def _number(value: float) -> float: result = float(value) if not math.isfinite(result): raise ValueError("DesignIR cannot contain non-finite geometry") return result def _point(value: Any) -> list[float]: return [_number(value.X()), _number(value.Y()), _number(value.Z())] def _direction(value: Any) -> list[float]: return _point(value) def _axis3(position: Any) -> dict[str, Any]: direct = ( bool(position.Direct()) if hasattr(position, "Direct") else bool( position.XDirection() .Crossed(position.YDirection()) .Dot(position.Direction()) > 0 ) ) return { "location": _point(position.Location()), "axis": _direction(position.Direction()), "x_direction": _direction(position.XDirection()), "y_direction": _direction(position.YDirection()), "direct": direct, } def _bspline_curve(curve: Any) -> dict[str, Any]: return { "type": "bspline", "degree": int(curve.Degree()), "periodic": bool(curve.IsPeriodic()), "rational": bool(curve.IsRational()), "poles": [_point(curve.Pole(index)) for index in range(1, curve.NbPoles() + 1)], "weights": [ _number(curve.Weight(index)) for index in range(1, curve.NbPoles() + 1) ], "knots": [ _number(curve.Knot(index)) for index in range(1, curve.NbKnots() + 1) ], "multiplicities": [ int(curve.Multiplicity(index)) for index in range(1, curve.NbKnots() + 1) ], } def _curve_record(edge: Any) -> dict[str, Any]: if bool(BRep_Tool.Degenerated_s(edge)): return {"type": "degenerate"} adaptor = BRepAdaptor_Curve(edge) first, last = map(_number, BRep_Tool.Range_s(edge)) kind = adaptor.GetType() if kind == GeomAbs_Line: line = adaptor.Line() payload = { "type": "line", "location": _point(line.Location()), "direction": _direction(line.Direction()), } elif kind == GeomAbs_Circle: circle = adaptor.Circle() payload = { "type": "circle", "position": _axis3(circle.Position()), "radius": _number(circle.Radius()), } elif kind == GeomAbs_Ellipse: ellipse = adaptor.Ellipse() payload = { "type": "ellipse", "position": _axis3(ellipse.Position()), "major_radius": _number(ellipse.MajorRadius()), "minor_radius": _number(ellipse.MinorRadius()), } elif kind == GeomAbs_BSplineCurve: payload = _bspline_curve(adaptor.BSpline()) else: raise ValueError(f"Unsupported curve type: {int(kind)}") payload["parameter_range"] = [first, last] return payload def _point2d(value: Any) -> list[float]: return [_number(value.X()), _number(value.Y())] def _pcurve_record(edge: Any, face: Any) -> dict[str, Any] | None: curve = BRep_Tool.CurveOnSurface_s(edge, face, 0.0, 0.0) if curve is None: return None adaptor = Geom2dAdaptor_Curve(curve) first, last = map(_number, BRep_Tool.Range_s(edge, face)) kind = adaptor.GetType() if kind == GeomAbs_Line: line = adaptor.Line() payload = { "type": "line", "location": _point2d(line.Location()), "direction": _point2d(line.Direction()), } elif kind == GeomAbs_Circle: circle = adaptor.Circle() position = circle.Position() payload = { "type": "circle", "location": _point2d(position.Location()), "x_direction": _point2d(position.XDirection()), "y_direction": _point2d(position.YDirection()), "radius": _number(circle.Radius()), } elif kind == GeomAbs_Ellipse: ellipse = adaptor.Ellipse() position = ellipse.Axis() payload = { "type": "ellipse", "location": _point2d(position.Location()), "x_direction": _point2d(position.XDirection()), "y_direction": _point2d(position.YDirection()), "major_radius": _number(ellipse.MajorRadius()), "minor_radius": _number(ellipse.MinorRadius()), } elif kind == GeomAbs_BSplineCurve: spline = adaptor.BSpline() payload = { "type": "bspline", "degree": int(spline.Degree()), "periodic": bool(spline.IsPeriodic()), "rational": bool(spline.IsRational()), "poles": [ _point2d(spline.Pole(index)) for index in range(1, spline.NbPoles() + 1) ], "weights": [ _number(spline.Weight(index)) for index in range(1, spline.NbPoles() + 1) ], "knots": [ _number(spline.Knot(index)) for index in range(1, spline.NbKnots() + 1) ], "multiplicities": [ int(spline.Multiplicity(index)) for index in range(1, spline.NbKnots() + 1) ], } else: raise ValueError(f"Unsupported pcurve type: {int(kind)}") payload["parameter_range"] = [first, last] return payload def _bspline_surface(surface: Any) -> dict[str, Any]: return { "type": "bspline", "u_degree": int(surface.UDegree()), "v_degree": int(surface.VDegree()), "u_periodic": bool(surface.IsUPeriodic()), "v_periodic": bool(surface.IsVPeriodic()), "u_rational": bool(surface.IsURational()), "v_rational": bool(surface.IsVRational()), "poles": [ [ _point(surface.Pole(u_index, v_index)) for v_index in range(1, surface.NbVPoles() + 1) ] for u_index in range(1, surface.NbUPoles() + 1) ], "weights": [ [ _number(surface.Weight(u_index, v_index)) for v_index in range(1, surface.NbVPoles() + 1) ] for u_index in range(1, surface.NbUPoles() + 1) ], "u_knots": [ _number(surface.UKnot(index)) for index in range(1, surface.NbUKnots() + 1) ], "v_knots": [ _number(surface.VKnot(index)) for index in range(1, surface.NbVKnots() + 1) ], "u_multiplicities": [ int(surface.UMultiplicity(index)) for index in range(1, surface.NbUKnots() + 1) ], "v_multiplicities": [ int(surface.VMultiplicity(index)) for index in range(1, surface.NbVKnots() + 1) ], } def _surface_record(face: Any) -> dict[str, Any]: adaptor = BRepAdaptor_Surface(face) kind = adaptor.GetType() if kind == GeomAbs_Plane: plane = adaptor.Plane() return {"type": "plane", "position": _axis3(plane.Position())} if kind == GeomAbs_Cylinder: cylinder = adaptor.Cylinder() return { "type": "cylinder", "position": _axis3(cylinder.Position()), "radius": _number(cylinder.Radius()), } if kind == GeomAbs_Cone: cone = adaptor.Cone() return { "type": "cone", "position": _axis3(cone.Position()), "reference_radius": _number(cone.RefRadius()), "semi_angle_radians": _number(cone.SemiAngle()), } if kind == GeomAbs_Sphere: sphere = adaptor.Sphere() return { "type": "sphere", "position": _axis3(sphere.Position()), "radius": _number(sphere.Radius()), } if kind == GeomAbs_Torus: torus = adaptor.Torus() return { "type": "torus", "position": _axis3(torus.Position()), "major_radius": _number(torus.MajorRadius()), "minor_radius": _number(torus.MinorRadius()), } if kind == GeomAbs_BSplineSurface: return _bspline_surface(adaptor.BSpline()) raise ValueError(f"Unsupported surface type: {int(kind)}") def _edge_records( wire: Any, face: Any, prefix: str, edge_map: TopTools_IndexedMapOfShape, vertex_map: TopTools_IndexedMapOfShape, ) -> list[dict[str, Any]]: records = [] explorer = BRepTools_WireExplorer(wire) index = 0 while explorer.More(): index += 1 edge = explorer.Current() canonical_edge = TopoDS.Edge_s(edge.Oriented(TopAbs_FORWARD)) topology_index = int(edge_map.FindIndex(edge)) if topology_index <= 0: raise ValueError("Cannot resolve shared topology edge") first_vertex = TopoDS_Vertex() last_vertex = TopoDS_Vertex() # Store topology-edge endpoint order independently of this wire # occurrence. BRepTools_WireExplorer returns a context-oriented edge; # using that orientation here and then applying it again during replay # reverses endpoint constraints twice and can create self-intersecting # wires on periodic faces. TopExp.Vertices_s(canonical_edge, first_vertex, last_vertex) vertex_ids = [] for vertex in (first_vertex, last_vertex): if vertex.IsNull(): continue vertex_index = int(vertex_map.FindIndex(vertex)) if vertex_index <= 0: raise ValueError("Cannot resolve shared topology vertex") vertex_ids.append(f"vertex_{vertex_index}") records.append( { "id": f"{prefix}.edge_{index}", "topology_edge_id": f"edge_{topology_index}", "vertex_ids": vertex_ids, "orientation": int(explorer.Orientation()), "closed": bool(canonical_edge.Closed()), "tolerance": _number(BRep_Tool.Tolerance_s(edge)), "curve": _curve_record(canonical_edge), "pcurve": _pcurve_record(edge, face), } ) explorer.Next() return records def _wire_records( face: Any, prefix: str, edge_map: TopTools_IndexedMapOfShape, vertex_map: TopTools_IndexedMapOfShape, ) -> list[dict[str, Any]]: records = [] # Preserve OCCT's contextual traversal orientation. In particular, a seam # edge can occur twice on one periodic face and the occurrence orientation # selects which of its two pcurves is returned by BRep_Tool. outer = BRepTools.OuterWire_s(face) explorer = TopExp_Explorer(face, TopAbs_WIRE) index = 0 while explorer.More(): index += 1 wire = TopoDS.Wire_s(explorer.Current()) canonical_wire = TopoDS.Wire_s( wire.Oriented(TopAbs_FORWARD) ) wire_id = f"{prefix}.wire_{index}" records.append( { "id": wire_id, "orientation": int(wire.Orientation()), "outer": bool(wire.IsSame(outer)), "edges": _edge_records( canonical_wire, face, wire_id, edge_map, vertex_map, ), } ) explorer.Next() return records def _face_records( shell: Any, prefix: str, edge_map: TopTools_IndexedMapOfShape, vertex_map: TopTools_IndexedMapOfShape, ) -> list[dict[str, Any]]: records = [] explorer = TopExp_Explorer(shell, TopAbs_FACE) index = 0 while explorer.More(): index += 1 face = TopoDS.Face_s(explorer.Current()) canonical_face = TopoDS.Face_s(face.Oriented(TopAbs_FORWARD)) face_id = f"{prefix}.face_{index}" properties = GProp_GProps() BRepGProp.SurfaceProperties_s(canonical_face, properties) volume_properties = GProp_GProps() BRepGProp.VolumeProperties_s(face, volume_properties) bounds = Bnd_Box() BRepBndLib.AddOptimal_s(canonical_face, bounds) ( x_min, y_min, z_min, x_max, y_max, z_max, ) = bounds.Get() records.append( { "id": face_id, "orientation": int(face.Orientation()), "area": _number(abs(float(properties.Mass()))), "centroid": _point(properties.CentreOfMass()), "bounds": [ _number(value) for value in ( x_min, y_min, z_min, x_max, y_max, z_max, ) ], "volume_contribution": _number( float(volume_properties.Mass()) ), "surface": _surface_record(canonical_face), "uv_bounds": [ _number(value) for value in BRepTools.UVBounds_s(canonical_face) ], "wires": _wire_records( canonical_face, face_id, edge_map, vertex_map ), } ) explorer.Next() return records def _vertex_records(vertex_map: TopTools_IndexedMapOfShape) -> list[dict[str, Any]]: records = [] for index in range(1, vertex_map.Extent() + 1): vertex = TopoDS.Vertex_s(vertex_map.FindKey(index)) records.append( { "id": f"vertex_{index}", "point": _point(BRep_Tool.Pnt_s(vertex)), "tolerance": _number(BRep_Tool.Tolerance_s(vertex)), } ) return records def _signed_volume(shape: Any) -> float: properties = GProp_GProps() BRepGProp.VolumeProperties_s(shape, properties) return float(properties.Mass()) def _topology_records(shape: Any) -> dict[str, Any]: edge_map = TopTools_IndexedMapOfShape() TopExp.MapShapes_s(shape, TopAbs_EDGE, edge_map) vertex_map = TopTools_IndexedMapOfShape() TopExp.MapShapes_s(shape, TopAbs_VERTEX, vertex_map) solids = [] solid_shell_shapes = [] solid_explorer = TopExp_Explorer(shape, TopAbs_SOLID) solid_index = 0 while solid_explorer.More(): solid_index += 1 solid = TopoDS.Solid_s(solid_explorer.Current()) canonical_solid = TopoDS.Solid_s(solid.Oriented(TopAbs_FORWARD)) shells = [] shell_explorer = TopExp_Explorer(canonical_solid, TopAbs_SHELL) shell_index = 0 while shell_explorer.More(): shell_index += 1 shell = TopoDS.Shell_s(shell_explorer.Current()) canonical_shell = TopoDS.Shell_s( shell.Oriented(TopAbs_FORWARD) ) solid_shell_shapes.append(shell) shell_id = f"solid_{solid_index}.shell_{shell_index}" shells.append( { "id": shell_id, "orientation": int(shell.Orientation()), "faces": _face_records( canonical_shell, shell_id, edge_map, vertex_map ), } ) shell_explorer.Next() solids.append( { "id": f"solid_{solid_index}", "orientation": int(solid.Orientation()), "signed_volume": _number( _signed_volume(solid) ), "shells": shells, } ) solid_explorer.Next() free_shells = [] shell_explorer = TopExp_Explorer(shape, TopAbs_SHELL) free_shell_index = 0 while shell_explorer.More(): shell = TopoDS.Shell_s(shell_explorer.Current()) if not any(shell.IsSame(candidate) for candidate in solid_shell_shapes): free_shell_index += 1 shell_id = f"free_shell_{free_shell_index}" canonical_shell = TopoDS.Shell_s( shell.Oriented(TopAbs_FORWARD) ) free_shells.append( { "id": shell_id, "orientation": int(shell.Orientation()), "faces": _face_records( canonical_shell, shell_id, edge_map, vertex_map ), } ) shell_explorer.Next() return { "vertices": _vertex_records(vertex_map), "solids": solids, "free_shells": free_shells, } def _normalized_axis(position: dict[str, Any]) -> tuple[list[float], list[float]]: axis = [float(value) for value in position["axis"]] magnitude = math.sqrt(sum(value * value for value in axis)) axis = [value / magnitude for value in axis] for value in axis: if abs(value) > 1e-9: if value < 0: axis = [-item for item in axis] break location = [float(value) for value in position["location"]] distance = sum(location[index] * axis[index] for index in range(3)) anchor = [ location[index] - distance * axis[index] for index in range(3) ] return axis, anchor def _parallel(first: list[float], second: list[float], tolerance: float = 1e-6) -> bool: return abs(abs(sum(a * b for a, b in zip(first, second))) - 1.0) <= tolerance def _distance(first: list[float], second: list[float]) -> float: return math.sqrt(sum((a - b) ** 2 for a, b in zip(first, second))) def _dot(first: list[float], second: list[float]) -> float: return sum(a * b for a, b in zip(first, second)) def _cross(first: list[float], second: list[float]) -> list[float]: return [ first[1] * second[2] - first[2] * second[1], first[2] * second[0] - first[0] * second[2], first[0] * second[1] - first[1] * second[0], ] def _surfaceir_faces(topology: dict[str, Any]) -> list[dict[str, Any]]: solid_faces = [ face for solid in topology["solids"] for shell in solid["shells"] for face in shell["faces"] ] free_faces = [ face for shell in topology.get("free_shells", []) for face in shell["faces"] ] return solid_faces + free_faces def _observed_internal_cylinders(topology: dict[str, Any]) -> list[dict[str, Any]]: observations = [] for face in _surfaceir_faces(topology): surface = face["surface"] if surface["type"] != "cylinder" or int(face["orientation"]) != 1: continue axis, anchor = _normalized_axis(surface["position"]) observations.append( { "face_id": face["id"], "axis": axis, "anchor": anchor, "radius": float(surface["radius"]), "axial_span": abs( float(face["uv_bounds"][3]) - float(face["uv_bounds"][2]) ), } ) return observations def _cylinder_semantics(topology: dict[str, Any]) -> dict[str, Any]: """Infer conservative paper-style names without claiming source history.""" cylinders = [] for face in _surfaceir_faces(topology): surface = face["surface"] if surface["type"] != "cylinder": continue axis, anchor = _normalized_axis(surface["position"]) cylinders.append( { "face_id": face["id"], "internal": int(face["orientation"]) == 1, "axis": axis, "anchor": anchor, "radius": float(surface["radius"]), "diameter": 2.0 * float(surface["radius"]), "axial_span": abs( float(face["uv_bounds"][3]) - float(face["uv_bounds"][2]) ), "boundary_edge_ids": sorted( { edge["topology_edge_id"] for wire in face["wires"] for edge in wire["edges"] } ), } ) parameters: dict[str, Any] = {} datums: dict[str, Any] = {} features: list[dict[str, Any]] = [] constraints: list[dict[str, Any]] = [] internal = [item for item in cylinders if item["internal"]] external = [item for item in cylinders if not item["internal"]] # Coaxial cylinder segments represent one canonical hole axis. This avoids # treating counterbores and stepped bores as unrelated holes. hole_axes: list[dict[str, Any]] = [] for item in sorted(internal, key=lambda value: value["face_id"]): match = next( ( candidate for candidate in hole_axes if _parallel(candidate["axis"], item["axis"]) and _distance(candidate["anchor"], item["anchor"]) <= 1e-5 ), None, ) if match is None: match = { "axis": item["axis"], "anchor": item["anchor"], "segments": [], } hole_axes.append(match) match["segments"].append(item) hole_axes.sort( key=lambda hole: ( tuple(round(value, 9) for value in hole["axis"]), tuple(round(value, 9) for value in hole["anchor"]), ) ) for index, hole in enumerate(hole_axes, 1): hole["id"] = f"hole_axis_{index}" for hole in hole_axes: datums[hole["id"]] = { "kind": "axis", "origin": [_number(value) for value in hole["anchor"]], "direction": [_number(value) for value in hole["axis"]], "inference": "canonical_from_final_brep", } # Equal-diameter parallel axes are candidates for a repeated hole feature. unused = set(range(len(hole_axes))) pattern_index = 0 while unused: seed_index = min(unused) seed = hole_axes[seed_index] seed_diameter = min(segment["diameter"] for segment in seed["segments"]) group = [ index for index in sorted(unused) if _parallel(seed["axis"], hole_axes[index]["axis"]) and abs( min( segment["diameter"] for segment in hole_axes[index]["segments"] ) - seed_diameter ) <= max(1e-5, seed_diameter * 1e-6) ] for index in group: unused.discard(index) if len(group) < 2: continue points = [hole_axes[index]["anchor"] for index in group] centroid = [ sum(point[dimension] for point in points) / len(points) for dimension in range(3) ] radial_distances = [_distance(point, centroid) for point in points] radius_mean = sum(radial_distances) / len(radial_distances) radial_equal = ( len(group) >= 3 and radius_mean > 1e-7 and max(abs(value - radius_mean) for value in radial_distances) <= max(1e-5, radius_mean * 1e-5) ) angular_equal = False if radial_equal: reference = [ (points[0][index] - centroid[index]) / radius_mean for index in range(3) ] transverse = _cross(seed["axis"], reference) transverse_length = math.sqrt(_dot(transverse, transverse)) if transverse_length > 1e-7: transverse = [ value / transverse_length for value in transverse ] angles = sorted( math.atan2( _dot( [ point[index] - centroid[index] for index in range(3) ], transverse, ), _dot( [ point[index] - centroid[index] for index in range(3) ], reference, ), ) % (2.0 * math.pi) for point in points ) angle_gaps = [ ( angles[(index + 1) % len(angles)] - angles[index] ) % (2.0 * math.pi) for index in range(len(angles)) ] expected_gap = 2.0 * math.pi / len(angles) angular_equal = max( abs(gap - expected_gap) for gap in angle_gaps ) <= 1e-5 polar = radial_equal and angular_equal # A covariance-free collinearity test is sufficient for exact CAD axes. direction = [ points[-1][dimension] - points[0][dimension] for dimension in range(3) ] direction_length = math.sqrt(sum(value * value for value in direction)) linear = direction_length > 1e-7 if linear: direction = [value / direction_length for value in direction] linear = all( math.sqrt( sum( ( (point[dimension] - points[0][dimension]) - sum( (point[index] - points[0][index]) * direction[index] for index in range(3) ) * direction[dimension] ) ** 2 for dimension in range(3) ) ) <= 1e-5 for point in points ) pattern_index += 1 feature_id = f"hole_pattern_{pattern_index}" diameter_name = ( "hole_diameter" if pattern_index == 1 else f"hole_diameter_{pattern_index}" ) count_name = ( "hole_count" if pattern_index == 1 else f"hole_count_{pattern_index}" ) parameters[diameter_name] = { "value": _number(seed_diameter), "unit": "mm", "editable": False, "semantic_role": "hole_diameter", "inference": "cylindrical_cut_surface", "confidence": 0.98, "affected_face_ids": [ segment["face_id"] for index in group for segment in hole_axes[index]["segments"] if abs(segment["diameter"] - seed_diameter) <= max(1e-5, seed_diameter * 1e-6) ], "edit_state": "executable_enlarge_cut_binding", } parameters[count_name] = { "value": len(group), "unit": "count", "editable": False, "semantic_role": "hole_count", "inference": "equal_diameter_parallel_axes", "confidence": 0.95, "edit_state": "awaiting_executable_binding", } pattern_kind = "polar" if polar else "linear" if linear else "set" feature = { "id": feature_id, "operation": f"{pattern_kind}_hole_pattern", "member_axis_ids": [hole_axes[index]["id"] for index in group], "diameter_parameter": diameter_name, "count_parameter": count_name, "inference": "canonical_from_final_brep", "original_history_recovered": False, "confidence": 0.95 if pattern_kind != "set" else 0.75, } if polar: parameter_name = ( "bolt_circle_diameter" if pattern_index == 1 else f"bolt_circle_diameter_{pattern_index}" ) parameters[parameter_name] = { "value": _number(2.0 * radius_mean), "unit": "mm", "editable": False, "semantic_role": "bolt_circle_diameter", "inference": "equal_radius_axis_distribution", "confidence": 0.98, "edit_state": "awaiting_executable_binding", } feature["pitch_diameter_parameter"] = parameter_name constraints.append( { "id": f"{feature_id}_equal_angular_spacing", "kind": "equal_angular_spacing", "participants": [feature_id], "confidence": 0.9, } ) elif linear: projections = sorted( sum( (point[index] - points[0][index]) * direction[index] for index in range(3) ) for point in points ) gaps = [ projections[index + 1] - projections[index] for index in range(len(projections) - 1) ] if gaps and max(gaps) - min(gaps) <= max(1e-5, max(gaps) * 1e-5): parameter_name = ( "hole_spacing" if pattern_index == 1 else f"hole_spacing_{pattern_index}" ) parameters[parameter_name] = { "value": _number(sum(gaps) / len(gaps)), "unit": "mm", "editable": False, "semantic_role": "hole_spacing", "inference": "equal_linear_axis_spacing", "confidence": 0.98, "edit_state": "awaiting_executable_binding", } feature["spacing_parameter"] = parameter_name constraints.append( { "id": f"{feature_id}_equal_spacing", "kind": "equal_linear_spacing", "participants": [feature_id], "confidence": 0.98, } ) features.append(feature) patterned_axes = { axis_id for feature in features for axis_id in feature.get("member_axis_ids", []) } vertex_points = [ [float(value) for value in vertex["point"]] for vertex in topology.get("vertices", []) ] bounds_center = ( [ ( min(point[index] for point in vertex_points) + max(point[index] for point in vertex_points) ) / 2.0 for index in range(3) ] if vertex_points else [0.0, 0.0, 0.0] ) bounds_diagonal = ( _distance( [min(point[index] for point in vertex_points) for index in range(3)], [max(point[index] for point in vertex_points) for index in range(3)], ) if vertex_points else 0.0 ) primary_external = ( max(external, key=lambda item: item["radius"]) if external else None ) unpatterned_index = 0 for hole in hole_axes: if hole["id"] in patterned_axes: continue unpatterned_index += 1 diameter = min(segment["diameter"] for segment in hole["segments"]) center_delta = [ bounds_center[index] - hole["anchor"][index] for index in range(3) ] axial_component = _dot(center_delta, hole["axis"]) axis_distance_from_center = math.sqrt( sum( ( center_delta[index] - axial_component * hole["axis"][index] ) ** 2 for index in range(3) ) ) central = ( primary_external is not None and _parallel(hole["axis"], primary_external["axis"]) and _distance(hole["anchor"], primary_external["anchor"]) <= max(1e-5, primary_external["diameter"] * 1e-6) ) or axis_distance_from_center <= max( 1e-5, bounds_diagonal * 1e-6, ) parameter_name = ( "bore_diameter" if central else f"hole_diameter_single_{unpatterned_index}" ) parameters[parameter_name] = { "value": _number(diameter), "unit": "mm", "editable": False, "semantic_role": ( "bore_diameter" if central else "hole_diameter" ), "inference": "cylindrical_cut_surface", "confidence": 0.9 if len(hole_axes) == 1 else 0.8, "affected_face_ids": [ segment["face_id"] for segment in hole["segments"] if abs(segment["diameter"] - diameter) <= max(1e-5, diameter * 1e-6) ], "edit_state": "executable_enlarge_cut_binding", } features.append( { "id": f"canonical_hole_{unpatterned_index}", "operation": ( "central_bore" if central else "cylindrical_cut" ), "axis_id": hole["id"], "diameter_parameter": parameter_name, "segment_face_ids": [ segment["face_id"] for segment in hole["segments"] ], "inference": "canonical_from_final_brep", "original_history_recovered": False, "confidence": 0.9 if len(hole_axes) == 1 else 0.8, } ) if external: has_feature_specific_cut_binding = any( parameter.get("edit_state") == "executable_enlarge_cut_binding" for parameter in parameters.values() ) primary_axis_edit_state = ( "awaiting_executable_binding" if has_feature_specific_cut_binding else "executable_axis_affine_binding" ) largest = max(external, key=lambda item: item["radius"]) parameters["outer_diameter"] = { "value": _number(largest["diameter"]), "unit": "mm", "editable": False, "semantic_role": "outer_diameter", "inference": "largest_external_cylinder", "confidence": 0.9, "affected_face_ids": [largest["face_id"]], "edit_state": primary_axis_edit_state, } feature = { "id": "primary_cylindrical_body", "operation": "cylindrical_body", "face_id": largest["face_id"], "diameter_parameter": "outer_diameter", "inference": "canonical_from_final_brep", "original_history_recovered": False, "confidence": 0.9, } if largest["axial_span"] > 1e-7: semantic_role = ( "plate_thickness" if largest["axial_span"] < largest["diameter"] * 0.35 else "body_length" ) parameters[semantic_role] = { "value": _number(largest["axial_span"]), "unit": "mm", "editable": False, "semantic_role": semantic_role, "inference": "external_cylinder_axial_span", "confidence": 0.85, "affected_face_ids": [largest["face_id"]], "edit_state": primary_axis_edit_state, } feature["length_parameter"] = semantic_role features.append(feature) return { "parameters": parameters, "datums": datums, "features": features, "constraints": constraints, "inference_policy": { "kind": "canonical_design_interpretation", "claims_original_feature_history": False, "editable_only_after_perturbation_validation": True, }, } def apply_semantic_parameter( payload: dict[str, Any], parameter_name: str, value: float ) -> dict[str, Any]: """Apply one explicitly bound semantic edit to an independent IR copy.""" result = copy.deepcopy(payload) parameters = result.get("semantic_layer", {}).get("parameters", {}) if parameter_name not in parameters: raise ValueError(f"Unknown semantic parameter: {parameter_name}") parameter = parameters[parameter_name] if parameter.get("edit_state") not in { "executable_enlarge_cut_binding", "executable_axis_affine_binding", "executable_uniform_scale_binding", "validated_executable_binding", }: raise ValueError( f"Semantic parameter {parameter_name} has no executable binding" ) role = parameter.get("semantic_role") if role not in { "hole_diameter", "bore_diameter", "outer_diameter", "body_length", "plate_thickness", "overall_size_x", "overall_size_y", "overall_size_z", "overall_scale", }: raise ValueError(f"Unsupported semantic edit role: {role}") new_value = float(value) if not math.isfinite(new_value) or new_value <= 0: raise ValueError("Semantic edits require a finite positive value") old_value = float(parameter["value"]) if role == "overall_scale": if math.isclose(new_value, old_value, rel_tol=0.0, abs_tol=1e-12): raise ValueError("Overall scale edit must change the value") pivot_name = parameter.get("pivot_datum") pivot = ( result.get("semantic_layer", {}) .get("datums", {}) .get(pivot_name, {}) .get("point") ) if not isinstance(pivot, list) or len(pivot) != 3: raise ValueError("Overall scale edit has no valid pivot datum") scale_factor = new_value / old_value parameter["value"] = _number(new_value) parameter["editable"] = True parameter["edit_state"] = "edited_pending_acceptance" result["edit_operations"] = [ { "operation": "uniform_scale", "factor": _number(scale_factor), "pivot": [_number(float(component)) for component in pivot], } ] preferred_boundary_strategy = result.get( "reconstruction_strategy", {} ).get("boundary_strategy", "analytic_uv_rect") result["reconstruction_strategy"] = { "boundary_strategy": preferred_boundary_strategy, "selection_status": "semantic_edit_candidate", } result["active_edit"] = { "parameter": parameter_name, "semantic_role": role, "old_value": _number(old_value), "new_value": _number(new_value), "operation_count": 1, "range_status": ( "within_validated_range" if parameter.get("validated_range") and float( parameter["validated_range"]["minimum_tested_inclusive"] ) <= new_value <= float( parameter["validated_range"]["maximum_tested_inclusive"] ) else "requires_revalidation" ), "acceptance_status": "pending", } return result if role in {"overall_size_x", "overall_size_y", "overall_size_z"}: if math.isclose(new_value, old_value, rel_tol=0.0, abs_tol=1e-12): raise ValueError(f"{role} edit must change the value") axis_index = {"overall_size_x": 0, "overall_size_y": 1, "overall_size_z": 2}[ role ] axis = [0.0, 0.0, 0.0] axis[axis_index] = 1.0 pivot_name = parameter.get("pivot_datum") pivot = ( result.get("semantic_layer", {}) .get("datums", {}) .get(pivot_name, {}) .get("point") ) if not isinstance(pivot, list) or len(pivot) != 3: raise ValueError(f"{role} has no valid pivot datum") factor = new_value / old_value parameter["value"] = _number(new_value) parameter["editable"] = True parameter["edit_state"] = "edited_pending_acceptance" result["edit_operations"] = [ { "operation": "axis_affine_scale", "axis": axis, "pivot": [_number(float(component)) for component in pivot], "axial_factor": _number(factor), "radial_factor": 1.0, "semantic_role": role, "target_measure": "bounds_extent", } ] preferred_boundary_strategy = result.get( "reconstruction_strategy", {} ).get("boundary_strategy", "analytic_uv_rect") result["reconstruction_strategy"] = { "boundary_strategy": preferred_boundary_strategy, "selection_status": "semantic_edit_candidate", } result["active_edit"] = { "parameter": parameter_name, "semantic_role": role, "old_value": _number(old_value), "new_value": _number(new_value), "operation_count": 1, "range_status": ( "within_validated_range" if parameter.get("validated_range") and float( parameter["validated_range"]["minimum_tested_inclusive"] ) <= new_value <= float( parameter["validated_range"]["maximum_tested_inclusive"] ) else "requires_revalidation" ), "acceptance_status": "pending", } return result if role in {"outer_diameter", "body_length", "plate_thickness"}: if math.isclose(new_value, old_value, rel_tol=0.0, abs_tol=1e-12): raise ValueError(f"{role} edit must change the value") affected_face_ids = set(parameter.get("affected_face_ids", [])) faces = _surfaceir_faces(result["surface_layer"]) target_face = next( ( face for face in faces if face["id"] in affected_face_ids and face.get("surface", {}).get("type") == "cylinder" ), None, ) if target_face is None: raise ValueError(f"{role} binding matched no primary cylinder") position = target_face["surface"]["position"] raw_axis = [float(component) for component in position["axis"]] axis_length = math.sqrt(sum(component**2 for component in raw_axis)) axis = [component / axis_length for component in raw_axis] location = [ float(component) for component in position["location"] ] v_min, v_max = sorted( [ float(target_face["uv_bounds"][2]), float(target_face["uv_bounds"][3]), ] ) midpoint = (v_min + v_max) / 2.0 pivot = [ _number(location[index] + midpoint * axis[index]) for index in range(3) ] factor = new_value / old_value axial_factor = ( factor if role in {"body_length", "plate_thickness"} else 1.0 ) radial_factor = factor if role == "outer_diameter" else 1.0 parameter["value"] = _number(new_value) parameter["editable"] = True parameter["edit_state"] = "edited_pending_acceptance" result["edit_operations"] = [ { "operation": "axis_affine_scale", "axis": [_number(component) for component in axis], "pivot": pivot, "axial_factor": _number(axial_factor), "radial_factor": _number(radial_factor), "target_face_id": target_face["id"], "semantic_role": role, } ] preferred_boundary_strategy = result.get( "reconstruction_strategy", {} ).get("boundary_strategy", "analytic_uv_rect") result["reconstruction_strategy"] = { "boundary_strategy": preferred_boundary_strategy, "selection_status": "semantic_edit_candidate", } result["active_edit"] = { "parameter": parameter_name, "semantic_role": role, "old_value": _number(old_value), "new_value": _number(new_value), "edited_face_ids": [target_face["id"]], "operation_count": 1, "range_status": ( "within_validated_range" if parameter.get("validated_range") and float( parameter["validated_range"]["minimum_tested_inclusive"] ) <= new_value <= float( parameter["validated_range"]["maximum_tested_inclusive"] ) else "requires_revalidation" ), "acceptance_status": "pending", } return result if new_value <= old_value: raise ValueError( f"Semantic parameter {parameter_name} currently supports " "diameter increases only" ) old_radius = old_value / 2.0 new_radius = new_value / 2.0 affected_face_ids = set(parameter.get("affected_face_ids", [])) if not affected_face_ids: raise ValueError(f"Semantic parameter {parameter_name} has no target faces") faces = _surfaceir_faces(result["surface_layer"]) edited_face_ids: list[str] = [] edit_operations: list[dict[str, Any]] = [] seen_operations: set[tuple[Any, ...]] = set() radius_tolerance = max(1e-6, abs(old_radius) * 1e-6) for face in faces: if face["id"] not in affected_face_ids: continue surface = face["surface"] if ( surface.get("type") != "cylinder" or abs(float(surface["radius"]) - old_radius) > radius_tolerance ): continue edited_face_ids.append(face["id"]) operation_axis, operation_anchor = _normalized_axis(surface["position"]) operation_v_range = sorted( [_number(face["uv_bounds"][2]), _number(face["uv_bounds"][3])] ) operation_key = ( tuple(round(value, 8) for value in operation_axis), tuple(round(value, 8) for value in operation_anchor), round(operation_v_range[0], 8), round(operation_v_range[1], 8), ) if operation_key in seen_operations: continue seen_operations.add(operation_key) edit_operations.append( { "operation": "enlarge_cylindrical_cut", "face_id": face["id"], "position": copy.deepcopy(surface["position"]), "v_range": operation_v_range, "old_radius": _number(old_radius), "new_radius": _number(new_radius), } ) if not edited_face_ids: raise ValueError( f"Semantic parameter {parameter_name} binding matched no cylinders" ) merged_operations: list[dict[str, Any]] = [] for operation in edit_operations: operation_axis, operation_anchor = _normalized_axis( operation["position"] ) location = [ float(value) for value in operation["position"]["location"] ] raw_axis = [ float(value) for value in operation["position"]["axis"] ] axis_sign = sum( first * second for first, second in zip(raw_axis, operation_axis) ) location_scalar = sum( first * second for first, second in zip(location, operation_axis) ) absolute_interval = sorted( location_scalar + axis_sign * float(value) for value in operation["v_range"] ) duplicate = None for candidate in merged_operations: candidate_axis = candidate["_canonical_axis"] candidate_anchor = candidate["_canonical_anchor"] candidate_interval = candidate["_absolute_interval"] span = max( absolute_interval[1] - absolute_interval[0], candidate_interval[1] - candidate_interval[0], 1e-9, ) interval_tolerance = max(1e-4, span * 1e-5) if ( _parallel(operation_axis, candidate_axis) and _distance(operation_anchor, candidate_anchor) <= 1e-5 and abs(absolute_interval[0] - candidate_interval[0]) <= interval_tolerance and abs(absolute_interval[1] - candidate_interval[1]) <= interval_tolerance ): duplicate = candidate break if duplicate is None: candidate = copy.deepcopy(operation) candidate["source_face_ids"] = [operation["face_id"]] candidate["_canonical_axis"] = operation_axis candidate["_canonical_anchor"] = operation_anchor candidate["_absolute_interval"] = absolute_interval merged_operations.append(candidate) else: duplicate["source_face_ids"].append(operation["face_id"]) duplicate["_absolute_interval"] = [ min(duplicate["_absolute_interval"][0], absolute_interval[0]), max(duplicate["_absolute_interval"][1], absolute_interval[1]), ] duplicate_location = [ float(value) for value in duplicate["position"]["location"] ] duplicate_axis = [ float(value) for value in duplicate["position"]["axis"] ] duplicate_location_scalar = sum( first * second for first, second in zip( duplicate_location, duplicate["_canonical_axis"] ) ) duplicate_sign = sum( first * second for first, second in zip( duplicate_axis, duplicate["_canonical_axis"] ) ) duplicate["v_range"] = sorted( ( value - duplicate_location_scalar ) / duplicate_sign for value in duplicate["_absolute_interval"] ) for operation in merged_operations: operation.pop("_canonical_axis", None) operation.pop("_canonical_anchor", None) operation.pop("_absolute_interval", None) operation["source_face_ids"] = sorted( set(operation["source_face_ids"]) ) parameter["value"] = _number(new_value) parameter["editable"] = True parameter["edit_state"] = "edited_pending_acceptance" result["edit_operations"] = merged_operations preferred_boundary_strategy = result.get( "reconstruction_strategy", {} ).get("boundary_strategy", "analytic_uv_rect") result["reconstruction_strategy"] = { "boundary_strategy": preferred_boundary_strategy, "selection_status": "semantic_edit_candidate", } result["active_edit"] = { "parameter": parameter_name, "semantic_role": role, "old_value": _number(old_value), "new_value": _number(new_value), "edited_face_ids": sorted(edited_face_ids), "operation_count": len(merged_operations), "range_status": ( "within_validated_range" if parameter.get("validated_range") and new_value <= float( parameter["validated_range"]["maximum_tested_inclusive"] ) else "requires_revalidation" ), "acceptance_status": "pending", } return result def _point_value(values: list[float]) -> gp_Pnt: return gp_Pnt(*map(float, values)) def _direction_value(values: list[float]) -> gp_Dir: return gp_Dir(*map(float, values)) def _ax3_value(payload: dict[str, Any]) -> gp_Ax3: result = gp_Ax3( _point_value(payload["location"]), _direction_value(payload["axis"]), _direction_value(payload["x_direction"]), ) if bool(result.Direct()) != bool(payload.get("direct", True)): result.YReverse() return result def _ax2_value(payload: dict[str, Any]) -> gp_Ax2: return gp_Ax2( _point_value(payload["location"]), _direction_value(payload["axis"]), _direction_value(payload["x_direction"]), ) def _real_array(values: list[float]) -> TColStd_Array1OfReal: result = TColStd_Array1OfReal(1, len(values)) for index, value in enumerate(values, 1): result.SetValue(index, float(value)) return result def _integer_array(values: list[int]) -> TColStd_Array1OfInteger: result = TColStd_Array1OfInteger(1, len(values)) for index, value in enumerate(values, 1): result.SetValue(index, int(value)) return result def _point_array(values: list[list[float]]) -> TColgp_Array1OfPnt: result = TColgp_Array1OfPnt(1, len(values)) for index, value in enumerate(values, 1): result.SetValue(index, _point_value(value)) return result def _point2d_array(values: list[list[float]]) -> TColgp_Array1OfPnt2d: result = TColgp_Array1OfPnt2d(1, len(values)) for index, value in enumerate(values, 1): result.SetValue(index, gp_Pnt2d(*map(float, value))) return result def _pcurve_from_record(payload: dict[str, Any]) -> Any: kind = payload["type"] if kind == "line": return Geom2d_Line( gp_Lin2d( gp_Pnt2d(*map(float, payload["location"])), gp_Dir2d(*map(float, payload["direction"])), ) ) if kind == "circle": axis = gp_Ax22d( gp_Pnt2d(*map(float, payload["location"])), gp_Dir2d(*map(float, payload["x_direction"])), gp_Dir2d(*map(float, payload["y_direction"])), ) return Geom2d_Circle(gp_Circ2d(axis, float(payload["radius"]))) if kind == "ellipse": axis = gp_Ax22d( gp_Pnt2d(*map(float, payload["location"])), gp_Dir2d(*map(float, payload["x_direction"])), gp_Dir2d(*map(float, payload["y_direction"])), ) return Geom2d_Ellipse( gp_Elips2d( axis, float(payload["major_radius"]), float(payload["minor_radius"]), ) ) if kind == "bspline": poles = _point2d_array(payload["poles"]) knots = _real_array(payload["knots"]) mults = _integer_array(payload["multiplicities"]) if payload.get("rational"): return Geom2d_BSplineCurve( poles, _real_array(payload["weights"]), knots, mults, int(payload["degree"]), bool(payload["periodic"]), ) return Geom2d_BSplineCurve( poles, knots, mults, int(payload["degree"]), bool(payload["periodic"]), ) raise ValueError(f"Unsupported pcurve record: {kind}") def _curve_from_record( payload: dict[str, Any], surface: Any | None = None, pcurve_payload: dict[str, Any] | None = None, vertices: list[Any] | None = None, ) -> Any: vertices = vertices or [] kind = payload["type"] if surface is not None and pcurve_payload is not None: first, last = map(float, pcurve_payload["parameter_range"]) arguments: list[Any] = [ _pcurve_from_record(pcurve_payload), surface, ] if len(vertices) == 2: arguments.extend(vertices) arguments.extend([first, last]) try: edge = BRepBuilderAPI_MakeEdge(*arguments).Edge() except Exception: # Imported STEP vertex tolerances can be tighter than the # reconstructed surface/pcurve pair. The pcurve and its parameter # range are authoritative for this face; let OCCT recover endpoint # vertices when the source vertices cannot be reused consistently. edge = BRepBuilderAPI_MakeEdge( arguments[0], surface, first, last ).Edge() if kind == "degenerate": BRep_Builder().Degenerated(edge, True) return edge first, last = map(float, payload.get("parameter_range", [0.0, 1.0])) if kind == "line": curve = Geom_Line( gp_Lin( _point_value(payload["location"]), _direction_value(payload["direction"]), ) ) arguments = [curve] if len(vertices) == 2: arguments.extend(vertices) arguments.extend([first, last]) return BRepBuilderAPI_MakeEdge(*arguments).Edge() if kind == "circle": curve = Geom_Circle( gp_Circ(_ax2_value(payload["position"]), float(payload["radius"])) ) arguments = [curve] if len(vertices) == 2: arguments.extend(vertices) arguments.extend([first, last]) return BRepBuilderAPI_MakeEdge(*arguments).Edge() if kind == "ellipse": curve = Geom_Ellipse( gp_Elips( _ax2_value(payload["position"]), float(payload["major_radius"]), float(payload["minor_radius"]), ) ) arguments = [curve] if len(vertices) == 2: arguments.extend(vertices) arguments.extend([first, last]) return BRepBuilderAPI_MakeEdge(*arguments).Edge() if kind == "bspline": poles = _point_array(payload["poles"]) knots = _real_array(payload["knots"]) mults = _integer_array(payload["multiplicities"]) if payload.get("rational"): curve = Geom_BSplineCurve( poles, _real_array(payload["weights"]), knots, mults, int(payload["degree"]), bool(payload["periodic"]), ) else: curve = Geom_BSplineCurve( poles, knots, mults, int(payload["degree"]), bool(payload["periodic"]), ) arguments = [curve] if len(vertices) == 2: arguments.extend(vertices) arguments.extend([first, last]) return BRepBuilderAPI_MakeEdge(*arguments).Edge() if kind == "degenerate": return None raise ValueError(f"Unsupported curve record: {kind}") def _surface_from_record(payload: dict[str, Any]) -> Any: kind = payload["type"] if kind == "plane": return Geom_Plane(_ax3_value(payload["position"])) if kind == "cylinder": return Geom_CylindricalSurface( _ax3_value(payload["position"]), float(payload["radius"]) ) if kind == "cone": return Geom_ConicalSurface( _ax3_value(payload["position"]), float(payload["semi_angle_radians"]), float(payload["reference_radius"]), ) if kind == "sphere": return Geom_SphericalSurface( _ax3_value(payload["position"]), float(payload["radius"]) ) if kind == "torus": return Geom_ToroidalSurface( _ax3_value(payload["position"]), float(payload["major_radius"]), float(payload["minor_radius"]), ) if kind == "bspline": poles_data = payload["poles"] weights_data = payload["weights"] poles = TColgp_Array2OfPnt( 1, len(poles_data), 1, len(poles_data[0]) ) weights = TColStd_Array2OfReal( 1, len(weights_data), 1, len(weights_data[0]) ) for u_index, row in enumerate(poles_data, 1): for v_index, value in enumerate(row, 1): poles.SetValue(u_index, v_index, _point_value(value)) weights.SetValue( u_index, v_index, float(weights_data[u_index - 1][v_index - 1]) ) arguments = [ poles, _real_array(payload["u_knots"]), _real_array(payload["v_knots"]), _integer_array(payload["u_multiplicities"]), _integer_array(payload["v_multiplicities"]), int(payload["u_degree"]), int(payload["v_degree"]), bool(payload["u_periodic"]), bool(payload["v_periodic"]), ] if payload.get("u_rational") or payload.get("v_rational"): arguments.insert(1, weights) return Geom_BSplineSurface(*arguments) raise ValueError(f"Unsupported surface record: {kind}") def _wire_from_record( payload: dict[str, Any], surface: Any, boundary_strategy: str, edge_cache: dict[str, Any], vertex_cache: dict[str, Any], ) -> Any: builder = BRepBuilderAPI_MakeWire() resolved_edges: list[tuple[dict[str, Any], Any, bool]] = [] for edge_payload in payload["edges"]: topology_edge_id = str( edge_payload.get("topology_edge_id", edge_payload["id"]) ) use_shared_edge = boundary_strategy != "first_pcurve" edge = edge_cache.get(topology_edge_id) if use_shared_edge else None edge_was_created = edge is None if edge is None: vertices = [ vertex_cache[vertex_id] for vertex_id in edge_payload.get("vertex_ids", []) if vertex_id in vertex_cache ] use_parametric_curve = ( boundary_strategy in {"first_pcurve", "shared_pcurve"} or edge_payload["curve"]["type"] == "degenerate" ) try: edge = _curve_from_record( edge_payload["curve"], surface if use_parametric_curve else None, edge_payload.get("pcurve") if use_parametric_curve else None, vertices, ) except Exception as exc: raise ValueError( f"Cannot rebuild edge {edge_payload['id']}: " f"{type(exc).__name__}: {exc}" ) from exc if edge is not None and use_shared_edge: edge_cache[topology_edge_id] = edge if edge is None: continue edge.Closed(bool(edge_payload.get("closed", edge.Closed()))) tolerance = max(float(edge_payload.get("tolerance", 1e-7)), 1e-7) BRep_Builder().UpdateEdge(edge, tolerance) resolved_edges.append((edge_payload, edge, edge_was_created)) if boundary_strategy in {"exact_3d_pcurve", "shared_pcurve"}: edge_occurrences: dict[ str, tuple[Any, list[dict[str, Any]], list[bool]] ] = {} for edge_payload, edge, edge_was_created in resolved_edges: pcurve = edge_payload.get("pcurve") if pcurve is None: continue topology_edge_id = str( edge_payload.get("topology_edge_id", edge_payload["id"]) ) if topology_edge_id not in edge_occurrences: edge_occurrences[topology_edge_id] = (edge, [], []) edge_occurrences[topology_edge_id][1].append(edge_payload) edge_occurrences[topology_edge_id][2].append(edge_was_created) edge_builder = BRep_Builder() location = TopLoc_Location() for topology_edge_id, ( edge, occurrences, creation_flags, ) in edge_occurrences.items(): if len(occurrences) > 2: raise ValueError( "Cannot bind more than two pcurves for shared edge " f"{topology_edge_id} on one face" ) tolerance = max( max( float(occurrence.get("tolerance", 1e-7)) for occurrence in occurrences ), 1e-7, ) if len(occurrences) == 2: # OCCT binds the first seam pcurve to the FORWARD occurrence # and the second to REVERSED. Wire traversal order is not # stable across imported STEP topology. occurrences = sorted( occurrences, key=lambda occurrence: int( occurrence.get("orientation", 0) ), ) pcurves = [ _pcurve_from_record(occurrence["pcurve"]) for occurrence in occurrences ] try: if ( boundary_strategy == "shared_pcurve" and len(occurrences) == 1 and creation_flags[0] ): # BRepBuilderAPI_MakeEdge already attached this pcurve to # the newly created edge. Reattaching it can replace the # valid trim with the full periodic surface interval. continue if len(pcurves) == 2: edge_builder.UpdateEdge( edge, pcurves[0], pcurves[1], surface, location, tolerance, ) else: edge_builder.UpdateEdge( edge, pcurves[0], surface, location, tolerance ) first, last = map( float, occurrences[0]["pcurve"]["parameter_range"] ) edge_builder.Range(edge, surface, location, first, last) except Exception as exc: raise ValueError( f"Cannot bind pcurve for edge {topology_edge_id}: " f"{type(exc).__name__}: {exc}" ) from exc for edge_payload, edge, _edge_was_created in resolved_edges: if int(edge_payload["orientation"]) == 1: edge = TopoDS.Edge_s(edge.Reversed()) builder.Add(edge) if not builder.IsDone(): raise ValueError(f"Cannot rebuild wire {payload['id']}") wire = builder.Wire() if int(payload.get("orientation", 0)) == 1: wire = TopoDS.Wire_s(wire.Reversed()) return wire def _face_from_record( payload: dict[str, Any], boundary_strategy: str, edge_cache: dict[str, Any], vertex_cache: dict[str, Any], ) -> Any: surface = _surface_from_record(payload["surface"]) wire_payloads = sorted(payload["wires"], key=lambda item: not item.get("outer")) rectangular_parametric_patch = ( ( payload.get("edit_rebuild_strategy") == "uv_rect" or ( boundary_strategy in {"uv_rect_all", "analytic_uv_rect"} and payload["surface"]["type"] != "plane" and ( boundary_strategy == "uv_rect_all" or payload["surface"]["type"] != "bspline" ) ) ) and len(wire_payloads) == 1 and all( edge.get("pcurve", {}).get("type") == "line" and ( abs(float(edge["pcurve"]["direction"][0])) <= 1e-9 or abs(float(edge["pcurve"]["direction"][1])) <= 1e-9 ) for edge in wire_payloads[0]["edges"] ) ) if not wire_payloads or rectangular_parametric_patch: u_min, u_max, v_min, v_max = map(float, payload["uv_bounds"]) face = BRepBuilderAPI_MakeFace( surface, u_min, u_max, v_min, v_max, 1e-7 ).Face() else: try: outer = _wire_from_record( wire_payloads[0], surface, boundary_strategy, edge_cache, vertex_cache, ) except Exception as exc: raise ValueError( f"Cannot rebuild outer wire {wire_payloads[0]['id']}: " f"{type(exc).__name__}: {exc}" ) from exc inner_wires = [] for inner_payload in wire_payloads[1:]: try: inner_wires.append( _wire_from_record( inner_payload, surface, boundary_strategy, edge_cache, vertex_cache, ) ) except Exception as exc: raise ValueError( f"Cannot rebuild inner wire {inner_payload['id']}: " f"{type(exc).__name__}: {exc}" ) from exc def make_trimmed_face(outer_wire: Any, inner_wire_list: list[Any]) -> Any: def build( selected_outer_wire: Any, inner_wires: list[Any], inside: bool, ) -> Any: face_builder = BRepBuilderAPI_MakeFace( surface, selected_outer_wire, inside ) for inner_wire in inner_wires: face_builder.Add(inner_wire) if not face_builder.IsDone(): raise ValueError(f"Cannot rebuild face {payload['id']}") return face_builder.Face() selected_outer = outer_wire selected_wires = list(inner_wire_list) selected_inside = True face = build(selected_outer, selected_wires, selected_inside) expected_area = payload.get("area") if isinstance(expected_area, (int, float)): expected = abs(float(expected_area)) def area_error(candidate: Any) -> float: properties = GProp_GProps() BRepGProp.SurfaceProperties_s(candidate, properties) return abs(abs(float(properties.Mass())) - expected) best_error = area_error(face) outer_candidates = [ (outer_wire, True), (TopoDS.Wire_s(outer_wire.Reversed()), True), (outer_wire, False), (TopoDS.Wire_s(outer_wire.Reversed()), False), ] for candidate_outer, candidate_inside in outer_candidates[1:]: try: trial_face = build( candidate_outer, selected_wires, candidate_inside, ) except Exception: continue trial_error = area_error(trial_face) if trial_error + 1e-10 < best_error: selected_outer = candidate_outer selected_inside = candidate_inside face = trial_face best_error = trial_error for index, inner_wire in enumerate(list(selected_wires)): trial_wires = list(selected_wires) trial_wires[index] = TopoDS.Wire_s( inner_wire.Reversed() ) trial_face = build( selected_outer, trial_wires, selected_inside, ) trial_error = area_error(trial_face) if trial_error + 1e-10 < best_error: selected_wires = trial_wires face = trial_face best_error = trial_error return face face = make_trimmed_face(outer, inner_wires) BRepLib.BuildCurves3d_s(face) expected_area = payload.get("area") if isinstance(expected_area, (int, float)): expected = abs(float(expected_area)) def face_area(candidate: Any) -> float: properties = GProp_GProps() BRepGProp.SurfaceProperties_s(candidate, properties) return abs(float(properties.Mass())) original_error = abs(face_area(face) - expected) area_tolerance = max(1e-8, expected * 1e-7) if original_error > area_tolerance: edge_tolerance = max( [ float(edge.get("tolerance", 1e-7)) for wire in payload.get("wires", []) for edge in wire.get("edges", []) ] or [1e-7] ) fixer = ShapeFix_Face(face) fixer.SetPrecision(max(edge_tolerance, 1e-7)) fixer.SetMaxTolerance(max(edge_tolerance, 1e-3)) fixer.Perform() repaired = fixer.Face() if abs(face_area(repaired) - expected) < original_error: face = repaired if int(payload["orientation"]) == 1: face = TopoDS.Face_s(face.Reversed()) expected_contribution = payload.get("volume_contribution") if isinstance(expected_contribution, (int, float)): contribution_properties = GProp_GProps() BRepGProp.VolumeProperties_s(face, contribution_properties) contribution = float(contribution_properties.Mass()) expected = float(expected_contribution) if abs(-contribution - expected) + 1e-10 < abs( contribution - expected ): face = TopoDS.Face_s(face.Reversed()) return face def _axis_affine_transform( axis: list[float], pivot: list[float], axial_factor: float, radial_factor: float, ) -> gp_GTrsf: axis_length = math.sqrt(sum(float(value) ** 2 for value in axis)) if axis_length <= 1e-12: raise ValueError("Axis-affine edit requires a non-zero axis") unit = [float(value) / axis_length for value in axis] axial = float(axial_factor) radial = float(radial_factor) if ( not math.isfinite(axial) or not math.isfinite(radial) or axial <= 0 or radial <= 0 ): raise ValueError("Axis-affine factors must be finite and positive") values = [ radial * (1.0 if row == column else 0.0) + (axial - radial) * unit[row] * unit[column] for row in range(3) for column in range(3) ] matrix = gp_Mat(*values) pivot_values = [float(value) for value in pivot] transformed_pivot = [ sum( values[row * 3 + column] * pivot_values[column] for column in range(3) ) for row in range(3) ] translation = [ pivot_values[index] - transformed_pivot[index] for index in range(3) ] transform = gp_GTrsf() transform.SetVectorialPart(matrix) transform.SetTranslationPart(gp_XYZ(*translation)) return transform def _apply_edit_operations(shape: Any, payload: dict[str, Any]) -> Any: operations = payload.get("edit_operations", []) if not operations: return shape fixer = ShapeFix_Shape(shape) fixer.SetPrecision(1e-7) fixer.SetMaxTolerance(1e-5) fixer.Perform() result = fixer.Shape() for operation in operations: operation_name = operation.get("operation") if operation_name == "axis_affine_scale": transform = _axis_affine_transform( operation["axis"], operation["pivot"], float(operation["axial_factor"]), float(operation["radial_factor"]), ) builder = BRepBuilderAPI_GTransform(result, transform, True) if not builder.IsDone(): raise ValueError("Failed axis-affine scale operation") result = builder.Shape() continue if operation_name == "uniform_scale": factor = float(operation["factor"]) if not math.isfinite(factor) or factor <= 0: raise ValueError("Uniform scale factor must be finite and positive") transform = gp_Trsf() transform.SetScale( _point_value(operation["pivot"]), factor, ) builder = BRepBuilderAPI_Transform(result, transform, True) if not builder.IsDone(): raise ValueError("Failed uniform scale operation") result = builder.Shape() continue if operation_name != "enlarge_cylindrical_cut": raise ValueError( f"Unsupported SurfaceIR edit operation: {operation_name}" ) v_min, v_max = sorted(map(float, operation["v_range"])) height = v_max - v_min if height <= 1e-9: raise ValueError("Cylindrical cut edit has zero axial span") axial_margin = max(1e-6, height * 1e-7) v_min -= axial_margin height += 2.0 * axial_margin position = operation["position"] axis = _direction_value(position["axis"]) location = _point_value(position["location"]) base = gp_Pnt( location.X() + v_min * axis.X(), location.Y() + v_min * axis.Y(), location.Z() + v_min * axis.Z(), ) cutter_axis = gp_Ax2( base, axis, _direction_value(position["x_direction"]), ) cutter = BRepPrimAPI_MakeCylinder( cutter_axis, float(operation["new_radius"]), height ).Shape() algorithm = BRepAlgoAPI_Cut(result, cutter) algorithm.Build() if not algorithm.IsDone(): raise ValueError( f"Failed semantic cut for {operation.get('face_id')}" ) result = algorithm.Shape() return result def _solid_shell_from_record( shell_payload: dict[str, Any], boundary_strategy: str, edge_cache: dict[str, Any], vertex_cache: dict[str, Any], ) -> TopoDS_Shell: """Rebuild one closed shell, including one-face periodic shells.""" fallback_order = [ boundary_strategy, "analytic_uv_rect", "exact_3d", "exact_3d_pcurve", "shared_pcurve", "first_pcurve", "uv_rect_all", ] strategies = list(dict.fromkeys(fallback_order)) shell_tolerance = max( [ float(edge.get("tolerance", 1e-7)) for face in shell_payload["faces"] for wire in face["wires"] for edge in wire["edges"] ] or [1e-7] ) # The exact-pcurve representation already carries stable topology edge # identifiers. Reuse those edges across adjacent faces before resorting to # geometric sewing. This preserves the original incidence graph and avoids # turning a shared boundary into two nearly coincident edges when a # high-order curve is reconstructed independently on both faces. if boundary_strategy in {"exact_3d_pcurve", "shared_pcurve"}: shared_edge_cache = dict(edge_cache) try: shared_shell_builder = BRep_Builder() shared_shell = TopoDS_Shell() shared_shell_builder.MakeShell(shared_shell) for face_payload in shell_payload["faces"]: shared_shell_builder.Add( shared_shell, _face_from_record( face_payload, boundary_strategy, shared_edge_cache, vertex_cache, ), ) shared_shell.Closed(True) if BRepCheck_Analyzer(shared_shell).IsValid(): edge_cache.clear() edge_cache.update(shared_edge_cache) return shared_shell # Shared edges can be geometrically exact while their imported # face orientations are not accepted as a closed shell by OCCT. # Re-orient that same shared topology before falling back to # sewing. Accept the repair only when it preserves the exact # incidence counts encoded by SurfaceIR. shell_fixer = ShapeFix_Shell(shared_shell) shell_fixer.SetPrecision(max(shell_tolerance, 1e-7)) shell_fixer.SetMaxTolerance(max(shell_tolerance, 1e-3)) shell_fixer.Perform() oriented_shell = shell_fixer.Shell() expected_face_count = len(shell_payload["faces"]) expected_edge_count = len( { str(edge["topology_edge_id"]) for face in shell_payload["faces"] for wire in face["wires"] for edge in wire["edges"] } ) expected_vertex_count = len( { str(vertex_id) for face in shell_payload["faces"] for wire in face["wires"] for edge in wire["edges"] for vertex_id in edge.get("vertex_ids", []) } ) actual_counts = {} for label, shape_type in ( ("face", TopAbs_FACE), ("edge", TopAbs_EDGE), ("vertex", TopAbs_VERTEX), ): shape_map = TopTools_IndexedMapOfShape() TopExp.MapShapes_s(oriented_shell, shape_type, shape_map) actual_counts[label] = int(shape_map.Extent()) if ( BRepCheck_Analyzer(oriented_shell).IsValid() and actual_counts["face"] == expected_face_count and actual_counts["edge"] == expected_edge_count and actual_counts["vertex"] == expected_vertex_count ): edge_cache.clear() edge_cache.update(shared_edge_cache) return oriented_shell except Exception: # The independent-face sewing path below is intentionally retained # for malformed or non-manifold source topology. pass sewing = BRepBuilderAPI_Sewing( max(shell_tolerance, 1e-3), True, True, True, False ) rebuilt_faces: list[tuple[dict[str, Any], Any]] = [] for face_payload in shell_payload["faces"]: candidates = [] failures = [] expected_area = abs(float(face_payload.get("area", 0.0))) expected_centroid = face_payload.get("centroid") expected_bounds = face_payload.get("bounds") def candidate_score( candidate_face: Any, strategy_index: int ) -> tuple[Any, ...]: properties = GProp_GProps() BRepGProp.SurfaceProperties_s(candidate_face, properties) area_error = abs(abs(float(properties.Mass())) - expected_area) area_tolerance = max(1e-8, expected_area * 1e-7) centroid_error = 0.0 if ( isinstance(expected_centroid, list) and len(expected_centroid) == 3 ): center = properties.CentreOfMass() centroid_error = math.sqrt( sum( ( value - float(expected_centroid[index]) ) ** 2 for index, value in enumerate( (center.X(), center.Y(), center.Z()) ) ) ) bounds_error = 0.0 if isinstance(expected_bounds, list) and len(expected_bounds) == 6: bounds = Bnd_Box() BRepBndLib.AddOptimal_s(candidate_face, bounds) candidate_bounds = bounds.Get() bounds_error = max( abs( float(candidate_bounds[index]) - float(expected_bounds[index]) ) for index in range(6) ) return ( not BRepCheck_Analyzer(candidate_face).IsValid(), area_error > area_tolerance, centroid_error, bounds_error, area_error, strategy_index, candidate_face, ) for strategy_index, strategy in enumerate(strategies): try: face = _face_from_record( face_payload, strategy, {}, vertex_cache ) candidates.append(candidate_score(face, strategy_index)) except Exception as exc: failures.append( f"{strategy}: {type(exc).__name__}: {exc}" ) if len(face_payload.get("wires", [])) == 1: try: natural_payload = dict(face_payload) natural_payload["wires"] = [] natural_face = _face_from_record( natural_payload, boundary_strategy, {}, vertex_cache, ) natural_properties = GProp_GProps() BRepGProp.SurfaceProperties_s( natural_face, natural_properties ) natural_area_error = abs( abs(float(natural_properties.Mass())) - expected_area ) if natural_area_error <= max( 1e-8, expected_area * 1e-7 ): candidates.append( candidate_score(natural_face, len(strategies)) ) except Exception as exc: failures.append( "natural_uv_if_exact: " f"{type(exc).__name__}: {exc}" ) if not candidates: raise ValueError( f"Cannot reconstruct face {face_payload['id']}; " + " | ".join(failures) ) candidates.sort(key=lambda item: item[:-1]) selected = candidates[0][-1] rebuilt_faces.append((face_payload, selected)) sewing.Add(selected) sewing.Perform() sewed_shape = sewing.SewedShape() sewed_shells: list[Any] = [] sewed_shell_explorer = TopExp_Explorer(sewed_shape, TopAbs_SHELL) while sewed_shell_explorer.More(): sewed_shells.append( TopoDS.Shell_s(sewed_shell_explorer.Current()) ) sewed_shell_explorer.Next() if ( len(sewed_shells) == 1 and BRepCheck_Analyzer(sewed_shells[0]).IsValid() ): # Use the actual sewed topology. In contrast to ShapeFix_Shell, this # retains the edge sharing established by BRepBuilderAPI_Sewing and # does not duplicate every boundary edge. return sewed_shells[0] shell_builder = BRep_Builder() shell = TopoDS_Shell() shell_builder.MakeShell(shell) for face_payload, original_face in rebuilt_faces: modified = ( sewing.Modified(original_face) if sewing.IsModified(original_face) else original_face ) face_explorer = TopExp_Explorer(modified, TopAbs_FACE) face = ( TopoDS.Face_s(face_explorer.Current()) if face_explorer.More() else TopoDS.Face_s(modified) ) expected_contribution = face_payload.get("volume_contribution") if isinstance(expected_contribution, (int, float)): properties = GProp_GProps() BRepGProp.VolumeProperties_s(face, properties) contribution = float(properties.Mass()) expected = float(expected_contribution) if abs(-contribution - expected) + 1e-10 < abs( contribution - expected ): face = TopoDS.Face_s(face.Reversed()) shell_builder.Add(shell, face) shell.Closed(True) if not BRepCheck_Analyzer(shell).IsValid(): shell_fixer = ShapeFix_Shell(shell) shell_fixer.SetPrecision(max(shell_tolerance, 1e-7)) shell_fixer.SetMaxTolerance(max(shell_tolerance, 1e-3)) shell_fixer.Perform() repaired_shell = shell_fixer.Shell() if BRepCheck_Analyzer(repaired_shell).IsValid(): shell = repaired_shell return shell def build_surfaceir( payload: dict[str, Any], boundary_strategy: str | None = None ) -> Any: boundary_strategy = boundary_strategy or payload.get( "reconstruction_strategy", {} ).get("boundary_strategy", "exact_3d") if boundary_strategy not in { "exact_3d", "exact_3d_pcurve", "shared_pcurve", "first_pcurve", "analytic_uv_rect", "uv_rect_all", }: raise ValueError(f"Unknown boundary strategy: {boundary_strategy}") rebuilt_solids = [] rebuilt_free_shells = [] edge_cache: dict[str, Any] = {} vertex_cache: dict[str, Any] = {} vertex_builder = BRep_Builder() for vertex_payload in payload["surface_layer"].get("vertices", []): vertex = BRepBuilderAPI_MakeVertex( _point_value(vertex_payload["point"]) ).Vertex() vertex_builder.UpdateVertex( vertex, max(float(vertex_payload.get("tolerance", 1e-7)), 1e-7) ) vertex_cache[str(vertex_payload["id"])] = vertex for solid_payload in payload["surface_layer"]["solids"]: rebuilt_shells = [] for shell_payload in solid_payload["shells"]: shell = _solid_shell_from_record( shell_payload, boundary_strategy, edge_cache, vertex_cache, ) if int(shell_payload.get("orientation", 0)) == 1: shell = TopoDS.Shell_s(shell.Reversed()) rebuilt_shells.append(shell) solid_builder = BRepBuilderAPI_MakeSolid() for shell in rebuilt_shells: solid_builder.Add(shell) if not solid_builder.IsDone(): raise ValueError(f"Cannot rebuild solid {solid_payload['id']}") solid = solid_builder.Solid() if int(solid_payload.get("orientation", 0)) == 1: solid = TopoDS.Solid_s(solid.Reversed()) expected_signed_volume = solid_payload.get("signed_volume") if isinstance(expected_signed_volume, (int, float)): signed_volume = _signed_volume(solid) if abs(-signed_volume - float(expected_signed_volume)) < abs( signed_volume - float(expected_signed_volume) ): solid = TopoDS.Solid_s(solid.Reversed()) rebuilt_solids.append(solid) for shell_payload in payload["surface_layer"].get("free_shells", []): shell_builder = BRep_Builder() shell = TopoDS_Shell() shell_builder.MakeShell(shell) free_faces = [] for face_payload in shell_payload["faces"]: try: face = _face_from_record( face_payload, boundary_strategy, edge_cache, vertex_cache, ) free_faces.append(face) shell_builder.Add(shell, face) except Exception as exc: raise ValueError( f"Cannot rebuild free-shell face {face_payload['id']}: " f"{type(exc).__name__}: {exc}" ) from exc # Keep a one-face open shell as an explicit surface-model component # instead of pretending it is a closed solid. free_component = free_faces[0] if len(free_faces) == 1 else shell if int(shell_payload.get("orientation", 0)) == 1: free_component = free_component.Reversed() rebuilt_free_shells.append(free_component) components = rebuilt_solids + rebuilt_free_shells if len(components) == 1: result = components[0] else: builder = BRep_Builder() compound = TopoDS_Compound() builder.MakeCompound(compound) for component in components: builder.Add(compound, component) result = compound return _apply_edit_operations(result, payload) def extract_surfaceir(step_path: Path) -> dict[str, Any]: source_bytes = step_path.read_bytes() reader = STEPControl_Reader() if reader.ReadFile(str(step_path)) != IFSelect_RetDone: raise ValueError("OCCT failed to read STEP") reader.TransferRoots() shape = reader.OneShape() topology = ( {"vertices": [], "solids": [], "free_shells": []} if shape.IsNull() else _topology_records(shape) ) solids = topology["solids"] free_shells = topology.get("free_shells", []) faces = _surfaceir_faces(topology) surface_types = sorted( {face["surface"]["type"] for face in faces} ) curve_types = sorted( { edge["curve"]["type"] for face in faces for wire in face["wires"] for edge in wire["edges"] } ) digest = hashlib.sha256(source_bytes).hexdigest() modification_levels = ["analytic_surface"] if "bspline" in surface_types or "bspline" in curve_types: modification_levels.append("spline_control") semantic_layer = _cylinder_semantics(topology) if (solids or free_shells) and topology["vertices"]: points = [ [float(value) for value in vertex["point"]] for vertex in topology["vertices"] ] scale_pivot = [ _number( ( min(point[axis_index] for point in points) + max(point[axis_index] for point in points) ) / 2.0 ) for axis_index in range(3) ] semantic_layer["datums"]["overall_scale_pivot"] = { "point": scale_pivot, "inference": "surfaceir_vertex_bounds_center", "confidence": 1.0, } semantic_layer["parameters"]["overall_scale"] = { "value": 1.0, "unit": "ratio", "editable": False, "semantic_role": "overall_scale", "inference": "universal_source_independent_transform", "confidence": 1.0, "pivot_datum": "overall_scale_pivot", "edit_state": "executable_uniform_scale_binding", } semantic_layer["features"].append( { "id": "overall_scale_control", "operation": "uniform_scale", "parameter": "overall_scale", "pivot_datum": "overall_scale_pivot", "inference": "canonical_edit_fallback", "original_history_recovered": False, "confidence": 1.0, } ) if len(semantic_layer["parameters"]) == 1: for axis_name, axis_index in zip(("x", "y", "z"), range(3)): extent = ( max(point[axis_index] for point in points) - min(point[axis_index] for point in points) ) if extent <= 1e-9: continue parameter_name = f"overall_size_{axis_name}" semantic_layer["parameters"][parameter_name] = { "value": _number(extent), "unit": "mm", "editable": False, "semantic_role": parameter_name, "inference": "canonical_global_axis_bounds_extent", "confidence": 1.0, "pivot_datum": "overall_scale_pivot", "axis": axis_name, "edit_state": "executable_axis_affine_binding", } semantic_layer["features"].append( { "id": f"{parameter_name}_control", "operation": "axis_affine_scale", "parameter": parameter_name, "pivot_datum": "overall_scale_pivot", "axis": axis_name, "inference": "canonical_envelope_dimension", "original_history_recovered": False, "confidence": 1.0, } ) executable_semantic_parameters = sorted( name for name, parameter in semantic_layer["parameters"].items() if parameter.get("edit_state") in { "executable_enlarge_cut_binding", "executable_axis_affine_binding", "executable_uniform_scale_binding", } ) return { "schema_version": "3.0", "designir_kind": "independent_parametric_cad", "model_id": f"surfaceir_{digest[:16]}", "units": "mm", "document_status": ( "geometry_present" if solids or free_shells else "empty_geometry" ), "reconstruction_mode": "surface_parametric", "semantic_layer": semantic_layer, "surface_layer": { "vertices": topology["vertices"], "solids": solids, "free_shells": free_shells, "surface_vocabulary": surface_types, "curve_vocabulary": curve_types, }, "edit_interface": { "semantic_parameters": executable_semantic_parameters, "surface_parameter_groups": [ "surface_layer.solids.*.shells.*.faces.*.surface", "surface_layer.solids.*.shells.*.faces.*.wires.*.edges.*.curve", ], "modification_levels": modification_levels, }, "validation_contract": { "source_independence": True, "geometry_checks": [ "solid_count", "volume", "center_of_mass", "bounding_box", "symmetric_difference", "surface_distance", ], "edit_checks": [ "edited_parameter_changes_geometry", "unselected_parameter_groups_remain_stable", "result_is_valid_manifold", ], }, } def write_json(path: Path, payload: dict[str, Any]) -> None: if path.name.endswith(".designir.json"): write_designir(path, payload) return path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8", ) def extract_folder(input_dir: Path, output_dir: Path) -> dict[str, Any]: sources = sorted( [ path for path in input_dir.iterdir() if path.is_file() and path.suffix.lower() in {".step", ".stp"} ] ) rows = [] surface_counts: Counter[str] = Counter() curve_counts: Counter[str] = Counter() for source in sources: output = output_dir / f"{source.stem}.designir.json" try: payload = extract_surfaceir(source) write_json(output, payload) surface_counts.update(payload["surface_layer"]["surface_vocabulary"]) curve_counts.update(payload["surface_layer"]["curve_vocabulary"]) rows.append( { "source_name": source.name, "model_id": payload["model_id"], "output_name": output.name, "state": "extracted", "solid_count": len(payload["surface_layer"]["solids"]), "free_shell_count": len( payload["surface_layer"].get("free_shells", []) ), "document_status": payload["document_status"], "surface_vocabulary": payload["surface_layer"][ "surface_vocabulary" ], "curve_vocabulary": payload["surface_layer"][ "curve_vocabulary" ], "byte_count": output.stat().st_size, } ) except Exception as exc: rows.append( { "source_name": source.name, "state": "failed", "error": f"{type(exc).__name__}: {exc}", } ) report = { "schema_version": "1.0", "report_kind": "designir_3_surface_extraction", "input_count": len(sources), "extracted_count": sum(row["state"] == "extracted" for row in rows), "failed_count": sum(row["state"] == "failed" for row in rows), "case_presence_by_surface_type": dict(sorted(surface_counts.items())), "case_presence_by_curve_type": dict(sorted(curve_counts.items())), "total_json_bytes": sum(row.get("byte_count", 0) for row in rows), "rows": rows, } write_json(output_dir / "manifest.json", report) return report def augment_overall_scale(designir_dir: Path, output: Path) -> dict[str, Any]: """Add a source-independent uniform-scale fallback to existing DesignIR.""" manifest = json.loads( (designir_dir / "manifest.json").read_text(encoding="utf-8") ) rows = [] for manifest_row in manifest["rows"]: if manifest_row["state"] != "extracted": continue designir_path = designir_dir / manifest_row["output_name"] payload = read_designir(designir_path) vertices = payload.get("surface_layer", {}).get("vertices", []) if ( payload.get("document_status") != "geometry_present" or not vertices ): rows.append( { "source_name": manifest_row["source_name"], "state": "empty_geometry_skipped", } ) continue points = [ [float(value) for value in vertex["point"]] for vertex in vertices ] pivot = [ _number( ( min(point[index] for point in points) + max(point[index] for point in points) ) / 2.0 ) for index in range(3) ] semantic_layer = payload.setdefault("semantic_layer", {}) datums = semantic_layer.setdefault("datums", {}) parameters = semantic_layer.setdefault("parameters", {}) features = semantic_layer.setdefault("features", []) datums["overall_scale_pivot"] = { "point": pivot, "inference": "surfaceir_vertex_bounds_center", "confidence": 1.0, } parameters["overall_scale"] = { "value": 1.0, "unit": "ratio", "editable": False, "semantic_role": "overall_scale", "inference": "universal_source_independent_transform", "confidence": 1.0, "pivot_datum": "overall_scale_pivot", "edit_state": "executable_uniform_scale_binding", } features[:] = [ feature for feature in features if feature.get("id") != "overall_scale_control" ] features.append( { "id": "overall_scale_control", "operation": "uniform_scale", "parameter": "overall_scale", "pivot_datum": "overall_scale_pivot", "inference": "canonical_edit_fallback", "original_history_recovered": False, "confidence": 1.0, } ) edit_parameters = payload.setdefault("edit_interface", {}).setdefault( "semantic_parameters", [] ) if "overall_scale" not in edit_parameters: edit_parameters.append("overall_scale") edit_parameters.sort() write_json(designir_path, payload) rows.append( { "source_name": manifest_row["source_name"], "state": "overall_scale_added", "pivot": pivot, } ) report = { "schema_version": "1.0", "report_kind": "designir_3_overall_scale_augmentation", "case_count": len(rows), "augmented_count": sum( row["state"] == "overall_scale_added" for row in rows ), "skipped_count": sum( row["state"] != "overall_scale_added" for row in rows ), "rows": rows, } write_json(output, report) return report def augment_axis_affine_bindings( designir_dir: Path, output: Path ) -> dict[str, Any]: """Bind primary-axis dimensions only where no feature edit is validated.""" manifest = json.loads( (designir_dir / "manifest.json").read_text(encoding="utf-8") ) rows = [] for manifest_row in manifest["rows"]: if manifest_row["state"] != "extracted": continue designir_path = designir_dir / manifest_row["output_name"] payload = read_designir(designir_path) if payload.get("document_status") != "geometry_present": continue parameters = ( payload.get("semantic_layer", {}).get("parameters", {}) ) has_validated_feature_edit = any( name != "overall_scale" and parameter.get("edit_state") == "validated_executable_binding" for name, parameter in parameters.items() ) bound = [] if not has_validated_feature_edit: for name, parameter in parameters.items(): if parameter.get("semantic_role") not in { "outer_diameter", "body_length", "plate_thickness", }: continue if not parameter.get("affected_face_ids"): continue parameter["editable"] = False parameter["edit_state"] = "executable_axis_affine_binding" bound.append(name) edit_parameters = payload.setdefault("edit_interface", {}).setdefault( "semantic_parameters", [] ) for name in bound: if name not in edit_parameters: edit_parameters.append(name) edit_parameters.sort() if bound: write_json(designir_path, payload) rows.append( { "source_name": manifest_row["source_name"], "state": ( "axis_affine_bound" if bound else "not_applicable" ), "parameters": sorted(bound), } ) report = { "schema_version": "1.0", "report_kind": "designir_3_axis_affine_binding_augmentation", "case_count": len(rows), "applicable_case_count": sum( row["state"] == "axis_affine_bound" for row in rows ), "parameter_count": sum(len(row["parameters"]) for row in rows), "rows": rows, } write_json(output, report) return report def augment_directional_size_bindings( designir_dir: Path, output: Path ) -> dict[str, Any]: """Add canonical global-axis dimensions to remaining scale-only models.""" manifest = json.loads( (designir_dir / "manifest.json").read_text(encoding="utf-8") ) rows = [] for manifest_row in manifest["rows"]: if manifest_row["state"] != "extracted": continue designir_path = designir_dir / manifest_row["output_name"] payload = read_designir(designir_path) if payload.get("document_status") != "geometry_present": continue semantic_layer = payload.get("semantic_layer", {}) parameters = semantic_layer.get("parameters", {}) has_validated_named_edit = any( name != "overall_scale" and parameter.get("edit_state") == "validated_executable_binding" for name, parameter in parameters.items() ) bound = [] if not has_validated_named_edit: vertices = payload["surface_layer"]["vertices"] points = [ [float(value) for value in vertex["point"]] for vertex in vertices ] pivot = [ _number( ( min(point[index] for point in points) + max(point[index] for point in points) ) / 2.0 ) for index in range(3) ] semantic_layer.setdefault("datums", {})[ "directional_size_pivot" ] = { "point": pivot, "inference": "surfaceir_vertex_bounds_center", "confidence": 1.0, } features = semantic_layer.setdefault("features", []) for axis_name, axis_index in zip(("x", "y", "z"), range(3)): extent = ( max(point[axis_index] for point in points) - min(point[axis_index] for point in points) ) if extent <= 1e-9: continue parameter_name = f"overall_size_{axis_name}" parameters[parameter_name] = { "value": _number(extent), "unit": "mm", "editable": False, "semantic_role": parameter_name, "inference": "canonical_global_axis_bounds_extent", "confidence": 1.0, "pivot_datum": "directional_size_pivot", "axis": axis_name, "edit_state": "executable_axis_affine_binding", } features[:] = [ feature for feature in features if feature.get("id") != f"{parameter_name}_control" ] features.append( { "id": f"{parameter_name}_control", "operation": "axis_affine_scale", "parameter": parameter_name, "pivot_datum": "directional_size_pivot", "axis": axis_name, "inference": "canonical_envelope_dimension", "original_history_recovered": False, "confidence": 1.0, } ) bound.append(parameter_name) edit_parameters = payload.setdefault("edit_interface", {}).setdefault( "semantic_parameters", [] ) for name in bound: if name not in edit_parameters: edit_parameters.append(name) edit_parameters.sort() if bound: write_json(designir_path, payload) rows.append( { "source_name": manifest_row["source_name"], "state": ( "directional_sizes_bound" if bound else "not_applicable" ), "parameters": sorted(bound), } ) report = { "schema_version": "1.0", "report_kind": "designir_3_directional_size_binding_augmentation", "case_count": len(rows), "applicable_case_count": sum( row["state"] == "directional_sizes_bound" for row in rows ), "parameter_count": sum(len(row["parameters"]) for row in rows), "rows": rows, } write_json(output, report) return report def _facts_from_shape(shape: Any) -> dict[str, Any]: wrapped = shape.wrapped if hasattr(shape, "wrapped") else shape bounds = Bnd_Box() BRepBndLib.AddOptimal_s(wrapped, bounds) if bounds.IsVoid(): x_min = y_min = z_min = x_max = y_max = z_max = 0.0 else: ( x_min, y_min, z_min, x_max, y_max, z_max, ) = bounds.Get() properties = GProp_GProps() BRepGProp.VolumeProperties_s(wrapped, properties) volume_value = float(properties.Mass()) surface_properties = GProp_GProps() BRepGProp.SurfaceProperties_s(wrapped, surface_properties) surface_area = float(surface_properties.Mass()) if abs(float(properties.Mass())) <= 1e-12: properties = surface_properties center = properties.CentreOfMass() counts = {} for name, shape_type in { "solid_count": TopAbs_SOLID, "face_count": TopAbs_FACE, "edge_count": TopAbs_EDGE, "vertex_count": TopAbs_VERTEX, }.items(): shape_map = TopTools_IndexedMapOfShape() TopExp.MapShapes_s(wrapped, shape_type, shape_map) counts[name] = shape_map.Extent() solid_subshape_counts = { "solid_face_count": 0, "solid_edge_count": 0, "solid_vertex_count": 0, } solid_explorer = TopExp_Explorer(wrapped, TopAbs_SOLID) while solid_explorer.More(): solid = TopoDS.Solid_s(solid_explorer.Current()) for name, shape_type in { "solid_face_count": TopAbs_FACE, "solid_edge_count": TopAbs_EDGE, "solid_vertex_count": TopAbs_VERTEX, }.items(): shape_map = TopTools_IndexedMapOfShape() TopExp.MapShapes_s(solid, shape_type, shape_map) solid_subshape_counts[name] += shape_map.Extent() solid_explorer.Next() return { **counts, **solid_subshape_counts, "volume": float(volume_value), "surface_area": surface_area, "center_of_mass": [float(center.X()), float(center.Y()), float(center.Z())], "bounds_min": [float(x_min), float(y_min), float(z_min)], "bounds_max": [float(x_max), float(y_max), float(z_max)], "valid": bool(BRepCheck_Analyzer(wrapped).IsValid()), } def _shape_facts(path: Path) -> tuple[Any, dict[str, Any]]: try: shape = import_step(str(path)) return shape, _facts_from_shape(shape) except AssertionError: reader = STEPControl_Reader() if reader.ReadFile(str(path)) != IFSelect_RetDone: raise ValueError(f"OCCT failed to read STEP facts: {path}") reader.TransferRoots() wrapped = reader.OneShape() if wrapped.IsNull(): builder = BRep_Builder() compound = TopoDS_Compound() builder.MakeCompound(compound) wrapped = compound shape = wrapped return shape, _facts_from_shape(shape) def _max_error(first: list[float], second: list[float]) -> float: return max(abs(a - b) for a, b in zip(first, second)) def _solid_geometry(shape: Any) -> Any: """Return only solid components for solid-reconstruction comparison.""" wrapped = shape.wrapped if hasattr(shape, "wrapped") else shape solids = [] explorer = TopExp_Explorer(wrapped, TopAbs_SOLID) while explorer.More(): solids.append(TopoDS.Solid_s(explorer.Current())) explorer.Next() if len(solids) == 1: return solids[0] builder = BRep_Builder() compound = TopoDS_Compound() builder.MakeCompound(compound) for solid in solids: builder.Add(compound, solid) return compound def _robust_cut_volume( first_shape: Any, second_shape: Any, fuzzy_tolerance: float = 1e-9 ) -> float: first = ( first_shape.wrapped if hasattr(first_shape, "wrapped") else first_shape ) second = ( second_shape.wrapped if hasattr(second_shape, "wrapped") else second_shape ) volumes = [] for tool in (second, second.Reversed()): operation = BRepAlgoAPI_Cut(first, tool) operation.SetFuzzyValue(fuzzy_tolerance) operation.Build() if not operation.IsDone(): continue properties = GProp_GProps() BRepGProp.VolumeProperties_s(operation.Shape(), properties) volumes.append(abs(float(properties.Mass()))) if not volumes: raise ValueError("Bidirectional B-Rep Boolean difference failed") return min(volumes) def _sample_shape_points(shape: Any) -> list[gp_Pnt]: wrapped = shape.wrapped if hasattr(shape, "wrapped") else shape points: list[gp_Pnt] = [] vertex_map = TopTools_IndexedMapOfShape() TopExp.MapShapes_s(wrapped, TopAbs_VERTEX, vertex_map) for index in range(1, vertex_map.Extent() + 1): points.append( BRep_Tool.Pnt_s(TopoDS.Vertex_s(vertex_map.FindKey(index))) ) edge_map = TopTools_IndexedMapOfShape() TopExp.MapShapes_s(wrapped, TopAbs_EDGE, edge_map) for index in range(1, edge_map.Extent() + 1): edge = TopoDS.Edge_s(edge_map.FindKey(index)) adaptor = BRepAdaptor_Curve(edge) first = float(adaptor.FirstParameter()) last = float(adaptor.LastParameter()) if not math.isfinite(first) or not math.isfinite(last): continue for fraction in (0.25, 0.5, 0.75): points.append(adaptor.Value(first + fraction * (last - first))) face_map = TopTools_IndexedMapOfShape() TopExp.MapShapes_s(wrapped, TopAbs_FACE, face_map) for index in range(1, face_map.Extent() + 1): face = TopoDS.Face_s(face_map.FindKey(index)) u_min, u_max, v_min, v_max = map( float, BRepTools.UVBounds_s(face) ) if not all( math.isfinite(value) for value in (u_min, u_max, v_min, v_max) ): continue adaptor = BRepAdaptor_Surface(face) for u_fraction in (0.25, 0.5, 0.75): for v_fraction in (0.25, 0.5, 0.75): u_value = u_min + u_fraction * (u_max - u_min) v_value = v_min + v_fraction * (v_max - v_min) classifier = BRepClass_FaceClassifier( face, gp_Pnt2d(u_value, v_value), 1e-7 ) if classifier.State() in {TopAbs_IN, TopAbs_ON}: points.append(adaptor.Value(u_value, v_value)) return points def _directed_sample_distance( source_shape: Any, target_shape: Any ) -> tuple[float, int]: target = ( target_shape.wrapped if hasattr(target_shape, "wrapped") else target_shape ) distance = BRepExtrema_DistShapeShape() distance.LoadS2(target) maximum = 0.0 points = _sample_shape_points(source_shape) for point in points: vertex = BRepBuilderAPI_MakeVertex(point).Vertex() distance.LoadS1(vertex) distance.Perform() if not distance.IsDone(): raise ValueError("Surface sample distance evaluation failed") maximum = max(maximum, float(distance.Value())) return maximum, len(points) def validate_folder( input_dir: Path, designir_dir: Path, rebuilt_dir: Path, limit: int | None, offset: int, boolean_check: bool, boundary_strategy: str, ) -> dict[str, Any]: manifest = json.loads( (designir_dir / "manifest.json").read_text(encoding="utf-8") ) rows = [row for row in manifest["rows"] if row["state"] == "extracted"] rows = rows[offset:] if limit is not None: rows = rows[:limit] results = [] for row in rows: source = input_dir / row["source_name"] designir_path = designir_dir / row["output_name"] rebuilt = rebuilt_dir / f"{Path(row['source_name']).stem}.step" payload = read_designir(designir_path) if payload.get("document_status") == "empty_geometry": results.append( { "source_name": row["source_name"], "state": "empty_source_no_geometry", "geometry_pass": False, "usable_reconstruction": False, "quality_tier": "not_applicable", "surface_vocabulary": [], "curve_vocabulary": [], "reason": ( "The source STEP document contains no faces or solids; " "there is no geometry to reconstruct or validate." ), } ) continue try: shape = build_surfaceir(payload, boundary_strategy) export_step_shape(shape, rebuilt) teacher_shape, teacher = _shape_facts(source) rebuilt_shape, rebuilt_facts = _shape_facts(rebuilt) volume_base = max(abs(teacher["volume"]), 1e-12) relative_volume_error = ( abs(teacher["volume"] - rebuilt_facts["volume"]) / volume_base ) area_base = max(abs(teacher["surface_area"]), 1e-12) relative_surface_area_error = abs( teacher["surface_area"] - rebuilt_facts["surface_area"] ) / area_base center_error = math.dist( teacher["center_of_mass"], rebuilt_facts["center_of_mass"] ) bounds_error = max( _max_error(teacher["bounds_min"], rebuilt_facts["bounds_min"]), _max_error(teacher["bounds_max"], rebuilt_facts["bounds_max"]), ) topology_match = all( teacher[name] == rebuilt_facts[name] for name in [ "solid_count", "solid_face_count", "solid_edge_count", ] ) vertex_count_delta = ( rebuilt_facts["vertex_count"] - teacher["vertex_count"] ) symmetric_difference_ratio = None bidirectional_sample_distance = None sample_point_count = None if boolean_check and teacher["solid_count"] > 0: try: symmetric_difference_ratio = ( _robust_cut_volume( teacher_shape, rebuilt_shape ) + _robust_cut_volume( rebuilt_shape, teacher_shape ) ) / volume_base except Exception: symmetric_difference_ratio = math.inf if symmetric_difference_ratio > 1e-8: teacher_distance_shape = _solid_geometry(teacher_shape) rebuilt_distance_shape = _solid_geometry(rebuilt_shape) forward_distance, forward_count = ( _directed_sample_distance( teacher_distance_shape, rebuilt_distance_shape, ) ) reverse_distance, reverse_count = ( _directed_sample_distance( rebuilt_distance_shape, teacher_distance_shape, ) ) bidirectional_sample_distance = max( forward_distance, reverse_distance ) sample_point_count = forward_count + reverse_count geometry_pass = ( rebuilt_facts["valid"] and topology_match and ( relative_volume_error <= 1e-7 if teacher["solid_count"] > 0 else relative_surface_area_error <= 1e-7 ) and center_error <= 1e-6 and bounds_error <= 1e-6 and ( symmetric_difference_ratio is None or symmetric_difference_ratio <= 1e-8 or ( bidirectional_sample_distance is not None and bidirectional_sample_distance <= 1e-4 ) ) ) usable_reconstruction = ( rebuilt_facts["valid"] and topology_match and relative_volume_error <= 2e-3 and relative_surface_area_error <= 1e-3 and center_error <= 2e-2 and bounds_error <= 1e-5 and ( bidirectional_sample_distance is None or bidirectional_sample_distance <= 1e-1 ) ) results.append( { "source_name": row["source_name"], "state": "validated" if geometry_pass else "geometry_mismatch", "geometry_pass": geometry_pass, "usable_reconstruction": usable_reconstruction, "quality_tier": ( "surface_geometry_accepted_pending_brep" if geometry_pass else "review_required" if usable_reconstruction else "rejected" ), "surface_vocabulary": payload["surface_layer"][ "surface_vocabulary" ], "curve_vocabulary": payload["surface_layer"]["curve_vocabulary"], "teacher": teacher, "rebuilt": rebuilt_facts, "topology_match": topology_match, "vertex_count_delta": vertex_count_delta, "relative_volume_error": relative_volume_error, "relative_surface_area_error": relative_surface_area_error, "center_of_mass_error": center_error, "bounds_error": bounds_error, "symmetric_difference_ratio": symmetric_difference_ratio, "bidirectional_sample_distance": ( bidirectional_sample_distance ), "surface_sample_point_count": sample_point_count, } ) except Exception as exc: results.append( { "source_name": row["source_name"], "state": "rebuild_failed", "geometry_pass": False, "usable_reconstruction": False, "quality_tier": "rejected", "surface_vocabulary": payload["surface_layer"][ "surface_vocabulary" ], "curve_vocabulary": payload["surface_layer"]["curve_vocabulary"], "error": f"{type(exc).__name__}: {exc}", } ) state_counts = Counter(row["state"] for row in results) applicable_results = [ row for row in results if row["state"] != "empty_source_no_geometry" ] report = { "schema_version": "1.0", "report_kind": "designir_3_independent_rebuild_validation", "case_count": len(results), "applicable_case_count": len(applicable_results), "empty_source_count": len(results) - len(applicable_results), "offset": offset, "boundary_strategy": boundary_strategy, "geometry_pass_count": sum( row["geometry_pass"] for row in applicable_results ), "geometry_fail_count": sum( not row["geometry_pass"] for row in applicable_results ), "usable_reconstruction_count": sum( row.get("usable_reconstruction", False) for row in results ), "state_counts": dict(sorted(state_counts.items())), "thresholds": { "relative_volume_error_max": 1e-7, "relative_surface_area_error_max": 1e-7, "center_of_mass_error_mm_max": 1e-6, "bounds_error_mm_max": 1e-6, "symmetric_difference_ratio_max": 1e-8 if boolean_check else None, "bidirectional_sample_distance_mm_max": ( 1e-4 if boolean_check else None ), "review_required": { "relative_volume_error_max": 2e-3, "relative_surface_area_error_max": 1e-3, "center_of_mass_error_mm_max": 2e-2, "bounds_error_mm_max": 1e-5, "bidirectional_solid_sample_distance_mm_max": ( 1e-1 if boolean_check else None ), }, "topology_must_match": True, "vertex_count_normalization_is_diagnostic": True, }, "rows": results, } write_json(rebuilt_dir / "validation-report.json", report) return report def reclassify_validation_report( report_path: Path, output: Path ) -> dict[str, Any]: """Reapply the current deterministic thresholds to measured geometry.""" report = json.loads(report_path.read_text(encoding="utf-8")) for row in report["rows"]: if "teacher" not in row or "rebuilt" not in row: row["geometry_pass"] = False continue teacher = row["teacher"] rebuilt = row["rebuilt"] row["topology_match"] = all( int(teacher[name]) == int(rebuilt[name]) for name in ("solid_count", "face_count", "edge_count") ) row["vertex_count_delta"] = ( int(rebuilt["vertex_count"]) - int(teacher["vertex_count"]) ) metric_pass = ( float(row["relative_volume_error"]) <= 1e-7 if int(teacher["solid_count"]) > 0 else float(row["relative_surface_area_error"]) <= 1e-7 ) symmetric_difference = row.get("symmetric_difference_ratio") sample_distance = row.get("bidirectional_sample_distance") distance_pass = ( symmetric_difference is None or float(symmetric_difference) <= 1e-8 or ( sample_distance is not None and float(sample_distance) <= 1e-4 ) ) row["geometry_pass"] = bool( rebuilt["valid"] and row["topology_match"] and metric_pass and float(row["center_of_mass_error"]) <= 1e-6 and float(row["bounds_error"]) <= 1e-6 and distance_pass ) row["state"] = ( "validated" if row["geometry_pass"] else "geometry_mismatch" ) state_counts = Counter(row["state"] for row in report["rows"]) report["geometry_pass_count"] = sum( bool(row["geometry_pass"]) for row in report["rows"] ) report["geometry_fail_count"] = sum( not bool(row["geometry_pass"]) for row in report["rows"] ) report["state_counts"] = dict(sorted(state_counts.items())) report["thresholds"] = { "relative_volume_error_max": 1e-7, "relative_surface_area_error_max": 1e-7, "center_of_mass_error_mm_max": 1e-6, "bounds_error_mm_max": 1e-6, "symmetric_difference_ratio_max": 1e-8, "bidirectional_sample_distance_mm_max": 1e-4, "topology_must_match": True, "vertex_count_normalization_is_diagnostic": True, } report["report_kind"] = ( "designir_3_authoritative_reclassified_rebuild_validation" ) report["measurement_source_report"] = str(report_path) write_json(output, report) return report def select_validated_strategies( designir_dir: Path, primary_report_path: Path, fallback_report_paths: list[Path], output: Path, ) -> dict[str, Any]: primary = json.loads(primary_report_path.read_text(encoding="utf-8")) fallbacks = [ json.loads(path.read_text(encoding="utf-8")) for path in fallback_report_paths ] if any( primary["case_count"] != fallback["case_count"] for fallback in fallbacks ): raise ValueError("Strategy reports must contain the same number of cases") rows = [] for index, primary_row in enumerate(primary["rows"]): fallback_rows = [fallback["rows"][index] for fallback in fallbacks] if any( primary_row["source_name"] != fallback_row["source_name"] for fallback_row in fallback_rows ): raise ValueError("Strategy reports are not paired by source") if primary_row["geometry_pass"]: selected = primary["boundary_strategy"] selected_row = primary_row status = "geometry_validated" elif any(row["geometry_pass"] for row in fallback_rows): selected_index = next( position for position, row in enumerate(fallback_rows) if row["geometry_pass"] ) selected = fallbacks[selected_index]["boundary_strategy"] selected_row = fallback_rows[selected_index] status = "geometry_validated" else: selected = primary["boundary_strategy"] selected_row = primary_row status = "unresolved" designir_path = ( designir_dir / f"{Path(primary_row['source_name']).stem}.designir.json" ) payload = read_designir(designir_path) payload["reconstruction_strategy"] = { "boundary_strategy": selected, "selection_status": status, } write_json(designir_path, payload) rows.append( { "source_name": primary_row["source_name"], "selected_boundary_strategy": selected, "selection_status": status, "geometry_pass": bool(selected_row["geometry_pass"]), "selected_state": selected_row["state"], "primary_geometry_pass": primary_row["geometry_pass"], "fallback_geometry_passes": [ row["geometry_pass"] for row in fallback_rows ], } ) report = { "schema_version": "1.0", "report_kind": "designir_3_validated_strategy_selection", "case_count": len(rows), "geometry_validated_count": sum( row["selection_status"] == "geometry_validated" for row in rows ), "geometry_pass_count": sum(row["geometry_pass"] for row in rows), "geometry_fail_count": sum(not row["geometry_pass"] for row in rows), "unresolved_count": sum( row["selection_status"] == "unresolved" for row in rows ), "selected_strategy_counts": dict( sorted(Counter(row["selected_boundary_strategy"] for row in rows).items()) ), "rows": rows, } write_json(output, report) return report def export_step_shape(shape: Any, output: Path) -> None: writer = STEPControl_Writer() transfer_status = writer.Transfer(shape, STEPControl_AsIs) if transfer_status != IFSelect_RetDone: raise ValueError("OCCT failed to transfer rebuilt SurfaceIR shape") output.parent.mkdir(parents=True, exist_ok=True) write_status = writer.Write(str(output)) if write_status != IFSelect_RetDone: raise ValueError("OCCT failed to write rebuilt SurfaceIR STEP") def _observe_internal_cylinders_from_shape(shape: Any) -> list[dict[str, Any]]: wrapped = shape.wrapped if hasattr(shape, "wrapped") else shape observations = [] explorer = TopExp_Explorer(wrapped, TopAbs_FACE) index = 0 while explorer.More(): index += 1 face = TopoDS.Face_s(explorer.Current()) adaptor = BRepAdaptor_Surface(face) if adaptor.GetType() == GeomAbs_Cylinder and int(face.Orientation()) == 1: cylinder = adaptor.Cylinder() axis, anchor = _normalized_axis( { "axis": _direction(cylinder.Position().Direction()), "location": _point(cylinder.Position().Location()), } ) observations.append( { "face_id": f"observed_face_{index}", "axis": axis, "anchor": anchor, "radius": float(cylinder.Radius()), } ) explorer.Next() return observations def _match_edit_cylinders( edited: dict[str, Any], observed_cylinders: list[dict[str, Any]] ) -> list[dict[str, Any]]: target_rows = [] unmatched = list(observed_cylinders) for operation in edited.get("edit_operations", []): expected_axis, expected_anchor = _normalized_axis(operation["position"]) expected_radius = float(operation["new_radius"]) radius_tolerance = max(1e-5, expected_radius * 1e-5) axis_tolerance = 1e-5 match_index = next( ( index for index, cylinder in enumerate(unmatched) if _parallel(expected_axis, cylinder["axis"]) and _distance(expected_anchor, cylinder["anchor"]) <= axis_tolerance and abs(cylinder["radius"] - expected_radius) <= radius_tolerance ), None, ) observed_radius = None observed_face_id = None if match_index is not None: match = unmatched.pop(match_index) observed_radius = match["radius"] observed_face_id = match["face_id"] target_rows.append( { "source_face_id": operation["face_id"], "expected_radius": expected_radius, "observed_radius": observed_radius, "observed_face_id": observed_face_id, "pass": match_index is not None, } ) return target_rows def _observe_axis_affine_target( payload: dict[str, Any], operation: dict[str, Any], expected_value: float, ) -> dict[str, Any]: vertex_points = { vertex["id"]: [float(value) for value in vertex["point"]] for vertex in payload["surface_layer"]["vertices"] } axis = [float(value) for value in operation["axis"]] pivot = [float(value) for value in operation["pivot"]] role = operation["semantic_role"] if role in {"overall_size_x", "overall_size_y", "overall_size_z"}: coordinates = [ _dot( [ point[index] - pivot[index] for index in range(3) ], axis, ) for point in vertex_points.values() ] observed_value = max(coordinates) - min(coordinates) return { "semantic_role": role, "observed_value": _number(observed_value), "target_measure": "bounds_extent", "observed_vertex_count": len(coordinates), } candidates = [] for face in _surfaceir_faces(payload["surface_layer"]): vertex_ids = { vertex_id for wire in face["wires"] for edge in wire["edges"] for vertex_id in edge.get("vertex_ids", []) } points = [ vertex_points[vertex_id] for vertex_id in vertex_ids if vertex_id in vertex_points ] if not points: continue axial_coordinates = [] radial_distances = [] for point in points: delta = [ point[index] - pivot[index] for index in range(3) ] axial = _dot(delta, axis) axial_coordinates.append(axial) radial_distances.append( math.sqrt( sum( ( delta[index] - axial * axis[index] ) ** 2 for index in range(3) ) ) ) radial_mean = sum(radial_distances) / len(radial_distances) radial_spread = max( abs(value - radial_mean) for value in radial_distances ) radial_tolerance = max(1e-5, radial_mean * 1e-5) if radial_spread > radial_tolerance: continue observed_value = ( 2.0 * radial_mean if role == "outer_diameter" else max(axial_coordinates) - min(axial_coordinates) ) if observed_value <= 1e-9: continue candidates.append( { "face_id": face["id"], "observed_value": observed_value, "boundary_vertex_count": len(points), "absolute_error": abs(observed_value - expected_value), } ) if not candidates: return { "semantic_role": role, "observed_value": None, "reason": "no_axis_aligned_face_measurement", } match = min(candidates, key=lambda candidate: candidate["absolute_error"]) return { "semantic_role": role, "observed_value": _number(match["observed_value"]), "observed_face_id": match["face_id"], "source_target_face_id": operation["target_face_id"], "boundary_vertex_count": match["boundary_vertex_count"], "candidate_count": len(candidates), } def validate_semantic_edit( teacher_step: Path, designir_path: Path, parameter_name: str, value: float, output_dir: Path, ) -> dict[str, Any]: """Acceptance-Agent check for one isolated semantic perturbation.""" payload = read_designir(designir_path) edited = apply_semantic_parameter(payload, parameter_name, value) output_dir.mkdir(parents=True, exist_ok=True) edited_ir_path = output_dir / f"{designir_path.stem}.edited.json" edited_step_path = output_dir / f"{designir_path.stem}.edited.step" checks: dict[str, Any] = {} try: teacher_shape, teacher_facts = _shape_facts(teacher_step) operation_names = [ operation.get("operation") for operation in edited.get("edit_operations", []) ] is_uniform_scale = operation_names == ["uniform_scale"] is_axis_affine = operation_names == ["axis_affine_scale"] is_scale_transform = is_uniform_scale or is_axis_affine scale_operation = ( edited["edit_operations"][0] if is_scale_transform else None ) preferred = edited.get("reconstruction_strategy", {}).get( "boundary_strategy", "analytic_uv_rect" ) strategies = list( dict.fromkeys( [ preferred, "exact_3d_pcurve", "exact_3d", "analytic_uv_rect", "first_pcurve", "uv_rect_all", ] ) ) strategy_attempts = [] selected_result = None expected_removed_volume = ( 0.0 if is_scale_transform else sum( math.pi * ( float(operation["new_radius"]) ** 2 - float(operation["old_radius"]) ** 2 ) * abs( float(operation["v_range"][1]) - float(operation["v_range"][0]) ) for operation in edited.get("edit_operations", []) ) ) removal_tolerance = max( 1e-6, expected_removed_volume * 1e-4, abs(teacher_facts["volume"]) * 1e-8, ) for strategy in strategies: trial = copy.deepcopy(edited) trial["reconstruction_strategy"] = { "boundary_strategy": strategy, "selection_status": "semantic_edit_candidate", } try: rebuilt = build_surfaceir(trial) export_step_shape(rebuilt, edited_step_path) trial_shape, trial_facts = _shape_facts(edited_step_path) volume_delta = ( trial_facts["volume"] - teacher_facts["volume"] ) if is_scale_transform: if is_uniform_scale: factor = float(scale_operation["factor"]) pivot = [ float(component) for component in scale_operation["pivot"] ] expected_volume = ( teacher_facts["volume"] * factor**3 ) expected_center = [ pivot[index] + factor * ( teacher_facts["center_of_mass"][index] - pivot[index] ) for index in range(3) ] expected_bounds_min = [ pivot[index] + factor * ( teacher_facts["bounds_min"][index] - pivot[index] ) for index in range(3) ] expected_bounds_max = [ pivot[index] + factor * ( teacher_facts["bounds_max"][index] - pivot[index] ) for index in range(3) ] target_observation = { "semantic_role": "overall_scale", "factor": factor, } else: baseline_topology = build_surfaceir( payload, strategy ) expected_topology = _apply_edit_operations( baseline_topology, {"edit_operations": [scale_operation]}, ) expected_step_path = ( output_dir / "expected-axis-affine-normalized.step" ) export_step_shape( expected_topology, expected_step_path ) _, expected_facts = _shape_facts( expected_step_path ) expected_volume = expected_facts["volume"] expected_center = expected_facts["center_of_mass"] expected_bounds_min = expected_facts["bounds_min"] expected_bounds_max = expected_facts["bounds_max"] factor = ( float(scale_operation["radial_factor"]) if scale_operation["semantic_role"] == "outer_diameter" else float(scale_operation["axial_factor"]) ) observed_ir = extract_surfaceir(edited_step_path) expected_parameter_value = float(value) target_observation = _observe_axis_affine_target( observed_ir, scale_operation, expected_parameter_value, ) observed_value = target_observation[ "observed_value" ] semantic_tolerance = max( 1e-5, abs(expected_parameter_value) * 1e-5 ) target_observation.update( { "expected_value": expected_parameter_value, "pass": observed_value is not None and abs( observed_value - expected_parameter_value ) <= semantic_tolerance, } ) volume_error = abs( trial_facts["volume"] - expected_volume ) / max(abs(expected_volume), 1e-12) center_error = math.dist( trial_facts["center_of_mass"], expected_center ) bounds_error = max( _max_error( trial_facts["bounds_min"], expected_bounds_min ), _max_error( trial_facts["bounds_max"], expected_bounds_max ), ) topology_match = all( trial_facts[name] == teacher_facts[name] for name in ( "solid_count", "face_count", "edge_count", ) ) vertex_count_delta = ( trial_facts["vertex_count"] - teacher_facts["vertex_count"] ) basic_pass = ( bool(trial_facts["valid"]) and topology_match and volume_error <= 1e-7 and center_error <= 1e-6 and bounds_error <= 1e-6 and target_observation.get("pass", True) ) trial_targets = [ { **target_observation, "expected_volume": expected_volume, "observed_volume": trial_facts["volume"], "relative_volume_error": volume_error, "center_error": center_error, "bounds_error": bounds_error, "topology_match": topology_match, "vertex_count_delta": vertex_count_delta, "pass": basic_pass, } ] strategy_attempts.append( { "boundary_strategy": strategy, "pass": basic_pass, "solid_count": trial_facts["solid_count"], "valid": trial_facts["valid"], "volume_delta": volume_delta, "scale_factor": factor, "operation": operation_names[0], "relative_volume_error": volume_error, "center_error": center_error, "bounds_error": bounds_error, "topology_match": topology_match, "vertex_count_delta": vertex_count_delta, "target_observation": target_observation, } ) else: trial_observed = _observe_internal_cylinders_from_shape( trial_shape ) trial_targets = _match_edit_cylinders( trial, trial_observed ) trial_removed_volume = -volume_delta basic_pass = ( bool(trial_facts["valid"]) and trial_facts["solid_count"] == teacher_facts["solid_count"] and bool(trial_targets) and all(row["pass"] for row in trial_targets) and volume_delta < -max( 1e-8, abs(teacher_facts["volume"]) * 1e-10 ) ) strategy_attempts.append( { "boundary_strategy": strategy, "pass": basic_pass, "solid_count": trial_facts["solid_count"], "valid": trial_facts["valid"], "target_axis_match_count": sum( row["pass"] for row in trial_targets ), "target_axis_count": len(trial_targets), "volume_delta": volume_delta, "removed_volume": trial_removed_volume, "expected_removed_volume": expected_removed_volume, "removal_tolerance": removal_tolerance, } ) if basic_pass: selected_result = ( trial, trial_shape, trial_facts, trial_targets, ) break except Exception as strategy_exc: strategy_attempts.append( { "boundary_strategy": strategy, "pass": False, "error": ( f"{type(strategy_exc).__name__}: {strategy_exc}" ), } ) edited["active_edit"]["strategy_attempts"] = strategy_attempts if selected_result is None: raise ValueError("No semantic edit reconstruction strategy passed") ( edited, edited_shape, edited_facts, target_rows, ) = selected_result edited["active_edit"]["selected_boundary_strategy"] = edited[ "reconstruction_strategy" ]["boundary_strategy"] write_designir(edited_ir_path, edited) expected_parameters = payload["semantic_layer"]["parameters"] expected_value = float(value) checks["source_independence"] = { "pass": not any( token in edited_ir_path.read_text(encoding="utf-8").lower() for token in ( str(teacher_step).lower(), teacher_step.name.lower(), "import_step", "stepcontrol_reader", "step_text", "\"brep\"", "\"mesh\"", ) ) } checks["target_parameter"] = { "pass": bool(target_rows) and all(row["pass"] for row in target_rows), "expected": expected_value, "observed_semantic_value": ( expected_value if all(row["pass"] for row in target_rows) else None ), "axis_radius_matches": target_rows, } checks["valid_solid"] = { "pass": bool(edited_facts["valid"]) and edited_facts["solid_count"] == teacher_facts["solid_count"], "teacher_solid_count": teacher_facts["solid_count"], "edited_solid_count": edited_facts["solid_count"], "edited_valid": edited_facts["valid"], } volume_delta = edited_facts["volume"] - teacher_facts["volume"] surface_area_delta = ( edited_facts["surface_area"] - teacher_facts["surface_area"] ) minimum_change = max( 1e-8, abs(teacher_facts["volume"]) * 1e-10 ) minimum_surface_change = max( 1e-8, abs(teacher_facts["surface_area"]) * 1e-10 ) checks["geometry_changed_as_expected"] = { "pass": ( ( abs(volume_delta) > minimum_change if teacher_facts["solid_count"] > 0 else abs(surface_area_delta) > minimum_surface_change ) if is_scale_transform else volume_delta < -minimum_change ), "teacher_volume": teacher_facts["volume"], "edited_volume": edited_facts["volume"], "volume_delta": volume_delta, "teacher_surface_area": teacher_facts["surface_area"], "edited_surface_area": edited_facts["surface_area"], "surface_area_delta": surface_area_delta, "expected_direction": ( "increase" if is_scale_transform and ( float(scale_operation.get("factor", 1.0)) > 1.0 or float(scale_operation.get("axial_factor", 1.0)) * float(scale_operation.get("radial_factor", 1.0)) ** 2 > 1.0 ) else "decrease" ), } invariant_rows = [] for name, expected in expected_parameters.items(): if name == parameter_name: continue expected_number = float(expected["value"]) edited_number = float( edited["semantic_layer"]["parameters"][name]["value"] ) invariant_rows.append( { "parameter": name, "expected": expected_number, "edited_designir": edited_number, "pass": edited_number == expected_number, } ) removed_volume = teacher_facts["volume"] - edited_facts["volume"] checks["non_target_parameters"] = { "pass": all(row["pass"] for row in invariant_rows) and ( operation_names in (["uniform_scale"], ["axis_affine_scale"]) if is_scale_transform else all( operation == "enlarge_cylindrical_cut" for operation in operation_names ) and removed_volume > minimum_change ), "rows": invariant_rows, "geometry_guard": { "removed_volume": removed_volume, "expected_removed_volume": expected_removed_volume, "tolerance": removal_tolerance, "operation_whitelist": operation_names, "expected": ( "only one declared scale transform may change geometry" if is_scale_transform else "only declared cylindrical Cut operations may remove material; no additive operation is allowed" ), }, } accepted = all(check["pass"] for check in checks.values()) state = "accepted" if accepted else "rejected" except Exception as exc: accepted = False state = "execution_failed" checks["execution"] = { "pass": False, "error": f"{type(exc).__name__}: {exc}", } edited["active_edit"]["acceptance_status"] = state write_designir(edited_ir_path, edited) report = { "schema_version": "1.0", "report_kind": "designir_3_semantic_edit_acceptance", "source_name": teacher_step.name, "model_id": payload["model_id"], "parameter": parameter_name, "requested_value": float(value), "state": state, "accepted": accepted, "checks": checks, "artifacts": { "edited_designir": edited_ir_path.name, "edited_step": ( edited_step_path.name if edited_step_path.is_file() else None ), }, } write_json(output_dir / "acceptance-report.json", report) return report def validate_edits_folder( input_dir: Path, designir_dir: Path, output_dir: Path, geometry_report_path: Path | None, limit: int | None, offset: int, workers: int, max_parameters_per_case: int, timeout_seconds: int, parameter_filter: str | None = None, ) -> dict[str, Any]: """Run isolated Acceptance-Agent workers over a STEP corpus.""" manifest = json.loads( (designir_dir / "manifest.json").read_text(encoding="utf-8") ) rows = [row for row in manifest["rows"] if row["state"] == "extracted"] rows = rows[offset:] if limit is not None: rows = rows[:limit] geometry_pass: dict[str, bool] = {} if geometry_report_path is not None: geometry_report = json.loads( geometry_report_path.read_text(encoding="utf-8") ) geometry_pass = { row["source_name"]: bool(row["geometry_pass"]) for row in geometry_report["rows"] } attempts = [] case_rows = [] for row in rows: source_name = row["source_name"] designir_path = designir_dir / row["output_name"] payload = read_designir(designir_path) if geometry_pass and not geometry_pass.get(source_name, False): case_rows.append( { "source_name": source_name, "state": "baseline_geometry_not_validated", "attempt_count": 0, } ) continue parameters = payload.get("semantic_layer", {}).get("parameters", {}) candidates = [ name for name in payload.get("edit_interface", {}).get( "semantic_parameters", [] ) if name in parameters and ( parameter_filter is None or name == parameter_filter ) and parameters[name].get("edit_state") in { "executable_enlarge_cut_binding", "executable_axis_affine_binding", "executable_uniform_scale_binding", "validated_executable_binding", } ][:max_parameters_per_case] if not candidates: case_rows.append( { "source_name": source_name, "state": "no_executable_semantic_parameter", "attempt_count": 0, } ) continue case_rows.append( { "source_name": source_name, "state": "scheduled", "attempt_count": len(candidates), } ) for parameter_name in candidates: old_value = float(parameters[parameter_name]["value"]) requested_value = old_value * 1.1 safe_parameter = re.sub(r"[^a-zA-Z0-9_.-]+", "_", parameter_name) attempt_dir = ( output_dir / Path(source_name).stem / safe_parameter ) attempts.append( { "source_name": source_name, "teacher": input_dir / source_name, "designir": designir_path, "parameter": parameter_name, "old_value": old_value, "requested_value": requested_value, "output_dir": attempt_dir, } ) def run_attempt(attempt: dict[str, Any]) -> dict[str, Any]: report_path = attempt["output_dir"] / "acceptance-report.json" if report_path.is_file(): report = json.loads(report_path.read_text(encoding="utf-8")) if ( report.get("parameter") == attempt["parameter"] and abs( float(report.get("requested_value", math.nan)) - attempt["requested_value"] ) <= 1e-12 and bool(report.get("accepted")) ): return { **attempt, "state": report["state"], "accepted": bool(report["accepted"]), "resumed": True, "report": str(report_path), } command = [ sys.executable, str(Path(__file__).resolve()), "validate-edit", str(attempt["teacher"]), str(attempt["designir"]), "--parameter", attempt["parameter"], "--value", str(attempt["requested_value"]), "--output-dir", str(attempt["output_dir"]), ] try: completed = subprocess.run( command, capture_output=True, text=True, timeout=timeout_seconds, check=False, ) if completed.returncode == 0 and report_path.is_file(): report = json.loads(report_path.read_text(encoding="utf-8")) return { **attempt, "state": report["state"], "accepted": bool(report["accepted"]), "resumed": False, "report": str(report_path), } return { **attempt, "state": "worker_failed", "accepted": False, "returncode": completed.returncode, "stderr_tail": completed.stderr[-1000:], } except subprocess.TimeoutExpired: return { **attempt, "state": "worker_timeout", "accepted": False, } attempt_rows = [] output_dir.mkdir(parents=True, exist_ok=True) with ThreadPoolExecutor(max_workers=max(1, workers)) as executor: futures = [executor.submit(run_attempt, attempt) for attempt in attempts] for future in as_completed(futures): attempt_rows.append(future.result()) attempt_rows.sort(key=lambda row: (row["source_name"], row["parameter"])) accepted_by_source = { row["source_name"] for row in attempt_rows if row["accepted"] } attempted_by_source = {row["source_name"] for row in attempt_rows} for row in case_rows: if row["source_name"] in accepted_by_source: row["state"] = "semantic_edit_accepted" elif row["source_name"] in attempted_by_source: row["state"] = "semantic_edit_rejected" state_counts = Counter(row["state"] for row in case_rows) attempt_state_counts = Counter(row["state"] for row in attempt_rows) report = { "schema_version": "1.0", "report_kind": "designir_3_semantic_edit_batch_acceptance", "case_count": len(case_rows), "applicable_case_count": len(attempted_by_source), "accepted_case_count": len(accepted_by_source), "attempt_count": len(attempt_rows), "accepted_attempt_count": sum(row["accepted"] for row in attempt_rows), "case_state_counts": dict(sorted(state_counts.items())), "attempt_state_counts": dict(sorted(attempt_state_counts.items())), "perturbation": { "kind": "multiply", "factor": 1.1, "maximum_parameters_per_case": max_parameters_per_case, }, "worker_policy": { "process_isolation": True, "workers": workers, "timeout_seconds": timeout_seconds, }, "cases": case_rows, "attempts": [ { key: ( str(value) if isinstance(value, Path) else value ) for key, value in row.items() } for row in attempt_rows ], } write_json(output_dir / "batch-acceptance-report.json", report) return report def merge_acceptance_reports( baseline_path: Path, retry_paths: list[Path], output: Path ) -> dict[str, Any]: baseline = json.loads(baseline_path.read_text(encoding="utf-8")) attempts = { (row["source_name"], row["parameter"]): row for row in baseline["attempts"] } baseline_factor = float( baseline.get("perturbation", {}).get("factor", 0.0) ) if baseline_factor > 0.0: for row in attempts.values(): if "old_value" not in row and "requested_value" in row: row["old_value"] = ( float(row["requested_value"]) / baseline_factor ) lineage = [str(baseline_path)] for retry_path in retry_paths: lineage.append(str(retry_path)) if retry_path.is_dir(): retry_rows = [] for report_path in sorted( retry_path.rglob("acceptance-report.json") ): retry_report = json.loads( report_path.read_text(encoding="utf-8") ) retry_rows.append( { "source_name": retry_report["source_name"], "parameter": retry_report["parameter"], "requested_value": retry_report["requested_value"], "state": retry_report["state"], "accepted": bool(retry_report["accepted"]), "resumed": False, "report": str(report_path), } ) else: retry = json.loads(retry_path.read_text(encoding="utf-8")) retry_rows = retry["attempts"] for row in retry_rows: key = (row["source_name"], row["parameter"]) previous = attempts.get(key, {}) merged_row = {**previous, **row} if "old_value" in previous and "old_value" not in row: merged_row["old_value"] = previous["old_value"] attempts[key] = merged_row attempt_rows = sorted( attempts.values(), key=lambda row: (row["source_name"], row["parameter"]) ) accepted_by_source = { row["source_name"] for row in attempt_rows if row["accepted"] } attempted_by_source = {row["source_name"] for row in attempt_rows} case_rows = copy.deepcopy(baseline["cases"]) for row in case_rows: if row["source_name"] in accepted_by_source: row["state"] = "semantic_edit_accepted" elif row["source_name"] in attempted_by_source: row["state"] = "semantic_edit_rejected" report = copy.deepcopy(baseline) report["report_kind"] = "designir_3_semantic_edit_merged_acceptance" report["lineage"] = lineage report["attempts"] = attempt_rows report["cases"] = case_rows report["applicable_case_count"] = len(attempted_by_source) report["accepted_case_count"] = len(accepted_by_source) report["attempt_count"] = len(attempt_rows) report["accepted_attempt_count"] = sum( row["accepted"] for row in attempt_rows ) report["case_state_counts"] = dict( sorted(Counter(row["state"] for row in case_rows).items()) ) report["attempt_state_counts"] = dict( sorted(Counter(row["state"] for row in attempt_rows).items()) ) factors = sorted( { round( float(row["requested_value"]) / float(row["old_value"]), 8, ) for row in attempt_rows if float(row.get("old_value", 0.0)) != 0.0 } ) if len(factors) > 1: report["perturbation"] = { "kind": "adaptive_multiply", "validated_factors": factors, "selection_policy": "largest attempted factor with a passing edit contract", } write_json(output, report) return report def apply_edit_validation_to_designir( designir_dir: Path, acceptance_report_path: Path, output: Path ) -> dict[str, Any]: """Persist measured edit bounds and rejections in private DesignIR files.""" report = json.loads( acceptance_report_path.read_text(encoding="utf-8") ) rows = [] for attempt in report["attempts"]: source_name = attempt["source_name"] parameter_name = attempt["parameter"] designir_path = ( designir_dir / f"{Path(source_name).stem}.designir.json" ) payload = read_designir(designir_path) parameters = payload.get("semantic_layer", {}).get("parameters", {}) if parameter_name not in parameters: raise ValueError( f"{source_name} is missing semantic parameter {parameter_name}" ) parameter = parameters[parameter_name] old_value = float( attempt.get("old_value", parameter["value"]) ) requested_value = float(attempt["requested_value"]) if attempt["accepted"]: parameter["editable"] = True parameter["edit_state"] = "validated_executable_binding" parameter["validated_range"] = { "direction": "increase", "minimum_tested_inclusive": _number(old_value), "maximum_tested_inclusive": _number(requested_value), "acceptance_contract": "designir_3_semantic_edit_acceptance", } state = "validated_edit_range" else: parameter["editable"] = False parameter["edit_state"] = "rejected_by_edit_contract" parameter["validation_failure"] = { "requested_value": _number(requested_value), "attempt_state": attempt["state"], } parameter.pop("validated_range", None) state = "rejected_edit_binding" write_json(designir_path, payload) rows.append( { "source_name": source_name, "parameter": parameter_name, "state": state, "old_value": _number(old_value), "requested_value": _number(requested_value), } ) summary = { "schema_version": "1.0", "report_kind": "designir_3_persisted_edit_validation", "case_count": len({row["source_name"] for row in rows}), "parameter_count": len(rows), "validated_parameter_count": sum( row["state"] == "validated_edit_range" for row in rows ), "rejected_parameter_count": sum( row["state"] == "rejected_edit_binding" for row in rows ), "acceptance_report": str(acceptance_report_path), "rows": rows, } write_json(output, summary) return summary def main() -> None: parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) extract = commands.add_parser("extract") extract.add_argument("step", type=Path) extract.add_argument("--output", type=Path, required=True) rebuild = commands.add_parser("rebuild") rebuild.add_argument("designir", type=Path) rebuild.add_argument("--output", type=Path, required=True) rebuild.add_argument( "--boundary-strategy", choices=[ "exact_3d", "exact_3d_pcurve", "shared_pcurve", "first_pcurve", "analytic_uv_rect", "uv_rect_all", ], ) edit = commands.add_parser("edit") edit.add_argument("designir", type=Path) edit.add_argument("--parameter", required=True) edit.add_argument("--value", type=float, required=True) edit.add_argument("--output-designir", type=Path, required=True) edit.add_argument("--output-step", type=Path) validate_edit = commands.add_parser("validate-edit") validate_edit.add_argument("teacher", type=Path) validate_edit.add_argument("designir", type=Path) validate_edit.add_argument("--parameter", required=True) validate_edit.add_argument("--value", type=float, required=True) validate_edit.add_argument("--output-dir", type=Path, required=True) validate_edits = commands.add_parser("validate-edits-folder") validate_edits.add_argument("input", type=Path) validate_edits.add_argument("--designir-dir", type=Path, required=True) validate_edits.add_argument("--output-dir", type=Path, required=True) validate_edits.add_argument("--geometry-report", type=Path) validate_edits.add_argument("--limit", type=int) validate_edits.add_argument("--offset", type=int, default=0) validate_edits.add_argument("--workers", type=int, default=4) validate_edits.add_argument("--max-parameters-per-case", type=int, default=1) validate_edits.add_argument("--timeout-seconds", type=int, default=60) validate_edits.add_argument("--parameter") merge_acceptance = commands.add_parser("merge-acceptance-reports") merge_acceptance.add_argument("--baseline", type=Path, required=True) merge_acceptance.add_argument( "--retry", type=Path, action="append", required=True ) merge_acceptance.add_argument("--output", type=Path, required=True) persist_edit_validation = commands.add_parser("persist-edit-validation") persist_edit_validation.add_argument( "--designir-dir", type=Path, required=True ) persist_edit_validation.add_argument( "--acceptance-report", type=Path, required=True ) persist_edit_validation.add_argument("--output", type=Path, required=True) augment_scale = commands.add_parser("augment-overall-scale") augment_scale.add_argument("designir_dir", type=Path) augment_scale.add_argument("--output", type=Path, required=True) augment_axis = commands.add_parser("augment-axis-affine-bindings") augment_axis.add_argument("designir_dir", type=Path) augment_axis.add_argument("--output", type=Path, required=True) augment_directional = commands.add_parser( "augment-directional-size-bindings" ) augment_directional.add_argument("designir_dir", type=Path) augment_directional.add_argument("--output", type=Path, required=True) batch = commands.add_parser("extract-folder") batch.add_argument("input", type=Path) batch.add_argument("--output-dir", type=Path, required=True) validate = commands.add_parser("validate-folder") validate.add_argument("input", type=Path) validate.add_argument("--designir-dir", type=Path, required=True) validate.add_argument("--rebuilt-dir", type=Path, required=True) validate.add_argument("--limit", type=int) validate.add_argument("--offset", type=int, default=0) validate.add_argument("--boolean", action="store_true") validate.add_argument( "--boundary-strategy", choices=[ "exact_3d", "exact_3d_pcurve", "shared_pcurve", "first_pcurve", "analytic_uv_rect", "uv_rect_all", ], default="exact_3d", ) reclassify = commands.add_parser("reclassify-validation-report") reclassify.add_argument("report", type=Path) reclassify.add_argument("--output", type=Path, required=True) select = commands.add_parser("select-strategies") select.add_argument("--designir-dir", type=Path, required=True) select.add_argument("--primary-report", type=Path, required=True) select.add_argument( "--fallback-report", type=Path, action="append", required=True ) select.add_argument("--output", type=Path, required=True) args = parser.parse_args() if args.command == "extract": payload = extract_surfaceir(args.step.resolve()) write_json(args.output.resolve(), payload) print( json.dumps( { "output": str(args.output.resolve()), "solid_count": len(payload["surface_layer"]["solids"]), "surface_vocabulary": payload["surface_layer"][ "surface_vocabulary" ], "curve_vocabulary": payload["surface_layer"][ "curve_vocabulary" ], }, ensure_ascii=False, ) ) elif args.command == "rebuild": payload = read_designir(args.designir) shape = build_surfaceir(payload, args.boundary_strategy) export_step_shape(shape, args.output) print(json.dumps({"output": str(args.output.resolve())})) elif args.command == "edit": payload = read_designir(args.designir) edited = apply_semantic_parameter( payload, args.parameter, args.value ) write_designir(args.output_designir.resolve(), edited) output = { "output_designir": str(args.output_designir.resolve()), "parameter": args.parameter, "value": args.value, } if args.output_step is not None: shape = build_surfaceir(edited) export_step_shape(shape, args.output_step.resolve()) output["output_step"] = str(args.output_step.resolve()) print(json.dumps(output, ensure_ascii=False)) elif args.command == "validate-edit": report = validate_semantic_edit( args.teacher.resolve(), args.designir.resolve(), args.parameter, args.value, args.output_dir.resolve(), ) print( json.dumps( { "report": str( (args.output_dir.resolve() / "acceptance-report.json") ), "state": report["state"], "accepted": report["accepted"], }, ensure_ascii=False, ) ) elif args.command == "validate-edits-folder": report = validate_edits_folder( args.input.resolve(), args.designir_dir.resolve(), args.output_dir.resolve(), args.geometry_report.resolve() if args.geometry_report is not None else None, args.limit, args.offset, args.workers, args.max_parameters_per_case, args.timeout_seconds, args.parameter, ) print( json.dumps( { "report": str( ( args.output_dir.resolve() / "batch-acceptance-report.json" ) ), "case_count": report["case_count"], "applicable_case_count": report["applicable_case_count"], "accepted_case_count": report["accepted_case_count"], "attempt_count": report["attempt_count"], "accepted_attempt_count": report[ "accepted_attempt_count" ], "case_state_counts": report["case_state_counts"], "attempt_state_counts": report["attempt_state_counts"], }, ensure_ascii=False, ) ) elif args.command == "merge-acceptance-reports": report = merge_acceptance_reports( args.baseline.resolve(), [path.resolve() for path in args.retry], args.output.resolve(), ) print( json.dumps( { "output": str(args.output.resolve()), "applicable_case_count": report["applicable_case_count"], "accepted_case_count": report["accepted_case_count"], "attempt_state_counts": report["attempt_state_counts"], }, ensure_ascii=False, ) ) elif args.command == "persist-edit-validation": report = apply_edit_validation_to_designir( args.designir_dir.resolve(), args.acceptance_report.resolve(), args.output.resolve(), ) print( json.dumps( { "output": str(args.output.resolve()), "parameter_count": report["parameter_count"], "validated_parameter_count": report[ "validated_parameter_count" ], "rejected_parameter_count": report[ "rejected_parameter_count" ], }, ensure_ascii=False, ) ) elif args.command == "augment-overall-scale": report = augment_overall_scale( args.designir_dir.resolve(), args.output.resolve(), ) print( json.dumps( { "output": str(args.output.resolve()), "case_count": report["case_count"], "augmented_count": report["augmented_count"], "skipped_count": report["skipped_count"], }, ensure_ascii=False, ) ) elif args.command == "augment-axis-affine-bindings": report = augment_axis_affine_bindings( args.designir_dir.resolve(), args.output.resolve(), ) print( json.dumps( { "output": str(args.output.resolve()), "case_count": report["case_count"], "applicable_case_count": report[ "applicable_case_count" ], "parameter_count": report["parameter_count"], }, ensure_ascii=False, ) ) elif args.command == "augment-directional-size-bindings": report = augment_directional_size_bindings( args.designir_dir.resolve(), args.output.resolve(), ) print( json.dumps( { "output": str(args.output.resolve()), "case_count": report["case_count"], "applicable_case_count": report[ "applicable_case_count" ], "parameter_count": report["parameter_count"], }, ensure_ascii=False, ) ) elif args.command == "extract-folder": report = extract_folder(args.input.resolve(), args.output_dir.resolve()) print( json.dumps( { "manifest": str( (args.output_dir.resolve() / "manifest.json").resolve() ), "input_count": report["input_count"], "extracted_count": report["extracted_count"], "failed_count": report["failed_count"], "total_json_bytes": report["total_json_bytes"], }, ensure_ascii=False, ) ) elif args.command == "validate-folder": report = validate_folder( args.input.resolve(), args.designir_dir.resolve(), args.rebuilt_dir.resolve(), args.limit, args.offset, args.boolean, args.boundary_strategy, ) print( json.dumps( { "report": str( (args.rebuilt_dir.resolve() / "validation-report.json") ), "case_count": report["case_count"], "geometry_pass_count": report["geometry_pass_count"], "geometry_fail_count": report["geometry_fail_count"], "usable_reconstruction_count": report[ "usable_reconstruction_count" ], "state_counts": report["state_counts"], }, ensure_ascii=False, ) ) elif args.command == "reclassify-validation-report": report = reclassify_validation_report( args.report.resolve(), args.output.resolve(), ) print( json.dumps( { "output": str(args.output.resolve()), "case_count": report["case_count"], "geometry_pass_count": report[ "geometry_pass_count" ], "geometry_fail_count": report[ "geometry_fail_count" ], "state_counts": report["state_counts"], }, ensure_ascii=False, ) ) elif args.command == "select-strategies": report = select_validated_strategies( args.designir_dir.resolve(), args.primary_report.resolve(), [path.resolve() for path in args.fallback_report], args.output.resolve(), ) print( json.dumps( { "output": str(args.output.resolve()), "case_count": report["case_count"], "geometry_validated_count": report["geometry_validated_count"], "unresolved_count": report["unresolved_count"], "selected_strategy_counts": report["selected_strategy_counts"], }, ensure_ascii=False, ) ) if __name__ == "__main__": main()