"""Session-based CDSL execution with atomic executor registry.""" from __future__ import annotations from copy import deepcopy from dataclasses import dataclass, field import math from pathlib import Path from typing import Any, Callable, Protocol from .build123d_adapter import Build123dGeometryAdapter from .capabilities import CapabilityAnalyzer, pattern_transform_blocker, sketch_ids_required_by_contract from .runtime_types import ( AxisSpec, CapabilityResult, FeaturePlanNode, FeatureResult, HoleSpec, PlaneSpec, ThreadSpec, Vector3, RuntimeDiagnostic, SelectorResolution, TopologyRecord, TopologyRegistry, vector_add, vector_cross, vector_dot, vector_scale, vector_subtract, vector_unit, ) from .sketch_solver import CORE_SHAPE_GENERATORS, resolve_required_sketches ALL_ATOMIC_IDS = frozenset({ "extrude_add_blind", "extrude_add_blind_with_hole", "extrude_add_two_sided", "extrude_cut_blind", "extrude_cut_two_sided", "extrude_surface", "extrude_cut_through", "loft_add", "loft_add_with_cap_face", "sweep_add", "revolve_add", "revolve_cut", "revolve_surface", "hole_blind", "hole_countersink", "hole_counterbore", "sphere_add", "box_add", "cylinder_add", "reference_plane", "reference_axis", "hole_wizard", "fillet", "chamfer", "shell", "pattern_linear", "pattern_mirror", "pattern_circular", "boolean_bodies", "thread_add", "thread_cut", }) class RuntimeExecutionError(RuntimeError): """A feature execution failure with serializable runtime evidence.""" def __init__(self, diagnostic: RuntimeDiagnostic, selector_resolutions: list[dict[str, Any]]) -> None: super().__init__(diagnostic.message) self.diagnostic = diagnostic self.selector_resolutions = selector_resolutions class FeatureExecutionError(RuntimeError): """An expected feature-level execution rejection with a stable code.""" def __init__(self, code: str, message: str, **detail: Any) -> None: super().__init__(message) self.code = code self.detail = detail @dataclass(frozen=True) class ExtentVector: """Single-directional extrusion displacement for one profile face. ``trim_to`` stays ``None`` for an exact vector extrusion. When set, the extent means "extrude until the target face, trimming any profile region that does not reach it" (up_to_surface trim semantics, issue #5). The piercing distance is computed inside the adapter, so ``vector`` only supplies the direction. """ vector: Vector3 trim_to: Any | None = None class AtomicExecutor(Protocol): atomic_id: str def preflight(self, node: FeaturePlanNode, session: "ExecutionSession") -> CapabilityResult: ... def execute(self, node: FeaturePlanNode, session: "ExecutionSession") -> FeatureResult: ... class GeometryAdapter(Protocol): """Kernel boundary consumed by the session runtime. Geometry values remain opaque here. A future adapter may use a different B-rep kernel as long as it preserves these construction/query contracts. """ def topology_records(self, body: Any, feature_id: str, body_id: str) -> list[TopologyRecord]: ... def body_solids(self, body: Any) -> list[Any]: ... def body_geometry(self, body: Any) -> dict[str, Any]: ... def surface_geometry(self, surface: Any) -> dict[str, Any]: ... def faces_for_sketch(self, sketch: dict[str, Any]) -> list[Any]: ... def face_with_holes(self, outer: Any, holes: list[Any]) -> Any: ... def loft(self, sketches: list[dict[str, Any]]) -> Any: ... def loft_with_cap_face(self, cap_face: Any, sketches: list[dict[str, Any]]) -> Any: ... def sweep(self, section: Any, spine: Any, *, inner_wires: list[Any] | None = None, make_solid: bool = True, is_frenet: bool = False, transition: Any = None) -> Any: ... def sweep_path(self, points: list[Vector3], *, start_tangent: Vector3 | None = None, end_tangent: Vector3 | None = None, parameters: list[float] | None = None) -> Any: ... def extrude(self, face: Any, direction: Vector3) -> Any: ... def extrude_taper(self, face: Any, direction: Vector3, taper_deg: float) -> Any: ... def extrude_trimmed(self, face: Any, target: Any, direction: Vector3) -> Any: ... def surface_wires_for_sketch(self, sketch: dict[str, Any]) -> list[Any]: ... def extrude_surface(self, wires: list[Any], direction: Vector3) -> Any: ... def combine_surfaces(self, *surfaces: Any) -> Any: ... def revolve(self, face: Any, angle_deg: float, axis: AxisSpec) -> Any: ... def revolve_surface(self, wire: Any, angle_deg: float, axis: AxisSpec) -> Any: ... def intersect(self, left: Any, right: Any) -> Any: ... def fuse(self, body: Any | None, solid: Any) -> Any: ... def combine(self, body: Any | None, solid: Any) -> Any: ... def cut(self, body: Any, tool: Any) -> Any: ... def sphere(self, radius_mm: float, center_mm: Vector3) -> Any: ... def thread_solid(self, spec: ThreadSpec) -> Any: ... def hole_tool(self, spec: HoleSpec, starts: list[Vector3], inward: Vector3, through_depth_mm: float) -> Any: ... def body_center(self, body: Any) -> Vector3: ... def body_span(self, body: Any, direction: Vector3) -> float: ... def vertex_coordinates(self, vertex: Any) -> Vector3: ... def intersection_vertex(self, body: Any, face_sets: list[list[Any]]) -> Any: ... def profile_sample_points(self, face: Any) -> list[Any]: ... def profile_touches_target(self, target: Any, faces: list[Any]) -> bool: ... def next_body_face_after(self, body: Any, faces: list[Any], direction: Vector3, *, excluded_face: Any) -> Any: ... def uniform_intersection_distance(self, target: Any, faces: list[Any], direction: Vector3) -> float: ... def fillet(self, body: Any, radius_mm: float, edges: list[Any]) -> Any: ... def tangent_edges(self, body: Any, seeds: list[Any]) -> list[Any]: ... def chamfer(self, body: Any, distance_mm: float, distance_2_mm: float | None, edges: list[Any], face: Any | None = None) -> Any: ... def surface_limited_chamfer(self, body: Any, distance_mm: float, edges: list[Any], surfaces: list[Any]) -> Any: ... def shell(self, body: Any, faces: list[Any], thickness_mm: float, *, inward: bool = True) -> Any: ... def export(self, body: Any, path: str) -> None: ... @dataclass class ExecutionSession: sketches: dict[str, dict[str, Any]] nodes: dict[str, FeaturePlanNode] adapter: GeometryAdapter = field(default_factory=Build123dGeometryAdapter) topology: TopologyRegistry = field(default_factory=TopologyRegistry) body: Any | None = None body_id: str | None = None results: dict[str, FeatureResult] = field(default_factory=dict) replay_definitions: dict[str, FeaturePlanNode] = field(default_factory=dict) body_members: dict[str, Any] = field(default_factory=dict) surface_members: dict[str, Any] = field(default_factory=dict) selector_resolutions: list[dict[str, Any]] = field(default_factory=list) active_feature_id: str = "" def register_body( self, feature_id: str, body: Any, *, replay_node: FeaturePlanNode | None = None, body_members: dict[str, Any] | None = None, ) -> None: # #7 multi-body：主体可能是 Compound（多个独立实体，例如两个不相交的 # 拉伸）。body_id 现在反映真实实体结构而不是"最后一个特征的 id"： # 每个独立 Solid 一个 body:{feature}:{index}，供 selector 精确匹配目标 # 实体；单体保持 body:{feature}（与历史行为完全一致）。 self.body = body self.body_id = f"body:{feature_id}" self.body_members = dict(body_members) if body_members is not None else {feature_id: body} solids = self.adapter.body_solids(body) if len(solids) <= 1: self.topology.replace_body_topology(feature_id, self.body_id, self.adapter.topology_records(body, feature_id, self.body_id)) else: # 一个 Compound 的全部成员共享同一个前置 body snapshot。逐个登记会让 # 已登记的本轮成员成为下一个成员的 predecessor，进而把 pattern copy # 的 owner 错误转移到相邻实例。必须原子替换整个多 body 拓扑快照。 members = [ (member_id, self.adapter.topology_records(solid, feature_id, member_id)) for index, solid in enumerate(solids) for member_id in [f"{self.body_id}:{index}"] ] self.topology.replace_body_topologies(feature_id, members, active_body_id=self.body_id) self.topology.register(TopologyRecord( record_id=self.body_id, kind="body", feature_id=feature_id, body_id=self.body_id, geometry=self.adapter.body_geometry(body), value=body, owner_feature_ids=(feature_id,), )) if replay_node is not None: self.replay_definitions[feature_id] = replay_node def register_surface(self, feature_id: str, surface: Any) -> str: # 曲面 feature 与实体 body 生命周期相互独立：不能调用 register_body， # 否则 surface 会覆盖 active solid 并改变最终 STEP 的实体结果。 surface_id = f"surface:{feature_id}" self.surface_members[feature_id] = surface for record in self.adapter.topology_records(surface, feature_id, surface_id): self.topology.register(record) self.topology.register(TopologyRecord( record_id=surface_id, kind="surface", feature_id=feature_id, body_id=surface_id, geometry=self.adapter.surface_geometry(surface), value=surface, owner_feature_ids=(feature_id,), )) return surface_id def _record_selector_resolution(self, resolution: SelectorResolution) -> SelectorResolution: evidence = resolution.as_dict() evidence["feature_id"] = self.active_feature_id self.selector_resolutions.append(evidence) return resolution def _intersection_component_records(self, selector: dict[str, Any]) -> list[TopologyRecord]: matched = selector.get("matched_selectors") if selector.get("match_mode") == "all" else None if matched is not None: if not isinstance(matched, list) or not matched: raise FeatureExecutionError("intersection_selector_unbound", "Intersection selector has no bound face matches") resolved = [self._record_selector_resolution(self.topology.resolve(item, active_body_id=self.body_id)) for item in matched] else: binding_feature_id = selector.get("binding_feature_id") active_body_id = None if binding_feature_id and self.body_id != f"body:{binding_feature_id}" else self.body_id resolved = [self._record_selector_resolution(self.topology.resolve(selector, active_body_id=active_body_id))] failures = [item for item in resolved if item.status != "resolved" or item.record is None] if failures: detail = failures[0].diagnostic.message if failures[0].diagnostic else "intersection selector component was not resolved" raise FeatureExecutionError("intersection_selector_component_unresolved", detail) return [item.record for item in resolved if item.record is not None] def _resolve_intersection_vertex(self, selector: dict[str, Any]) -> SelectorResolution: components = selector.get("intersection_of") if self.body is None: return SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=RuntimeDiagnostic("missing_extent_body", "Intersection selector requires an existing body"), ) if not isinstance(components, list) or len(components) < 2: return SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=RuntimeDiagnostic("intersection_selector_incomplete", "Intersection selector requires at least two face components"), ) try: face_sets = [self._intersection_component_records(component) for component in components] if any(record.kind != "face" for records in face_sets for record in records): raise FeatureExecutionError("intersection_selector_kind", "Intersection selector components must resolve to faces") vertex = self.adapter.intersection_vertex(self.body, [[record.value for record in records] for records in face_sets]) except FeatureExecutionError as error: return SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=RuntimeDiagnostic(error.code, str(error), detail=error.detail), ) except ValueError as error: return SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=RuntimeDiagnostic("intersection_vertex_unresolved", str(error)), ) point = self.adapter.vertex_coordinates(vertex) record = TopologyRecord( record_id=str(selector.get("stable_id") or f"intersection:{id(vertex)}"), kind="vertex", feature_id=self.active_feature_id, body_id=self.body_id, geometry={"center_mm": list(point)}, value=vertex, owner_feature_ids=tuple(filter(None, [str(selector.get("owner_feature_id") or "")])), ) return SelectorResolution( selector=selector, status="resolved", record=record, candidates=({"score": 1.0, **record.public_dict()},), ) def resolve(self, selector: dict[str, Any]) -> SelectorResolution: if selector.get("intersection_of") is not None: return self._record_selector_resolution(self._resolve_intersection_vertex(selector)) owner = str(selector.get("owner_feature_id") or "") active_body_id = f"surface:{owner}" if owner in self.surface_members else self.body_id return self._record_selector_resolution(self.topology.resolve(selector, active_body_id=active_body_id)) def result( self, node: FeaturePlanNode, *, context: PlaneSpec | AxisSpec | None = None, diagnostics: list[RuntimeDiagnostic] | None = None, include_body: bool = True, surface_id: str | None = None, ) -> FeatureResult: result = FeatureResult( feature_id=node.feature_id, atomic_id=node.atomic_id, status="executed", body_id=self.body_id if include_body else None, surface_id=surface_id, context=context, replay_definition={"atomic_id": node.atomic_id, "params": deepcopy(node.params), "sketch_id": node.sketch_id}, diagnostics=diagnostics or [], ) self.results[node.feature_id] = result return result def replay_sources(self, source_feature_ids: list[Any]) -> list[FeaturePlanNode]: """Return selected source features in their original history order. A pattern's exported selection order is not an execution order. In particular, a boolean cut may appear before its parent boss in the raw selection array. The CDSL feature list is dependency-ordered by semantic validation, so it is the stable order for replay. """ requested = {str(feature_id) for feature_id in source_feature_ids} sources = [ feature for feature_id, feature in self.nodes.items() if feature_id in requested and feature_id in self.replay_definitions ] if len(sources) != len(requested): missing = sorted(requested - {source.feature_id for source in sources}) raise ValueError(f"pattern source features have no replay definitions: {', '.join(missing)}") return sources def _normal_from_sketch(sketch: dict[str, Any]) -> Vector3: return PlaneSpec.from_mapping(sketch.get("workplane") or {}).normal def _extent_reference(node: FeaturePlanNode, condition: dict[str, Any] | None = None) -> dict[str, Any]: condition = condition or node.params.get("end_condition") or {} reference = condition.get("reference") if not isinstance(reference, dict): raise FeatureExecutionError( "missing_extent_reference", "This end condition requires a captured target selector", extent=condition.get("type"), ) return reference def _targeted_extent_vector( node: FeaturePlanNode, faces: list[Any], direction: Vector3, session: ExecutionSession, condition: str, *, end_condition: dict[str, Any] | None = None, offset_mm: float | None = None, ) -> ExtentVector: if session.body is None: raise FeatureExecutionError("missing_extent_body", "Selector-dependent extent requires an existing body", extent=condition) if condition == "through_next": target = session.body else: reference = _extent_reference(node, end_condition) resolution = session.resolve(reference) if resolution.status != "resolved" or resolution.record is None: raise ValueError(resolution.diagnostic.message if resolution.diagnostic else "extent target was not resolved") expected_kind = {"up_to_vertex": "vertex", "up_to_body": "body"}.get(condition, "face") if resolution.record.kind != expected_kind: raise FeatureExecutionError( "unsupported_extent_target", "The resolved target kind is incompatible with this end condition", extent=condition, expected_kind=expected_kind, actual_kind=resolution.record.kind, ) target = resolution.record.value if condition == "up_to_vertex": target_point = session.adapter.vertex_coordinates(target) projections = [ vector_dot(vector_subtract(target_point, point), direction) for face in faces for point in session.adapter.profile_sample_points(face) ] if not projections or min(projections) <= 1e-6: raise FeatureExecutionError("extent_target_not_in_direction", "The target vertex is not ahead of the profile", extent=condition) if max(projections) - min(projections) > 1e-5: raise FeatureExecutionError("non_uniform_extent_target", "The target vertex does not define one extrusion distance", extent=condition) distance = sum(projections) / len(projections) else: if condition == "up_to_surface" and session.adapter.profile_touches_target(target, faces): # 草图轮廓本身就在所选终止面上时，selected face 只是拉伸的起始 # 边界。应沿实际拉伸方向穿过当前 body，取下一张完整截获 profile # 的边界面作为终止面；直接裁剪到 selected face 会生成零厚度工具体。 try: next_face = session.adapter.next_body_face_after( session.body, faces, direction, excluded_face=target, ) except ValueError as error: raise FeatureExecutionError("extent_target_not_reached", str(error), extent=condition) from error return ExtentVector(vector_scale(direction, 1.0), trim_to=next_face) try: distance = session.adapter.uniform_intersection_distance(target, faces, direction) except ValueError as error: message = str(error) code = "non_uniform_extent_target" if "non-uniform" in message else "extent_target_not_reached" if condition in {"up_to_surface", "through_next"}: # #5 高级终止条件：profile 与目标面非均匀相交（部分采样点未 # 命中目标 → 悬空；或各点命中距离不一 → 斜目标面）时不再整体 # 拒绝，而是"裁剪"——只保留从 profile 到目标面之间的材料。 # extrude_trimmed 内部做穿透拉伸 + 与目标面体层布尔求交，未达 # 目标的部分被切掉（CAD "拉伸到面"标准语义）。若全部采样点都 # 未命中（profile 与目标面无交叠），extrude_trimmed 内部仍抛 # "not reached"，保持显式拒绝。 # through_next 从当前主体中选取实际命中的下一张面； # up_to_vertex/up_to_body/offset_from_surface 无 face 可构造 # 裁剪体层，仍保持显式拒绝。 return ExtentVector(vector_scale(direction, 1.0), trim_to=target) raise FeatureExecutionError(code, message, extent=condition) from error offset = abs(float((end_condition or {}).get("offset_mm") or 0.0)) if condition == "offset_from_surface": offset = abs(float(offset_mm if offset_mm is not None else offset or node.params.get("distance_mm") or 0.0)) if offset: distance -= offset if distance <= 1e-6: raise FeatureExecutionError( "invalid_extent_offset", "Offset distance reaches or passes the target extent", extent=condition, offset_mm=offset, ) return ExtentVector(vector_scale(direction, distance)) def _side_extent_vectors( node: FeaturePlanNode, faces: list[Any], direction: Vector3, session: ExecutionSession, *, end_condition: dict[str, Any], distance_mm: float, ) -> list[ExtentVector]: """Resolve one directional extent without borrowing the opposite side. ``extrude_add_two_sided`` and ``extrude_cut_two_sided`` call this once for each independently captured termination. The regular one-sided executor also uses it for all simple termination modes, keeping the geometry adapter interface uniform. """ condition = str(end_condition.get("type") or "blind") distance = abs(float(distance_mm or 0.0)) if condition == "blind": if distance <= 0: raise ValueError("blind extent requires distance_mm > 0") return [ExtentVector(vector_scale(direction, distance))] if condition == "mid_plane": if distance <= 0: raise ValueError("mid_plane extent requires distance_mm > 0") return [ ExtentVector(vector_scale(direction, distance / 2)), ExtentVector(vector_scale(direction, -distance / 2)), ] if condition == "through_all": if session.body is None: if distance <= 0: raise ValueError("through_all on an initial feature has no body and no fallback distance") # 注意：Vector3 是 tuple，不能直接做 direction * distance（那是元组 # 重复），这里必须用 vector_scale 做数乘（顺带修复的隐藏 bug）。 return [ExtentVector(vector_scale(direction, distance))] return [ExtentVector(vector_scale(direction, max(session.adapter.body_span(session.body, direction), 1.0) + 2.0))] if condition in {"up_to_surface", "up_to_vertex", "offset_from_surface", "through_next", "up_to_body"}: return [ _targeted_extent_vector( node, faces, direction, session, condition, end_condition=end_condition, offset_mm=distance, ) ] raise ValueError(f"unsupported directional extent {condition!r}") def _extent_vectors( node: FeaturePlanNode, faces: list[Any], sketch: dict[str, Any], session: ExecutionSession, ) -> list[ExtentVector]: params = node.params normal = vector_unit(_normal_from_sketch(sketch), field_name="sketch normal") if bool(params.get("reverse")): normal = vector_scale(normal, -1) end_condition = params.get("end_condition") or {"type": "blind"} condition = end_condition.get("type", "blind") distance = abs(float(params.get("distance_mm") or 0.0)) if node.atomic_id in {"extrude_add_two_sided", "extrude_cut_two_sided"}: reverse_condition = params.get("reverse_end_condition") or {"type": "blind"} reverse_distance = abs(float(params.get("reverse_distance_mm") or 0.0)) if reverse_distance <= 0: raise ValueError("two-sided extrusion requires reverse_distance_mm > 0") return [ *_side_extent_vectors( node, faces, normal, session, end_condition=end_condition, distance_mm=distance, ), *_side_extent_vectors( node, faces, vector_scale(normal, -1), session, end_condition=reverse_condition, distance_mm=reverse_distance, ), ] if condition in {"through_all", "through_all_both", "through_all_and_blind"}: if session.body is None: # A first feature with through-all has no body to terminate # against. The source must provide a usable blind component. if distance <= 0: raise ValueError("through_all on an initial feature has no body and no fallback distance") return [ExtentVector(vector_scale(normal, distance))] span = max(session.adapter.body_span(session.body, normal), 1.0) + 2.0 if condition == "through_all": return [ExtentVector(vector_scale(normal, span))] if condition == "through_all_both": return [ExtentVector(vector_scale(normal, span)), ExtentVector(vector_scale(normal, -span))] # Through-all-and-blind is represented by a through direction plus # its captured opposite blind direction when available. reverse_distance = abs(float(params.get("reverse_distance_mm") or 0.0)) return [ ExtentVector(vector_scale(normal, span)), ExtentVector(vector_scale(normal, -(reverse_distance or span))), ] return _side_extent_vectors( node, faces, normal, session, end_condition=end_condition, distance_mm=distance, ) def _revolve_axis(node: FeaturePlanNode, session: ExecutionSession) -> AxisSpec: raw_axis = node.params.get("axis") or {} if raw_axis.get("origin_mm") is not None and raw_axis.get("direction") is not None: return AxisSpec.from_mapping(raw_axis) selector = raw_axis.get("selector") if isinstance(raw_axis, dict) else None if not isinstance(selector, dict): selector = next((item for item in node.selectors if item.get("kind") == "axis"), None) if not isinstance(selector, dict): raise FeatureExecutionError( "missing_revolve_axis", "Revolve requires an explicit axis or an owner-qualified reference-axis selector", ) resolution = session.resolve(selector) if resolution.status != "resolved" or resolution.record is None: raise ValueError(resolution.diagnostic.message if resolution.diagnostic else "revolve axis was not resolved") if not isinstance(resolution.record.value, AxisSpec): raise FeatureExecutionError( "unsupported_revolve_axis", "The resolved context is not an axis", actual_kind=resolution.record.kind, ) return resolution.record.value def _validate_revolve_axis_in_sketch_plane(axis: AxisSpec, sketch: dict[str, Any]) -> None: """Defend direct CDSL execution from an out-of-plane revolve axis.""" plane = PlaneSpec.from_mapping(sketch.get("workplane") or {}) direction_normal_dot = abs(vector_dot(axis.direction, plane.normal)) if direction_normal_dot > 1e-7: raise ValueError( "REVOLVE_AXIS_NOT_IN_SKETCH_PLANE: params.axis.direction must be parallel to " f"sketch.workplane; abs(dot(axis_direction, plane_normal))={direction_normal_dot:.3g}" ) origin_plane_offset = abs(vector_dot(vector_subtract(axis.origin_mm, plane.origin_mm), plane.normal)) if origin_plane_offset > 1e-6: raise ValueError( "REVOLVE_AXIS_NOT_IN_SKETCH_PLANE: params.axis.origin_mm must lie in " f"sketch.workplane; plane_offset_mm={origin_plane_offset:.3g}" ) def _shape_from_primary(node: FeaturePlanNode, session: ExecutionSession, *, sketch: dict[str, Any] | None = None) -> FeatureResult: # 主形状特征（拉伸 / 旋转）的统一入口：由草图生成实体并与当前主体做布尔合并或切除。 # 1. 取草图：优先使用外部传入的 sketch_override（阵列/镜像等重放场景）， # 否则按 sketch_id 从会话草图表中取原始草图。 selected_sketch = sketch or session.sketches.get(str(node.sketch_id)) if selected_sketch is None: raise ValueError("primary feature has no resolved sketch") # 2. 从草图解析闭合轮廓区域（faces），没有闭合区域就无法生成实体。 faces = session.adapter.faces_for_sketch(selected_sketch) if not faces: raise ValueError("sketch does not create a closed profile region") if node.atomic_id == "extrude_add_blind_with_hole": resolved = [session.resolve(selector) for selector in node.selectors] failed = next((item for item in resolved if item.status != "resolved"), None) if failed or len(resolved) != 1 or resolved[0].record is None or resolved[0].record.kind != "face": raise ValueError(failed.diagnostic.message if failed and failed.diagnostic else "profile hole selector is unresolved") if len(faces) != 1: raise ValueError("profile hole extrusion requires exactly one outer sketch region") faces = [session.adapter.face_with_holes(faces[0], [resolved[0].record.value])] # 3. 按特征类型生成子实体： if node.atomic_id.startswith("extrude_"): # 拉伸：先按终止条件（盲孔/贯穿/至面/双侧等）求出位移向量， # 再对每个面沿每个向量做拉伸，得到实体列表。up_to_surface 在 # profile 与目标面非均匀相交时（extent.trim_to 非空）改用裁剪 # 拉伸：穿透后与目标面求交，只保留可达部分（issue #5）。 extents = _extent_vectors(node, faces, selected_sketch, session) draft = node.params.get("draft") taper_deg = 0.0 if isinstance(draft, dict): taper_deg = float(draft["angle_deg"]) if not bool(draft["pull_direction"]): taper_deg = -taper_deg solids: list[Any] = [] for face in faces: for extent in extents: if draft is not None: solids.append(session.adapter.extrude_taper(face, extent.vector, taper_deg)) elif extent.trim_to is None: solids.append(session.adapter.extrude(face, extent.vector)) else: solids.append(session.adapter.extrude_trimmed(face, extent.trim_to, extent.vector)) else: # 旋转：解析旋转轴并校验旋转角，然后绕轴旋转每个面得到实体列表。 axis = _revolve_axis(node, session) _validate_revolve_axis_in_sketch_plane(axis, selected_sketch) angle = float(node.params.get("angle_deg") or 0.0) if angle <= 0: raise ValueError("revolve requires angle_deg > 0") # reverse=true 表示绕轴反向扫掠（SolidWorks 旋转方向反转）：取负 # 旋转角，与 extrude 的 reverse（_extent_vectors 反转拉伸方向）同一 # 语义。profile_schema.json 已声明 revolve.* optional_params 含 # reverse，cdsl_schema.json revolveParams 也已允许，这里补齐 runtime # 侧实现，使三方合同一致。 if bool(node.params.get("reverse")): angle = -angle solids = [session.adapter.revolve(face, angle, axis) for face in faces] # 4. 将所有子实体做布尔并（fuse）合并为一个工具体（tool）。 tool = None for solid in solids: tool = session.adapter.fuse(tool, solid) if tool is None: raise ValueError("primary feature produced no solid") # 5. 与当前主体做布尔操作： if "cut" in node.atomic_id: # 切除类特征：要求已有主体，从主体上减去工具体（cut）。 if session.body is None: raise ValueError("cut feature has no body") body = session.adapter.cut(session.body, tool) members = {node.feature_id: body} elif node.params.get("result_mode") == "new_body": # FeatureScript NEW creates an independent result body even when it # intersects a prior body. Keep both shapes in the exported compound. body = session.adapter.combine(session.body, tool) members = {**session.body_members, node.feature_id: tool} else: # 添加类特征：将工具体并到当前主体上（fuse），首个特征时 body 为 None 也能直接成立。 body = session.adapter.fuse(session.body, tool) members = {node.feature_id: body} # 6. 登记新主体（更新拓扑、记录重放定义），并返回该特征的结果对象。 session.register_body(node.feature_id, body, replay_node=node, body_members=members) return session.result(node) def _execute_revolve_surface(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # Surface revolve 的 profile 是单一闭合 wire。它只生成独立 shell，不能参与 # 当前实体 body 的 fuse/cut，也不能把其结果误报为新的实体 body。 sketch = session.sketches.get(str(node.sketch_id)) if sketch is None: raise ValueError("surface revolve has no resolved sketch") faces = session.adapter.faces_for_sketch(sketch) if len(faces) != 1 or faces[0].inner_wires(): raise ValueError("surface revolve requires exactly one closed profile without holes") axis = _revolve_axis(node, session) _validate_revolve_axis_in_sketch_plane(axis, sketch) angle = float(node.params.get("angle_deg") or 0.0) if angle <= 0: raise ValueError("surface revolve requires angle_deg > 0") if bool(node.params.get("reverse")): angle = -angle surface_id = session.register_surface( node.feature_id, session.adapter.revolve_surface(faces[0].outer_wire(), angle, axis), ) return session.result(node, include_body=False, surface_id=surface_id) def _execute_extrude_surface(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # surfaceEntities 的曲面拉伸沿用实体特征已 lower 的距离，但始终独立登记为 # shell。它既不改变 active solid，也不以曲面参与实体 fuse/cut。 sketch = session.sketches.get(str(node.sketch_id)) if sketch is None: raise ValueError("surface extrude has no resolved sketch") direction = vector_unit(_normal_from_sketch(sketch), field_name="sketch normal") if bool(node.params.get("reverse")): direction = vector_scale(direction, -1) distance = float(node.params.get("distance_mm") or 0.0) if distance <= 0: raise ValueError("surface extrude requires distance_mm > 0") wires = session.adapter.surface_wires_for_sketch(sketch) surface = session.adapter.extrude_surface(wires, vector_scale(direction, distance)) reverse_distance = float(node.params.get("reverse_distance_mm") or 0.0) if reverse_distance > 0: opposite = session.adapter.extrude_surface(wires, vector_scale(direction, -reverse_distance)) surface = session.adapter.combine_surfaces(surface, opposite) surface_id = session.register_surface(node.feature_id, surface) return session.result(node, include_body=False, surface_id=surface_id) def _combine_members(session: ExecutionSession, members: dict[str, Any]) -> Any: body = None for member in members.values(): body = session.adapter.combine(body, member) if body is None: raise ValueError("booleanBodies produced no result bodies") return body def _execute_boolean_bodies(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # booleanBodies 总是作用于 source feature 的明确 body 输出，不能回退为 # 当前聚合 body。这样相邻独立实体不会意外成为工具或目标。 params = node.params target_ids = [str(value) for value in params.get("target_feature_ids") or []] tool_ids = [str(value) for value in params.get("tool_feature_ids") or []] missing = [feature_id for feature_id in target_ids + tool_ids if feature_id not in session.body_members] if missing: raise ValueError("booleanBodies source bodies are unavailable: " + ", ".join(missing)) targets = {feature_id: session.body_members[feature_id] for feature_id in target_ids} tools = {feature_id: session.body_members[feature_id] for feature_id in tool_ids} target = _combine_members(session, targets) tool = _combine_members(session, tools) operation = str(params.get("operation") or "") if operation == "union": result = session.adapter.fuse(target, tool) elif operation == "subtract": result = session.adapter.cut(target, tool) elif operation == "intersect": result = session.adapter.intersect(target, tool) else: raise ValueError(f"unsupported booleanBodies operation {operation!r}") members = { feature_id: body for feature_id, body in session.body_members.items() if feature_id not in set(target_ids + tool_ids) } members[node.feature_id] = result if bool(params.get("keep_tools")): members.update(tools) session.register_body(node.feature_id, _combine_members(session, members), body_members=members) return session.result(node) def _execute_loft_add(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 放样截面不占用 feature.sketch_id；按有序 profile_sketch_ids 取已解析 # 草图，并由 adapter 统一校验单闭环、无内环等内核输入约束。 profile_ids = node.params.get("profile_sketch_ids") or [] profiles: list[dict[str, Any]] = [] for sketch_id in profile_ids: sketch = session.sketches.get(str(sketch_id)) if sketch is None: raise ValueError(f"loft profile sketch {sketch_id!r} is not resolved") profiles.append(sketch) solid = session.adapter.loft(profiles) session.register_body(node.feature_id, session.adapter.fuse(session.body, solid), replay_node=node) return session.result(node) def _execute_loft_add_with_cap_face(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: resolved = [session.resolve(selector) for selector in node.selectors] failed = next((item for item in resolved if item.status != "resolved"), None) if failed or len(resolved) != 1 or resolved[0].record is None or resolved[0].record.kind != "face": raise ValueError(failed.diagnostic.message if failed and failed.diagnostic else "cap-face loft selector is unresolved") profile_ids = node.params.get("profile_sketch_ids") or [] profiles: list[dict[str, Any]] = [] for sketch_id in profile_ids: sketch = session.sketches.get(str(sketch_id)) if sketch is None: raise ValueError(f"loft profile sketch {sketch_id!r} is not resolved") profiles.append(sketch) solid = session.adapter.loft_with_cap_face(resolved[0].record.value, profiles) session.register_body(node.feature_id, session.adapter.fuse(session.body, solid), replay_node=node) return session.result(node) def _sweep_path(node: FeaturePlanNode, session: ExecutionSession) -> Any: # 路径是 self-contained CDSL 数据，避免重放时依赖临时草图或 source id。 path = node.params.get("path") or {} if not isinstance(path, dict): raise ValueError("sweep path must be an object") plane = PlaneSpec.from_mapping(path.get("workplane") or {}) segment = path.get("segment") or {} if not isinstance(segment, dict): raise ValueError("sweep path segment must be an object") kind = str(segment.get("type") or "") if kind == "line": local_points = [segment.get("start"), segment.get("end")] elif kind == "bspline": local_points = segment.get("points") or [] else: raise ValueError(f"unsupported sweep path segment {kind!r}") if len(local_points) < 2 or any(not isinstance(point, list) or len(point) != 2 for point in local_points): raise ValueError("sweep path requires two-dimensional points") def point(value: list[float]) -> Vector3: return vector_add( plane.origin_mm, vector_add(vector_scale(plane.x_dir, float(value[0])), vector_scale(plane.y_dir, float(value[1]))), ) def tangent(value: Any) -> Vector3 | None: if value is None: return None if not isinstance(value, list) or len(value) != 2: raise ValueError("sweep path tangent must contain two coordinates") return vector_add(vector_scale(plane.x_dir, float(value[0])), vector_scale(plane.y_dir, float(value[1]))) return session.adapter.sweep_path( [point(value) for value in local_points], start_tangent=tangent(segment.get("start_tangent")), end_tangent=tangent(segment.get("end_tangent")), parameters=[float(value) for value in segment.get("parameters") or []] or None, ) def _execute_sweep_add(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None = None) -> FeatureResult: profile = sketch or session.sketches.get(str(node.sketch_id)) if profile is None: raise ValueError("sweep has no resolved profile sketch") faces = session.adapter.faces_for_sketch(profile) if len(faces) != 1: raise ValueError("sweep requires exactly one closed profile region") solid = session.adapter.sweep( faces[0], _sweep_path(node, session), is_frenet=bool(node.params.get("is_frenet", False)), ) body = session.adapter.combine(session.body, solid) if node.params.get("result_mode") == "new_body" else session.adapter.fuse(session.body, solid) session.register_body(node.feature_id, body, replay_node=node) return session.result(node) def _execute_reference_plane(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 基准面特征（reference_plane）执行入口：从参数解析平面并登记为拓扑上下文。 # 1. 从特征参数 plane 中解析出平面定义 PlaneSpec（原点到法向）。 plane = PlaneSpec.from_mapping(node.params.get("plane") or {}) # 2. 将该平面注册到拓扑上下文，供后续特征（如草图基准、参考轴）引用。 session.topology.register_context(node.feature_id, plane) # 3. 返回结果对象，并将该平面作为上下文一并携带。 return session.result(node, context=plane) def _execute_reference_axis(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 基准轴特征（reference_axis）执行入口：由参数直接定义轴，或由两个基准平面求交线得到轴。 # 1. 尝试直接取参数：若同时给出原点 origin_mm 与方向 direction，则直接构造轴。 params = node.params.get("axis") or {} if params.get("origin_mm") and params.get("direction"): axis = AxisSpec.from_mapping(params) else: # 2. 否则从特征选择器中筛选出已解析的基准平面。 planes = [session.resolve(selector) for selector in node.selectors if selector.get("kind") == "plane"] resolved = [item.record.value for item in planes if item.status == "resolved" and isinstance(item.record.value, PlaneSpec)] # 3. 校验：轴需要两个非平行的平面，不足两个则报错。 if len(resolved) < 2: raise ValueError("reference axis requires two uniquely resolved planes") # 4. 用两平面法线叉积求交线方向；若方向长度接近 0 说明两平面平行，无法成轴。 first, second = resolved[0], resolved[1] n1, n2 = first.normal, second.normal direction = vector_cross(n1, n2) squared_length = vector_dot(direction, direction) if squared_length <= 1e-18: raise ValueError("reference planes are parallel and cannot define an axis") # 5. 求交线上的一点：两平面到各自原点的垂距参与线性组合，得到交线上的最近点。 d1 = vector_dot(n1, first.origin_mm) d2 = vector_dot(n2, second.origin_mm) point = vector_scale(vector_add(vector_scale(vector_cross(n2, direction), d1), vector_scale(vector_cross(direction, n1), d2)), 1 / squared_length) # 6. 由该点与归一化的交线方向组合成基准轴 AxisSpec。 axis = AxisSpec(origin_mm=point, direction=vector_unit(direction, field_name="reference axis")) # 7. 注册为拓扑上下文，并返回结果对象（携带该轴）。 session.topology.register_context(node.feature_id, axis) return session.result(node, context=axis) def _execute_sphere(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 球体特征（sphere_add）执行入口：按球心与半径生成球体并并入当前主体。 # 1. 解析参数：半径 radius_mm 与球心 center_mm。 radius = float(node.params.get("radius_mm") or 0.0) center = node.params.get("center_mm") or [] # 2. 校验：半径必须大于 0，球心必须是三维坐标。 if radius <= 0 or len(center) != 3: raise ValueError("sphere_add requires radius_mm and a three-dimensional center_mm") # 3. 由适配器创建球体实体。 solid = session.adapter.sphere(radius, (float(center[0]), float(center[1]), float(center[2]))) # 4. 球体与当前主体做布尔并（fuse）后登记为新主体，并返回该特征的结果对象。 session.register_body(node.feature_id, session.adapter.fuse(session.body, solid), replay_node=node) return session.result(node) def _execute_box(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 长方体特征（box_add）执行入口：以几何中心 center_mm 与三向尺寸生成原生长方体。 # 1. 解析并校验尺寸与中心，非法输入抛出带具体原因的 ValueError。 try: length = float(node.params.get("length_mm") or 0.0) width = float(node.params.get("width_mm") or 0.0) height = float(node.params.get("height_mm") or 0.0) center = node.params.get("center_mm") or [] except (TypeError, ValueError) as error: raise ValueError("box dimensions must be numeric") from error if length <= 0 or width <= 0 or height <= 0 or len(center) != 3: raise ValueError("box_add requires positive length_mm/width_mm/height_mm and a three-dimensional center_mm") # 2. 生成世界轴对齐的 plane frame：plane 原点是长方体的最小角点（中心减去半 # 尺寸），长/宽/高分别沿世界 x/y/z 生长（build123d Solid.make_box 语义）。 corner = ( float(center[0]) - length / 2, float(center[1]) - width / 2, float(center[2]) - height / 2, ) plane = PlaneSpec.from_mapping({"origin_mm": corner, "x_dir": [1, 0, 0], "normal": [0, 0, 1]}) solid = session.adapter.box(length, width, height, plane) # 3. 与当前主体做布尔并后登记为新主体，并返回该特征的结果对象。 session.register_body(node.feature_id, session.adapter.fuse(session.body, solid), replay_node=node) return session.result(node) def _execute_cylinder(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 圆柱特征（cylinder_add）执行入口：axis 的原点是底面圆心、方向为轴向； # axis 缺省为世界 +Z 过原点（底面圆心落在 (0,0,0)）。 # 1. 解析并校验半径与高度，非法输入抛出带具体原因的 ValueError。 try: radius = float(node.params.get("radius_mm") or 0.0) height = float(node.params.get("height_mm") or 0.0) except (TypeError, ValueError) as error: raise ValueError("cylinder dimensions must be numeric") from error if radius <= 0 or height <= 0: raise ValueError("cylinder_add requires positive radius_mm and height_mm") raw_axis = node.params.get("axis") if raw_axis is not None and not ( isinstance(raw_axis, dict) and raw_axis.get("origin_mm") is not None and raw_axis.get("direction") is not None ): raise ValueError("cylinder_add axis must define origin_mm and direction") axis = AxisSpec.from_mapping(raw_axis) if isinstance(raw_axis, dict) else None # 2. 由适配器创建原生圆柱（axis=None 即世界 +Z 过原点）。 solid = session.adapter.cylinder(radius, height, axis) # 3. 与当前主体做布尔并后登记为新主体，并返回该特征的结果对象。 session.register_body(node.feature_id, session.adapter.fuse(session.body, solid), replay_node=node) return session.result(node) def _execute_thread(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 螺纹特征（thread_add）执行入口：按规格生成参数化螺纹段并并入当前主体。 # 1. 解析并校验尺寸/牙距/轴，非法输入抛出带具体原因的 ValueError。 spec = ThreadSpec.from_feature(node.atomic_id, node.params) # 2. 由适配器门面生成沿 spec.axis 放置的外螺纹实心段。 solid = session.adapter.thread_solid(spec) # 3. 与当前主体做布尔并（fuse）后登记为新主体，并返回该特征的结果对象。 session.register_body(node.feature_id, session.adapter.fuse(session.body, solid), replay_node=node) return session.result(node) def _thread_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: # 螺纹特征（thread_add/thread_cut）不需要草图平面，丢弃该参数后执行。 # thread_cut 走布尔差分支：从已有主体切出内螺纹槽，而非并入外螺纹段。 del sketch if node.atomic_id == "thread_cut": return _execute_thread_cut(node, session) return _execute_thread(node, session) def _execute_thread_cut(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 内螺纹（thread_cut）执行入口：ThreadSpec.from_feature 对 thread_cut 恒置 # internal=True，生成牙顶外放 INTERNAL_CUT_OVERLAP_MM 的切削刀具，沿 # spec.axis 放置后从当前主体布尔差出全深螺旋牙槽（宿主通常已预打光孔， # 刀具 core 落在孔腔中，仅外放的牙槽层切入孔壁）。 # 1. 解析并校验尺寸/牙距/轴，非法输入抛出带具体原因的 ValueError。 spec = ThreadSpec.from_feature(node.atomic_id, node.params) # 2. 由适配器门面生成沿 spec.axis 放置的内螺纹切削刀具实心段。 tool = session.adapter.thread_solid(spec) # 3. 从当前主体布尔差（cut）后登记为新主体，并返回该特征的结果对象。 session.register_body(node.feature_id, session.adapter.cut(session.body, tool), replay_node=node) return session.result(node) def _host_plane(resolution: SelectorResolution) -> PlaneSpec: if resolution.record is None: raise ValueError(resolution.diagnostic.message if resolution.diagnostic else "host face was not resolved") geometry = resolution.record.geometry return PlaneSpec.from_mapping({ "origin_mm": geometry["center_mm"], "x_dir": [1, 0, 0] if abs(float(geometry["normal"][0])) < 0.9 else [0, 1, 0], "normal": geometry["normal"], }) def _hole_starts( spec: HoleSpec, *, host_plane: PlaneSpec, positions_are_local: bool, ) -> list[Vector3]: starts: list[Vector3] = [] for point in spec.positions_mm: if positions_are_local: start = vector_add( vector_add( vector_add(host_plane.origin_mm, vector_scale(host_plane.x_dir, point[0])), vector_scale(host_plane.y_dir, point[1]), ), vector_scale(host_plane.normal, point[2]), ) else: start = point starts.append(start) return starts def _execute_hole(node: FeaturePlanNode, session: ExecutionSession, *, wizard: bool = False) -> FeatureResult: # 孔特征（hole）执行入口：在指定宿主面上按孔规格生成切除工具，并从主体上减去。 # 1. 校验：孔是切除操作，必须先有主体。 if session.body is None: raise ValueError("hole feature has no body") # 2. 确定宿主面 host_face： host_selector = node.params.get("host_face") if isinstance(host_selector, dict) and isinstance(host_selector.get("frame"), dict): # 若直接带 frame（平面定义），则以该平面为宿主，孔位按局部坐标解释。 host = PlaneSpec.from_mapping(host_selector["frame"]) positions_are_local = True else: # 否则从特征选择器中取 face，解析出宿主平面，孔位按世界坐标解释。 selectors = list(node.selectors) if isinstance(host_selector, dict): selectors.append(host_selector) selector = next((item for item in selectors if item.get("kind") == "face"), None) if selector is None: raise ValueError("hole requires host_face selector or frame") host = _host_plane(session.resolve(selector)) positions_are_local = False # 3. 解析孔规格 HoleSpec（直径、深度、类型等，wizard 模式提供额外默认值）。 spec = HoleSpec.from_feature(node.atomic_id, node.params, wizard=wizard) # 4. A host-face normal is an outward B-rep orientation, so its inverse # always enters the material. Inferring direction from the global body # centre fails for concave or multi-leg parts: for example, the top face # of an L bracket can sit below the whole body's centre and the old rule # drilled outward, producing a no-op feature reported as successful. # The selected topology face is the local, authoritative orientation. inward = vector_scale(host.normal, -1) # 5. 生成孔切除工具：按孔规格、起始位置、内方向及“贯穿到主体底面”的深度构造工具实体。 tool = session.adapter.hole_tool( spec, _hole_starts(spec, host_plane=host, positions_are_local=positions_are_local), inward, session.adapter.body_span(session.body, inward) + 2.0, ) # 6. 从主体上减去工具实体，登记新主体并返回结果。 # thread 是装饰螺纹（无螺距、不进实体几何，SolidWorks/STEP 的螺纹孔 # 即光滑孔）：孔按光滑圆柱孔执行，同时记录 info 级诊断便于批量报告 # 追溯降级数量（issue #9，capabilities 已不再拒绝 thread）。 diagnostics: list[RuntimeDiagnostic] = [] if wizard and node.params.get("thread"): diagnostics.append(RuntimeDiagnostic( code="thread_decoration_ignored", message="Thread decoration is not modeled; the hole falls back to a plain cylindrical bore", feature_id=node.feature_id, )) session.register_body(node.feature_id, session.adapter.cut(session.body, tool), replay_node=node) return session.result(node, diagnostics=diagnostics) def _selector_edges(node: FeaturePlanNode, session: ExecutionSession, *, tangent_propagation: bool = False) -> list[Any]: resolved: list[SelectorResolution] = [session.resolve(selector) for selector in node.selectors] failed = next((item for item in resolved if item.status != "resolved"), None) if failed: raise ValueError(failed.diagnostic.message if failed.diagnostic else "selector resolution failed") def is_body_boundary(edge: Any) -> bool: # 圆柱、圆锥等周期面会带一条仅属于自身的参数 seam。该线不是实体 # 边界；FeatureScript 以 FACE 选择倒角时不应将其当作额外的待倒角边， # 否则连续的锥面会被错误切成两段。显式 EDGE selector 仍可表达真正的 # 单边选择，所以这里只约束由 FACE 展开的候选边。 face_count = sum( 1 for face in session.body.faces() if any(candidate.is_same(edge) for candidate in face.edges()) ) return face_count >= 2 edges: list[Any] = [] for item in resolved: if item.record.kind == "edge": edges.append(item.record.value) elif item.record.kind == "face": edges.extend(edge for edge in item.record.value.edges() if is_body_boundary(edge)) if not edges: raise ValueError("selectors did not resolve any edges") return session.adapter.tangent_edges(session.body, edges) if tangent_propagation else edges def _shell_target(node: FeaturePlanNode, session: ExecutionSession) -> tuple[Any, list[Any]]: # shell 的 remove-face selector 必须全部属于同一实体。CADFS 允许一个 # Compound 中保留多个独立 body，不能将整组 body 交给 OCC 后由内核猜测 # 应抽壳的成员。 resolved = [session.resolve(selector) for selector in node.selectors] failed = next((item for item in resolved if item.status != "resolved"), None) if failed: raise ValueError(failed.diagnostic.message if failed.diagnostic else "selector resolution failed") records = [item.record for item in resolved if item.record is not None] if not records or any(record.kind != "face" for record in records): raise ValueError("shell selectors must resolve to faces") target_ids = {record.body_id for record in records} if len(target_ids) != 1: raise ValueError("shell faces must belong to one target body") target_id = next(iter(target_ids)) members = session.adapter.body_solids(session.body) if len(members) == 1: return members[0], [record.value for record in records] if target_id is None or session.body_id is None: raise ValueError("shell target body is unresolved") prefix = f"{session.body_id}:" if not target_id.startswith(prefix): raise ValueError("shell target body is outside the active body set") try: member_index = int(target_id[len(prefix):]) except ValueError as error: raise ValueError("shell target body has an invalid member id") from error if member_index < 0 or member_index >= len(members): raise ValueError("shell target body member is unavailable") return members[member_index], [record.value for record in records] def _replace_shell_target(session: ExecutionSession, target: Any, replacement: Any) -> Any: # 仅替换抽壳目标实体；其他独立实体保持原样和原有相对顺序。 members = session.adapter.body_solids(session.body) if len(members) == 1: return replacement replaced = False result = None for member in members: if member.is_same(target): result = session.adapter.combine(result, replacement) replaced = True else: result = session.adapter.combine(result, member) if not replaced or result is None: raise ValueError("shell target solid is no longer part of the active body") return result def _execute_shell(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 抽壳特征：移除 selector 所指面，并按 CADFS thickness 向实体内部偏置。 if session.body is None: raise ValueError("shell has no body") thickness = float(node.params.get("thickness_mm") or 0) if thickness <= 0: raise ValueError("shell thickness_mm must be > 0") target, faces = _shell_target(node, session) result = session.adapter.shell(target, faces, thickness, inward=bool(node.params.get("inward", True))) session.register_body(node.feature_id, _replace_shell_target(session, target, result), replay_node=node) return session.result(node) def _execute_fillet(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 圆角特征（fillet）执行入口：对选中边按半径做圆角，平滑尖角与棱边。 # 1. 校验：圆角作用于已有主体，必须先有主体。 if session.body is None: raise ValueError("fillet has no body") # 2. 解析圆角半径并校验必须大于 0。 radius = float(node.params.get("radius_mm") or 0) if radius <= 0: raise ValueError("fillet radius_mm must be > 0") # 3. 解析目标边（支持 tangent_propagation 相切传播），并执行圆角。 body = session.adapter.fillet( session.body, radius, _selector_edges(node, session, tangent_propagation=bool(node.params.get("tangent_propagation"))), ) # 4. 登记新主体并返回结果。 session.register_body(node.feature_id, body, replay_node=node) return session.result(node) def _execute_chamfer(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: # 倒角特征（chamfer）执行入口：对选中边按距离做倒角（可带第二距离形成不对称倒角）。 # 1. 校验：倒角作用于已有主体，必须先有主体。 if session.body is None: raise ValueError("chamfer has no body") # 2. 解析主距离并校验必须大于 0。 distance = float(node.params.get("distance_mm") or 0) if distance <= 0: raise ValueError("chamfer distance_mm must be > 0") # 3. 解析第二距离与角度（importer 对 SolidWorks Distance-Angle 倒角产出 # angle_rad，单位为弧度）。第二距离 = 主距离 * tan(angle)；angle=45° 时 # tan=1，退化为等距倒角（与历史行为一致，零回归）。 # 注意：build123d 的 length/length2 侧向分配依赖面的枚举顺序，对非 45° # 倒角仅保证量级正确，距离所在侧可能反转。 distance_2 = node.params.get("distance_2_mm") angle_rad = node.params.get("angle_rad") if distance_2 is None and angle_rad is not None: distance_2 = distance * math.tan(float(angle_rad)) # 4. 解析目标边（支持相切传播），执行倒角。 edges = _selector_edges(node, session, tangent_propagation=bool(node.params.get("tangent_propagation"))) diagnostics: list[RuntimeDiagnostic] = [] try: body = session.adapter.chamfer(session.body, distance, distance_2, edges) except ValueError as error: # 显式 surfaceEntities 可以在后续实体上留下曲面分区边界。若标准 # OCC 倒角因环域宽度不足而拒绝，只允许在该 shell 给出同轴边界证据 # 时按原始距离构造受限倒角；没有证明时仍保留原始内核失败。 if distance_2 is not None or not session.surface_members: raise try: body = session.adapter.surface_limited_chamfer( session.body, distance, edges, list(session.surface_members.values()), ) except ValueError: raise error diagnostics.append(RuntimeDiagnostic( "chamfer_surface_limited", "Chamfer was limited by an explicit coaxial surface boundary", feature_id=node.feature_id, detail={"distance_mm": distance, "surface_count": len(session.surface_members)}, )) # 5. 登记新主体并返回结果。 session.register_body(node.feature_id, body, replay_node=node) return session.result(node, diagnostics=diagnostics) def _translated_sketch(sketch: dict[str, Any], offset: Vector3) -> dict[str, Any]: output = deepcopy(sketch) components = offset workplane = output.get("workplane") or {} origin = workplane.get("origin_mm") or [0, 0, 0] workplane["origin_mm"] = [float(origin[index]) + components[index] for index in range(3)] output["workplane"] = workplane for key in ("contour_edges_mm", "contour_regions_mm"): def translate(value: Any) -> None: if isinstance(value, dict): for point_key in ("start_mm", "end_mm", "center_mm"): if point_key in value: value[point_key] = [float(value[point_key][index]) + components[index] for index in range(3)] if "points_mm" in value: value["points_mm"] = [ [float(point[index]) + components[index] for index in range(3)] for point in value["points_mm"] ] for child in value.values(): translate(child) elif isinstance(value, list): for child in value: translate(child) translate(output.get(key)) return output def _transformed_loft_profiles( node: FeaturePlanNode, params: dict[str, Any], instance_id: str, session: ExecutionSession, transform: Callable[[dict[str, Any]], dict[str, Any]], ) -> None: """为 pattern replay 创建放样截面的变换副本。""" if node.atomic_id != "loft_add": return profile_ids = params.get("profile_sketch_ids") or [] transformed_ids: list[str] = [] for index, sketch_id in enumerate(profile_ids): source = session.sketches.get(str(sketch_id)) if source is None: raise ValueError(f"loft profile sketch {sketch_id!r} has no replay definition") transformed_id = f"{instance_id}.profile.{index}" # 不复用原 profile：pattern 中的每个截面都必须与 source feature # 使用相同的平移、镜像或旋转，才能保持放样的真实空间位置。 session.sketches[transformed_id] = transform(source) transformed_ids.append(transformed_id) params["profile_sketch_ids"] = transformed_ids def _owner_plane_frame(session: ExecutionSession, selector: dict[str, Any]) -> dict[str, Any] | None: """解析 selector 的 owner 特征（reference_plane）注册的显式平面 frame。 #6 pattern 引用重解析：pattern 重放 source（pattern_mirror）时，镜像面 是 selector，其 owner 是 reference_plane 特征；该特征执行时把显式 PlaneSpec 登记为拓扑上下文，这里取出该 frame 供随实例变换使用。 """ owner = selector.get("owner_feature_id") if not owner: return None for record in session.topology.records_for_feature(str(owner)): if record.kind == "plane" and isinstance(record.value, PlaneSpec): return record.value.as_dict() return None def _translated_node(node: FeaturePlanNode, instance_id: str, offset: Vector3, session: ExecutionSession) -> FeaturePlanNode: params = deepcopy(node.params) components = offset if isinstance(params.get("plane"), dict) and params["plane"].get("origin_mm"): params["plane"]["origin_mm"] = [float(params["plane"]["origin_mm"][index]) + components[index] for index in range(3)] host = params.get("host_face") host_frame = host.get("frame") if isinstance(host, dict) else None positions_are_local = isinstance(host_frame, dict) and all( host_frame.get(key) is not None for key in ("origin_mm", "x_dir", "normal") ) if positions_are_local and host_frame.get("origin_mm"): host_frame["origin_mm"] = [float(host_frame["origin_mm"][index]) + components[index] for index in range(3)] if not positions_are_local: for position in params.get("positions") or []: if position.get("mm"): position["mm"] = [float(position["mm"][index]) + components[index] for index in range(3)] axis = params.get("axis") or {} if axis.get("origin_mm"): axis["origin_mm"] = [float(axis["origin_mm"][index]) + components[index] for index in range(3)] center = params.get("center_mm") if center: # box_add/sphere_add 以世界坐标几何中心定位；平移重放必须随实例移动该中心， # 否则阵列副本会静默重合在原位置。 params["center_mm"] = [float(center[index]) + components[index] for index in range(3)] _transformed_loft_profiles( node, params, instance_id, session, lambda sketch: _translated_sketch(sketch, offset), ) mirror_plane = params.get("mirror_plane") if isinstance(mirror_plane, dict) and node.atomic_id == "pattern_mirror": # #6 pattern 引用重解析：镜像面是 reference_plane 引用，随实例平移 # 到新位置后内联为显式 frame；否则重放时 resolve 到原始面，镜像 # 副本会错误地重合在源特征附近。同时源特征也必须平移后重放：镜像 # 副本 = reflect(源@t, 面@t)，只平移面不平移源会落在 2P+t-x 处 # 而非正确位置 2P-x+t。 frame = _owner_plane_frame(session, mirror_plane) if frame is None: raise ValueError("mirror plane reference cannot be transformed for pattern replay") cloned_selector = deepcopy(mirror_plane) cloned_selector["frame"] = { "origin_mm": [frame["origin_mm"][index] + components[index] for index in range(3)], "x_dir": list(frame["x_dir"]), "normal": list(frame["normal"]), } params["mirror_plane"] = cloned_selector transformed_ids: list[str] = [] for source_id in node.params.get("source_feature_ids") or []: source_node = session.replay_definitions.get(str(source_id)) if source_node is None: raise ValueError(f"mirror pattern source feature {source_id} has no replay definition") temp_id = f"{instance_id}.src.{source_id}" shifted = _translated_node(source_node, temp_id, offset, session) if shifted.sketch_id: source_sketch = session.sketches.get(str(source_node.sketch_id)) if source_sketch is not None: temp_sketch_id = f"{temp_id}.sk" session.sketches[temp_sketch_id] = _translated_sketch(source_sketch, offset) shifted = FeaturePlanNode( shifted.feature_id, shifted.atomic_id, shifted.name, shifted.depends_on, shifted.params, shifted.selectors, temp_sketch_id, shifted.declared_status, shifted.source_feature, ) # 临时 replay 定义同样进入 nodes 表（replay_sources 以此过滤）。 session.nodes[temp_id] = shifted session.replay_definitions[temp_id] = shifted transformed_ids.append(temp_id) params["source_feature_ids"] = transformed_ids return FeaturePlanNode(instance_id, node.atomic_id, node.name, (), params, node.selectors, node.sketch_id, node.declared_status, node.source_feature) def _execute_linear_pattern(node: FeaturePlanNode, session: ExecutionSession, execute: Callable[[FeaturePlanNode, ExecutionSession, dict[str, Any] | None], FeatureResult]) -> FeatureResult: # 线性阵列特征（pattern）执行入口：沿两个方向按数量与间距重放源特征形成阵列。 # 1. 取源特征的 replay 定义（源特征按 feature_id 在会话中登记，供本阵列重放）。 params = node.params sources = session.replay_sources(params.get("source_feature_ids") or []) if not sources: raise ValueError("pattern source features have no replay definitions") # 2. 解析两个方向的实例数量。 count_1 = int(params.get("pattern_count_1") or 1) count_2 = int(params.get("pattern_count_2") or 1) # 3. 解析两个方向的步长向量（方向单位向量 × 间距），作为阵列位移基准。 direction_1 = vector_scale(vector_unit(tuple(float(value) for value in (params.get("direction_1") or [1, 0, 0])), field_name="pattern direction_1"), float(params.get("spacing_1_mm") or 0)) direction_2 = vector_scale(vector_unit(tuple(float(value) for value in (params.get("direction_2") or [0, 1, 0])), field_name="pattern direction_2"), float(params.get("spacing_2_mm") or 0)) # 4. 双重循环生成每个阵列实例（跳过原点 0,0 处，那里是源特征本身）。 for first in range(count_1): for second in range(count_2): if first == 0 and second == 0: continue # 计算当前实例相对源特征的偏移向量。 offset = vector_add(vector_scale(direction_1, first), vector_scale(direction_2, second)) for source in sources: # 逐个源特征克隆并按偏移平移后重放执行（草图也同步平移）。 dependency = pattern_transform_blocker(source) if dependency: raise ValueError(f"pattern source uses an unsupported {dependency}") cloned = _translated_node(source, f"{node.feature_id}.p{first}_{second}.{source.feature_id}", offset, session) sketch = session.sketches.get(str(source.sketch_id)) execute(cloned, session, _translated_sketch(sketch, offset) if sketch else None) # 5. 记录本阵列的 replay 定义：后续阵列若选中本阵列，按定义递归重放， # 而非复制当前主体做近似。 # A later pattern may select this pattern feature. The definition is # replayed recursively, never approximated by copying the current body. session.replay_definitions[node.feature_id] = node return session.result(node) def _reflect_point(point: list[float] | tuple[float, float, float], plane: PlaneSpec, *, vector: bool = False) -> list[float]: value = tuple(float(component) for component in point) offset = value if vector else vector_subtract(value, plane.origin_mm) mirrored = vector_subtract(value, vector_scale(plane.normal, 2 * vector_dot(offset, plane.normal))) return list(mirrored) def _mirrored_sketch(sketch: dict[str, Any], plane: PlaneSpec) -> dict[str, Any]: output = deepcopy(sketch) workplane = output.get("workplane") or {} if workplane.get("origin_mm"): workplane["origin_mm"] = _reflect_point(workplane["origin_mm"], plane) for key in ("x_dir", "y_dir", "normal"): if workplane.get(key): workplane[key] = _reflect_point(workplane[key], plane, vector=True) output["workplane"] = workplane # A reflection reverses handedness. ``PlaneSpec`` reconstructs its local # y direction as normal x x, so keeping the reflected normal means that # local y is the inverse of the reflected source y. Profiles represented # as local circles (rather than already-transformed contour edges) must # therefore invert v to remain at their actual reflected world position. def mirror_local_coordinates(value: Any) -> None: if isinstance(value, dict): for point_key in ("center", "start", "end"): point = value.get(point_key) if isinstance(point, list) and len(point) == 2: value[point_key] = [float(point[0]), -float(point[1])] if isinstance(value.get("points"), list): value["points"] = [ [float(point[0]), -float(point[1])] for point in value["points"] if isinstance(point, list) and len(point) == 2 ] for child in value.values(): mirror_local_coordinates(child) elif isinstance(value, list): for child in value: mirror_local_coordinates(child) mirror_local_coordinates(output.get("entities")) # This is not consumed after sketch resolution, but retaining the same # local semantics makes an overridden sketch safe to inspect or replay. mirror_local_coordinates(output.get("profile")) def mirror(value: Any) -> None: if isinstance(value, dict): for point_key in ("start_mm", "end_mm", "center_mm"): if point_key in value: value[point_key] = _reflect_point(value[point_key], plane) if "points_mm" in value: value["points_mm"] = [_reflect_point(point, plane) for point in value["points_mm"]] if value.get("normal"): value["normal"] = _reflect_point(value["normal"], plane, vector=True) for child in value.values(): mirror(child) elif isinstance(value, list): for child in value: mirror(child) mirror(output.get("contour_edges_mm")) mirror(output.get("contour_regions_mm")) return output def _mirrored_node(node: FeaturePlanNode, instance_id: str, plane: PlaneSpec, session: ExecutionSession) -> FeaturePlanNode: params = deepcopy(node.params) if isinstance(params.get("plane"), dict): for key in ("origin_mm", "x_dir", "y_dir", "normal"): if params["plane"].get(key): params["plane"][key] = _reflect_point(params["plane"][key], plane, vector=key != "origin_mm") host = params.get("host_face") host_frame = host.get("frame") if isinstance(host, dict) else None positions_are_local = isinstance(host_frame, dict) and all( host_frame.get(key) is not None for key in ("origin_mm", "x_dir", "normal") ) if positions_are_local: for key in ("origin_mm", "x_dir", "normal"): if host_frame.get(key): host_frame[key] = _reflect_point(host_frame[key], plane, vector=key != "origin_mm") # #1 y_dir 保留：PlaneSpec 现在会尊重显式正交 y_dir。镜像后 frame 的 # canonical y 轴必须是 n×x（x 已反射 → y 反转），否则反射后的 frame # 会保留反射前的 y_dir，与下方"局部坐标 v 取反"双重翻转。 x_reflected = host_frame.get("x_dir") n_reflected = host_frame.get("normal") if x_reflected is not None and n_reflected is not None: host_frame["y_dir"] = [ n_reflected[1] * x_reflected[2] - n_reflected[2] * x_reflected[1], n_reflected[2] * x_reflected[0] - n_reflected[0] * x_reflected[2], n_reflected[0] * x_reflected[1] - n_reflected[1] * x_reflected[0], ] # See _mirrored_sketch: the canonical reflected plane reverses local # y, so local hole coordinates must do the same. for position in params.get("positions") or []: point = position.get("mm") if isinstance(point, list) and len(point) == 3: position["mm"] = [float(point[0]), -float(point[1]), float(point[2])] else: for position in params.get("positions") or []: if position.get("mm"): position["mm"] = _reflect_point(position["mm"], plane) axis = params.get("axis") or {} if axis.get("origin_mm"): axis["origin_mm"] = _reflect_point(axis["origin_mm"], plane) if axis.get("direction"): axis["direction"] = _reflect_point(axis["direction"], plane, vector=True) center = params.get("center_mm") if isinstance(center, list) and len(center) == 3: # box_add/sphere_add 以世界坐标几何中心定位：反射该中心即可。box_add 固定 # 世界轴对齐，跨坐标平面镜像后仍保持朝向（斜镜像面在 _execute_mirror_pattern # 中已被显式拒绝）。 params["center_mm"] = _reflect_point(center, plane) _transformed_loft_profiles( node, params, instance_id, session, lambda sketch: _mirrored_sketch(sketch, plane), ) mirror_plane = params.get("mirror_plane") if isinstance(mirror_plane, dict) and node.atomic_id == "pattern_mirror": # #6 pattern 引用重解析：镜像重放 mirror source 时，其镜像面引用 # 随本实例的镜像面一起反射（内联为显式 frame），否则重放 resolve # 到原始面，嵌套镜像会退化成与源镜像重合的错误几何。源特征同样 # 反射后重放：镜像副本 = reflect(源@P_B, reflect(面,P_B))。 frame = _owner_plane_frame(session, mirror_plane) if frame is None: raise ValueError("mirror plane reference cannot be transformed for pattern replay") cloned_selector = deepcopy(mirror_plane) cloned_selector["frame"] = { "origin_mm": _reflect_point(frame["origin_mm"], plane), "x_dir": _reflect_point(frame["x_dir"], plane, vector=True), "normal": _reflect_point(frame["normal"], plane, vector=True), } params["mirror_plane"] = cloned_selector transformed_ids: list[str] = [] for source_id in node.params.get("source_feature_ids") or []: source_node = session.replay_definitions.get(str(source_id)) if source_node is None: raise ValueError(f"mirror pattern source feature {source_id} has no replay definition") temp_id = f"{instance_id}.src.{source_id}" shifted = _mirrored_node(source_node, temp_id, plane, session) if shifted.sketch_id: source_sketch = session.sketches.get(str(source_node.sketch_id)) if source_sketch is not None: temp_sketch_id = f"{temp_id}.sk" session.sketches[temp_sketch_id] = _mirrored_sketch(source_sketch, plane) shifted = FeaturePlanNode( shifted.feature_id, shifted.atomic_id, shifted.name, shifted.depends_on, shifted.params, shifted.selectors, temp_sketch_id, shifted.declared_status, shifted.source_feature, ) # 临时 replay 定义同样进入 nodes 表（replay_sources 以此过滤）。 session.nodes[temp_id] = shifted session.replay_definitions[temp_id] = shifted transformed_ids.append(temp_id) params["source_feature_ids"] = transformed_ids return FeaturePlanNode(instance_id, node.atomic_id, node.name, (), params, node.selectors, node.sketch_id, node.declared_status, node.source_feature) def _normal_is_coordinate_axis(normal: Any) -> bool: # 判断单位法向是否平行于任一世界坐标轴：跨这样的平面镜像会保持轴对齐朝向。 return ( isinstance(normal, (list, tuple)) and len(normal) == 3 and any(abs(float(normal[index])) > 1 - 1e-9 for index in range(3)) ) def _execute_mirror_pattern(node: FeaturePlanNode, session: ExecutionSession) -> FeatureResult: mirror = node.params.get("mirror_plane") or {} resolution = session.resolve(mirror) if resolution.status != "resolved" or not isinstance(resolution.record.value, PlaneSpec): raise ValueError(resolution.diagnostic.message if resolution.diagnostic else "mirror plane was not resolved") if node.params.get("mirror_current_body"): # CADFS SWEPT_BODY 表示被后续 feature 持续修改的同一实体。这里复制 # 当前 B-rep 再镜像并合并，不能重放其初始 additive feature，否则会 # 丢失后续 cut/fillet 并生成独立错误实体。 if session.body is None: raise ValueError("mirror current body has no active body") mirrored = session.adapter.mirror(session.body, resolution.record.value) session.register_body(node.feature_id, session.adapter.fuse(session.body, mirrored), replay_node=node) return session.result(node) sources = session.replay_sources(node.params.get("source_feature_ids") or []) if not sources: raise ValueError("mirror pattern source features have no replay definitions") for source in sources: dependency = pattern_transform_blocker(source) if dependency: raise ValueError(f"mirror pattern source uses an unsupported {dependency}") if source.atomic_id == "box_add" and not _normal_is_coordinate_axis(resolution.record.value.normal): # box_add 是固定世界轴对齐的原生图元：跨非坐标平面镜像会产生倾斜朝向， # 当前参数语义无法表达，静默重放会得到错误几何 → 明确拒绝。跨坐标平面 # （法向平行于任一坐标轴）的镜像仍然精确。 raise ValueError("box_add mirror is exact only across coordinate-aligned mirror planes") cloned = _mirrored_node(source, f"{node.feature_id}.m.{source.feature_id}", resolution.record.value, session) sketch = session.sketches.get(str(source.sketch_id)) _execute_node(cloned, session, _mirrored_sketch(sketch, resolution.record.value) if sketch else None) session.replay_definitions[node.feature_id] = node return session.result(node) def _coordinate_axis_direction(direction: Any) -> bool: # 判断方向是否平行于任一世界坐标轴（circular 的 box 限制用）。 # 不依赖输入已是单位向量：非零向量至多一个分量非零即为坐标轴方向 # （_box_circular_is_exact 对任意长度/含小残差的 direction 都稳健）。 if not isinstance(direction, (list, tuple)) or len(direction) != 3: return False return sum(1 for component in direction if abs(float(component)) > 1e-9) == 1 def _rotated_vector(value: Vector3, axis: AxisSpec, angle_rad: float) -> Vector3: # Rodrigues 旋转公式：绕单位轴 axis.direction 旋转向量（无平移项）。 cosine = math.cos(angle_rad) sine = math.sin(angle_rad) axis_direction = axis.direction cross = vector_cross(axis_direction, value) dot = vector_dot(axis_direction, value) return tuple( # type: ignore[return-value] value[index] * cosine + cross[index] * sine + axis_direction[index] * dot * (1.0 - cosine) for index in range(3) ) def _rotated_point(point: Any, axis: AxisSpec, angle_rad: float) -> list[float]: # 绕轴旋转三维点：先平移到轴原点、旋转向量、再平移回。 value = tuple(float(component) for component in point) relative = vector_subtract(value, axis.origin_mm) rotated = _rotated_vector(relative, axis, angle_rad) return [axis.origin_mm[index] + rotated[index] for index in range(3)] def _rotated_sketch(sketch: dict[str, Any], axis: AxisSpec, angle_rad: float) -> dict[str, Any]: # 环形阵列实例的草图：工作平面 frame（原点为点、x/y/normal 为向量）绕轴旋转； # 2D 局部实体坐标不动（frame 旋转后由草图求解器映射到新世界位置）。与 # _translated_sketch 对"世界坐标轮廓点"的处理对称，这里把 start/end/center # 世界坐标点和圆弧法向绕轴旋转。 output = deepcopy(sketch) workplane = output.get("workplane") or {} if workplane.get("origin_mm"): workplane["origin_mm"] = _rotated_point(workplane["origin_mm"], axis, angle_rad) for key in ("x_dir", "y_dir", "normal"): if workplane.get(key): workplane[key] = list(_rotated_vector(tuple(float(v) for v in workplane[key]), axis, angle_rad)) output["workplane"] = workplane def rotate(value: Any) -> None: if isinstance(value, dict): for point_key in ("start_mm", "end_mm", "center_mm"): if point_key in value: value[point_key] = _rotated_point(value[point_key], axis, angle_rad) if "points_mm" in value: value["points_mm"] = [_rotated_point(point, axis, angle_rad) for point in value["points_mm"]] if "normal" in value: value["normal"] = list(_rotated_vector(tuple(float(v) for v in value["normal"]), axis, angle_rad)) for child in value.values(): rotate(child) elif isinstance(value, list): for child in value: rotate(child) for key in ("contour_edges_mm", "contour_regions_mm"): rotate(output.get(key)) return output def _rotated_node(node: FeaturePlanNode, instance_id: str, axis: AxisSpec, angle_rad: float, session: ExecutionSession) -> FeaturePlanNode: # 环形阵列实例节点：把源特征的全部绝对坐标参数绕 axis 旋转（参数键布局与 # _translated_node/_mirrored_node 一致）。workplane/宿主 frame 的轴方向旋转， # 世界坐标点旋转；局部 positions（随宿主 frame）不动。特征自带 axis（圆柱轴/ # 旋转轴/嵌套 circular 轴）与几何中心 center_mm 随实例旋转。嵌套 pattern # source（pattern_mirror/pattern_circular）带绝对引用：镜像面 frame / 内层 # 源需连同本实例一起旋转，否则重放会退化成与源重合的错误几何。 params = deepcopy(node.params) plane = params.get("plane") if isinstance(plane, dict): for key in ("origin_mm", "x_dir", "y_dir", "normal"): if plane.get(key): if key == "origin_mm": plane[key] = _rotated_point(plane[key], axis, angle_rad) else: plane[key] = list(_rotated_vector(tuple(float(v) for v in plane[key]), axis, angle_rad)) path = params.get("path") path_plane = path.get("workplane") if isinstance(path, dict) else None if isinstance(path_plane, dict): if path_plane.get("origin_mm"): path_plane["origin_mm"] = _rotated_point(path_plane["origin_mm"], axis, angle_rad) for key in ("x_dir", "y_dir", "normal"): if path_plane.get(key): path_plane[key] = list(_rotated_vector(tuple(float(v) for v in path_plane[key]), axis, angle_rad)) host = params.get("host_face") host_frame = host.get("frame") if isinstance(host, dict) else None positions_are_local = isinstance(host_frame, dict) and all( host_frame.get(key) is not None for key in ("origin_mm", "x_dir", "normal") ) if positions_are_local: if host_frame.get("origin_mm"): host_frame["origin_mm"] = _rotated_point(host_frame["origin_mm"], axis, angle_rad) for key in ("x_dir", "normal"): if host_frame.get(key): host_frame[key] = list(_rotated_vector(tuple(float(v) for v in host_frame[key]), axis, angle_rad)) else: for position in params.get("positions") or []: if position.get("mm"): position["mm"] = _rotated_point(position["mm"], axis, angle_rad) feature_axis = params.get("axis") if isinstance(feature_axis, dict): if feature_axis.get("origin_mm"): feature_axis["origin_mm"] = _rotated_point(feature_axis["origin_mm"], axis, angle_rad) if feature_axis.get("direction"): feature_axis["direction"] = list(_rotated_vector(tuple(float(v) for v in feature_axis["direction"]), axis, angle_rad)) center = params.get("center_mm") if isinstance(center, list) and len(center) == 3: params["center_mm"] = _rotated_point(center, axis, angle_rad) _transformed_loft_profiles( node, params, instance_id, session, lambda sketch: _rotated_sketch(sketch, axis, angle_rad), ) if node.atomic_id in {"pattern_mirror", "pattern_circular"}: # pattern 引用旋转重解析：镜像面 / 内层源随本实例一起旋转，否则嵌套 # pattern 作为 circular source 时重放会退化成错误几何（见 _translated_node）。 if node.atomic_id == "pattern_mirror": mirror_plane = params.get("mirror_plane") if not isinstance(mirror_plane, dict): raise ValueError("mirror pattern replayed by circular pattern has no mirror plane reference") frame = _owner_plane_frame(session, mirror_plane) if frame is None: raise ValueError("mirror plane reference cannot be transformed for circular pattern replay") cloned_selector = deepcopy(mirror_plane) cloned_selector["frame"] = { "origin_mm": _rotated_point(frame["origin_mm"], axis, angle_rad), "x_dir": list(_rotated_vector(tuple(frame["x_dir"]), axis, angle_rad)), "normal": list(_rotated_vector(tuple(frame["normal"]), axis, angle_rad)), } params["mirror_plane"] = cloned_selector transformed_ids: list[str] = [] for source_id in node.params.get("source_feature_ids") or []: source_node = session.replay_definitions.get(str(source_id)) if source_node is None: raise ValueError(f"pattern source feature {source_id} has no replay definition") temp_id = f"{instance_id}.src.{source_id}" shifted = _rotated_node(source_node, temp_id, axis, angle_rad, session) if shifted.sketch_id: source_sketch = session.sketches.get(str(source_node.sketch_id)) if source_sketch is not None: temp_sketch_id = f"{temp_id}.sk" session.sketches[temp_sketch_id] = _rotated_sketch(source_sketch, axis, angle_rad) shifted = FeaturePlanNode( shifted.feature_id, shifted.atomic_id, shifted.name, shifted.depends_on, shifted.params, shifted.selectors, temp_sketch_id, shifted.declared_status, shifted.source_feature, ) # 临时 replay 定义同样进入 nodes 表（replay_sources 以此过滤）。 session.nodes[temp_id] = shifted session.replay_definitions[temp_id] = shifted transformed_ids.append(temp_id) params["source_feature_ids"] = transformed_ids return FeaturePlanNode(instance_id, node.atomic_id, node.name, (), params, node.selectors, node.sketch_id, node.declared_status, node.source_feature) def _pattern_operation_node(node: FeaturePlanNode, operation_mode: str) -> FeaturePlanNode: # REMOVE pattern 的实例必须沿用 source 的 profile/extent，但以 cut 而不是 # add 写入当前主体。lowering 已将初始 source 同步改写，运行时保留此处以 # 支持完整的 CDSL replay contract。 if operation_mode != "remove": return node atomic_id = { "extrude_add_blind": "extrude_cut_blind", "extrude_add_two_sided": "extrude_cut_two_sided", "revolve_add": "revolve_cut", }.get(node.atomic_id, node.atomic_id) if atomic_id == node.atomic_id and "cut" not in atomic_id: raise ValueError("REMOVE pattern source is not a replayable cutting feature") params = {key: value for key, value in node.params.items() if key != "result_mode"} return FeaturePlanNode( node.feature_id, atomic_id, node.name, node.depends_on, params, node.selectors, node.sketch_id, node.declared_status, node.source_feature, ) def _execute_circular_pattern(node: FeaturePlanNode, session: ExecutionSession, execute: Callable[[FeaturePlanNode, ExecutionSession, dict[str, Any] | None], FeatureResult]) -> FeatureResult: # 环形阵列特征（pattern_circular）执行入口：绕显式轴按数量与包角重放源特征 # 形成环形阵列。源特征整体绕轴旋转（绝对坐标变换），非复制当前主体的近似。 params = node.params raw_axis = params.get("axis") if not (isinstance(raw_axis, dict) and raw_axis.get("origin_mm") is not None and raw_axis.get("direction") is not None): raise ValueError("circular pattern requires an explicit axis with origin_mm and direction") axis = AxisSpec.from_mapping(raw_axis) count = int(params.get("pattern_count") or 1) if count < 1: raise ValueError("circular pattern pattern_count must be >= 1") sweep_angle_deg = float(params.get("sweep_angle_deg") or 360.0) operation_mode = str(params.get("operation_mode") or "add") if operation_mode not in {"add", "remove"}: raise ValueError("circular pattern operation_mode must be add or remove") excluded = {int(value) for value in params.get("excluded_instance_indices") or []} if any(instance < 1 or instance >= count for instance in excluded): raise ValueError("circular pattern excluded instance is outside the generated range") sources = session.replay_sources(params.get("source_feature_ids") or []) if not sources: raise ValueError("circular pattern source features have no replay definitions") for instance in range(1, count): if instance in excluded: continue # 实例 i 位于包角 sweep_angle_deg 的 i/count 处（i=0 即源特征本身）。 angle_deg = sweep_angle_deg * instance / count angle_rad = math.radians(angle_deg) for source in sources: dependency = pattern_transform_blocker(source) if dependency: raise ValueError(f"circular pattern source uses an unsupported {dependency}") if source.atomic_id == "box_add" and not _box_circular_is_exact(axis, angle_rad): raise ValueError( "box_add circular pattern is exact only for coordinate-axis rotation " "by multiples of 180 degrees" ) cloned = _rotated_node(source, f"{node.feature_id}.c{instance}.{source.feature_id}", axis, angle_rad, session) cloned = _pattern_operation_node(cloned, operation_mode) # CADFS pattern instances are copies of the source result, not # independent `NEW` operations. Replay them through normal add # semantics: intersecting or face-sharing instances fuse, while # spatially separate copies remain separate solids in the result. if cloned.params.get("result_mode") == "new_body": cloned = FeaturePlanNode( cloned.feature_id, cloned.atomic_id, cloned.name, cloned.depends_on, {key: value for key, value in cloned.params.items() if key != "result_mode"}, cloned.selectors, cloned.sketch_id, cloned.declared_status, cloned.source_feature, ) sketch = session.sketches.get(str(source.sketch_id)) execute(cloned, session, _rotated_sketch(sketch, axis, angle_rad) if sketch else None) # 环形阵列本身是完整 B-rep 结果的 producer。每个 replay 子特征都会更新 # active body；循环结束后必须用 pattern feature 重新登记最终快照，否则后续 # selector binding 会只保留最后一个实例的 body id，漏掉其它 COPY 实例。 if session.body is None: raise ValueError("circular pattern produced no body") session.register_body(node.feature_id, session.body, replay_node=node) return session.result(node) def _box_circular_is_exact(axis: AxisSpec, angle_rad: float) -> bool: # box_add 是固定世界轴对齐的原生图元：绕轴旋转任意角度会使其棱偏离坐标轴， # 当前参数语义无法表达 → 仅坐标轴旋转且每份转角为 180° 的整数倍时精确 # （180° 翻转把轴对齐 box 映射回轴对齐 box）。与 _execute_mirror_pattern 的 # box 坐标平面限制同思路：宁可显式拒绝，也不静默产出错误几何。 if not _coordinate_axis_direction(axis.direction): return False half_turns = abs(math.degrees(angle_rad)) / 180.0 return abs(half_turns - round(half_turns)) < 1e-9 def _circular_pattern_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_circular_pattern(node, session, _execute_node) def _execute_node(node: FeaturePlanNode, session: ExecutionSession, sketch_override: dict[str, Any] | None = None) -> FeatureResult: executor = EXECUTORS.get(node.atomic_id) if executor is None: raise ValueError(f"No executor registered for {node.atomic_id!r}") previous_feature_id = session.active_feature_id session.active_feature_id = node.feature_id try: return executor(node, session, sketch_override) finally: session.active_feature_id = previous_feature_id ExecutorFunction = Callable[[FeaturePlanNode, ExecutionSession, dict[str, Any] | None], FeatureResult] def _primary_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: return _shape_from_primary(node, session, sketch=sketch) def _revolve_surface_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_revolve_surface(node, session) def _extrude_surface_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_extrude_surface(node, session) def _loft_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_loft_add(node, session) def _loft_cap_face_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_loft_add_with_cap_face(node, session) def _sweep_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: return _execute_sweep_add(node, session, sketch) def _reference_plane_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_reference_plane(node, session) def _reference_axis_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_reference_axis(node, session) def _sphere_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_sphere(node, session) def _box_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_box(node, session) def _cylinder_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_cylinder(node, session) def _hole_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_hole(node, session) def _hole_wizard_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_hole(node, session, wizard=True) def _fillet_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_fillet(node, session) def _chamfer_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_chamfer(node, session) def _shell_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_shell(node, session) def _linear_pattern_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_linear_pattern(node, session, _execute_node) def _mirror_pattern_executor(node: FeaturePlanNode, session: ExecutionSession, sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_mirror_pattern(node, session) EXECUTORS: dict[str, ExecutorFunction] = { "reference_plane": _reference_plane_executor, "reference_axis": _reference_axis_executor, "sphere_add": _sphere_executor, "box_add": _box_executor, "cylinder_add": _cylinder_executor, "thread_add": _thread_executor, "thread_cut": _thread_executor, "extrude_add_blind": _primary_executor, "extrude_add_blind_with_hole": _primary_executor, "extrude_add_two_sided": _primary_executor, "extrude_cut_blind": _primary_executor, "extrude_cut_two_sided": _primary_executor, "extrude_cut_through": _primary_executor, "loft_add": _loft_executor, "loft_add_with_cap_face": _loft_cap_face_executor, "sweep_add": _sweep_executor, "revolve_add": _primary_executor, "revolve_cut": _primary_executor, "revolve_surface": _revolve_surface_executor, "extrude_surface": _extrude_surface_executor, "hole_blind": _hole_executor, "hole_countersink": _hole_executor, "hole_counterbore": _hole_executor, "hole_wizard": _hole_wizard_executor, "fillet": _fillet_executor, "chamfer": _chamfer_executor, "shell": _shell_executor, "boolean_bodies": lambda node, session, sketch: _execute_boolean_bodies(node, session), "pattern_linear": _linear_pattern_executor, "pattern_mirror": _mirror_pattern_executor, "pattern_circular": _circular_pattern_executor, } def analyze_cdsl(cdsl: dict[str, Any]): """Resolve profiles and return the current runtime capability analysis.""" sketch_errors: dict[str, str] = {} resolved = resolve_required_sketches( deepcopy(cdsl), sketch_ids_required_by_contract(cdsl), errors=sketch_errors, ) analyzer = CapabilityAnalyzer(atomic_ids=EXECUTORS, profile_types=CORE_SHAPE_GENERATORS) return analyzer.analyze(resolved, sketch_errors=sketch_errors) def rebuild_cdsl(cdsl: dict[str, Any], out_step: Path, *, strict: bool = True) -> dict[str, Any]: """Rebuild CDSL through session-scoped atomic executors only.""" sketch_errors: dict[str, str] = {} resolved = resolve_required_sketches( deepcopy(cdsl), sketch_ids_required_by_contract(cdsl), errors=sketch_errors, ) analysis = CapabilityAnalyzer(atomic_ids=EXECUTORS, profile_types=CORE_SHAPE_GENERATORS).analyze( resolved, sketch_errors=sketch_errors, ) if strict and not analysis.runtime_eligible: first = next((result for result in analysis.feature_results if not result.executable), None) if first is None: raise ValueError(analysis.document_blockers[0].code) if any(blocker.code == "unknown_atomic" for blocker in first.blockers): raise ValueError(f"unsupported atomic_id: {first.atomic_id}") detail = "; ".join(blocker.code for blocker in first.blockers) raise ValueError(f"Feature {first.feature_id} is not runtime eligible: {detail}") session = ExecutionSession( sketches={str(sketch.get("id")): sketch for sketch in (resolved.get("geometry") or {}).get("sketches") or []}, nodes={node.feature_id: node for node in analysis.plan}, ) diagnostics: list[RuntimeDiagnostic] = [] for node, preflight in zip(analysis.plan, analysis.feature_results): if not preflight.executable: diagnostics.extend(preflight.blockers) if strict: break continue try: _execute_node(node, session) except Exception as error: failed_resolution = next( (item for item in reversed(session.selector_resolutions) if item["status"] != "resolved"), None, ) diagnostic = ( RuntimeDiagnostic(error.code, str(error), feature_id=node.feature_id, detail=error.detail) if isinstance(error, FeatureExecutionError) else RuntimeDiagnostic( failed_resolution["diagnostic"]["code"], failed_resolution["diagnostic"]["message"], feature_id=node.feature_id, detail=failed_resolution["diagnostic"].get("detail") or {}, ) if failed_resolution and failed_resolution.get("diagnostic") else RuntimeDiagnostic("execution_failed", str(error), feature_id=node.feature_id) ) diagnostics.append(diagnostic) if strict: raise RuntimeExecutionError(diagnostic, list(session.selector_resolutions)) from error output = session.body surface_geometry: dict[str, Any] | None = None if output is None: if not session.surface_members: raise ValueError("CDSL execution produced no body") # 纯曲面文档没有 active solid，但依然是可执行的 CAD 结果。只有在 # 没有实体时才将 surface members 作为 STEP 输出，混合模型继续只导出 # 实体，避免曲面意外改变既有实体比较和下游消费语义。 output = session.adapter.combine_surfaces(*session.surface_members.values()) surface_geometry = session.adapter.surface_geometry(output) out_step.parent.mkdir(parents=True, exist_ok=True) session.adapter.export(output, str(out_step)) geometry = session.adapter.body_geometry(session.body) if session.body is not None else surface_geometry if geometry is None: raise ValueError("CDSL execution produced no exportable geometry") bbox = geometry["bbox_mm"] return { "engine": "cdsl_session_runtime", "out_step": str(out_step), "volume_mm3": float(geometry.get("volume_mm3") or 0.0), "bbox_mm": {"min": bbox[:3], "max": bbox[3:]}, # #7 multi-body：重建结果里的独立实体数（Compound 成员数）， # 与 batch 验证的 document_truth.geometry.solid_body_count 对齐。 "solid_count": len(session.adapter.body_solids(session.body)) if session.body is not None else 0, "surface_count": len(session.surface_members), "surface_face_count": int(surface_geometry["face_count"]) if surface_geometry is not None else 0, "surface_area_mm2": float(surface_geometry["area_mm2"]) if surface_geometry is not None else 0.0, "feature_results": [result.as_dict() for result in session.results.values()], "runtime_diagnostics": [diagnostic.as_dict() for diagnostic in diagnostics], "topology_records": [record.public_dict() for record in session.topology.records()], "selector_resolution": session.selector_resolutions, }