"""Helpers shared by executor family modules. Every function here is imported by two or more executor modules. Anything used by exactly one family lives in that family's module instead. """ from __future__ import annotations from dataclasses import replace import math from typing import TYPE_CHECKING, Any, Callable from ..extents import _extent_vectors_from_normal, _normal_from_sketch from ..runtime_base import ExtentVector, FeatureExecutionError from ..specs import AxisSpec, HoleSpec, PlaneSpec, Vector3, pattern_instance_member_id, transform_copy_member_id, vector_add, vector_cross, vector_dot, vector_scale, vector_subtract, vector_unit from ..topology import FeaturePlanNode, FeatureResult, RuntimeDiagnostic, SelectorResolution, TopologyDelta, TopologyDeltaRelation, TopologyRecord if TYPE_CHECKING: # pragma: no cover - import for type checkers only from ..session import ExecutionSession 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 _cut_explicit_body_members( session: "ExecutionSession", tool: Any, ) -> tuple[dict[str, Any], tuple[TopologyDelta, ...]]: """Apply a cut to each independently owned body without erasing ownership. A CADFS NEW body stays independently addressable even when a later REMOVE feature affects several active bodies. Cutting the aggregate first loses that identity, so this path uses the equivalent per-member set difference and drops only members that the tool removes completely. Each non-empty member result retains its own OCC builder history; callers may compose those disjoint exact relations into one operation-wide delta. """ members: dict[str, Any] = {} deltas: list[TopologyDelta] = [] for feature_id, body in session.body_members.items(): result, delta = session.adapter.cut_with_topology_delta(body, tool) if delta is not None: deltas.append(delta) if result is not None and abs(float(result.volume)) > 1e-12: members[feature_id] = result return members, tuple(deltas) def _compose_member_cut_deltas(deltas: tuple[TopologyDelta, ...]) -> TopologyDelta | None: """Combine exact independent member-cut histories without inventing links.""" if not deltas or any(delta.history_status != "proven" for delta in deltas): return None return TopologyDelta( operation="subtract", relations=tuple(relation for delta in deltas for relation in delta.relations), section_values=tuple(value for delta in deltas for value in delta.section_values), section_relations=tuple(relation for delta in deltas for relation in delta.section_relations), blend_relations=tuple(relation for delta in deltas for relation in delta.blend_relations), history_reason="per_member_exact_cut_history", ) def _can_register_primary_cut_tool_history( session: "ExecutionSession", tool: Any, topology_delta: TopologyDelta | None, topology_anchors: list[TopologyRecord] | None, ) -> bool: """Return whether a primary REMOVE can retain a transient tool snapshot. The implicit CADFS primary boolean normally has no independently active tool body. It may retain its direct-prism input snapshot whenever the tool, profile anchors, and prism builder history are singular and exact. The target may contain several independent members: that changes the cut result cardinality but not the identity of the one transient tool input. This gate does not make any multi-member selector executable; a later resolver must still prove its complete operation-wide relation and active member. The cut builder can independently prove a target-side continuation even when a trimmed or fallback tool has no direct-prism history. """ if ( session.body is None or topology_delta is None or topology_delta.operation != "extrude" or topology_delta.history_status != "proven" or not topology_delta.relations or not topology_anchors or len(session.adapter.body_solids(tool)) != 1 ): return False return all( anchor.kind in {"edge", "vertex"} and (anchor.source_entity is not None or anchor.source_entities) for anchor in topology_anchors ) def _extruded_tool( node: FeaturePlanNode, faces: list[Any], profile_normal: Vector3, session: "ExecutionSession", *, record_multiface_prism_history: bool = False, use_operation_wide_prism_history: bool = False, ) -> tuple[Any, TopologyDelta | None]: """Build an extrude tool, retaining complete direct builder history.""" extents = _extent_vectors_from_normal(node, faces, profile_normal, 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 if ( use_operation_wide_prism_history and draft is None and len(faces) > 1 and len(extents) == 1 and extents[0].trim_to is None ): composed = session.adapter.extrude_faces_with_composed_topology_delta(faces, extents[0].vector) if composed is not None: return composed topology_deltas: list[tuple[int, TopologyDelta]] = [] solids: list[Any] = [] exact_two_sided_prism = ( node.atomic_id in {"extrude_add_two_sided", "extrude_cut_two_sided"} and draft is None and len(faces) == 1 and len(extents) == 2 and all(extent.trim_to is None for extent in extents) ) for face in faces: for extent_index, extent in enumerate(extents): if draft is not None: if len(faces) == 1 and len(extents) == 1: solid, topology_delta = session.adapter.extrude_taper_with_topology_delta( face, extent.vector, taper_deg, ) if topology_delta is not None: topology_deltas.append((extent_index, topology_delta)) solids.append(solid) else: solids.append(session.adapter.extrude_taper(face, extent.vector, taper_deg)) elif extent.trim_to is None and (len(extents) == 1 or exact_two_sided_prism) and ( len(faces) == 1 or record_multiface_prism_history ): # Each independently constructed profile face has its own OCC # prism history only when the adapter retained direct source # anchors for every participating profile path. Other complex # multi-face profiles keep the established general-extrude # path; forcing them through MakePrism can make a previously # executable profile invalid without adding usable evidence. try: solid, topology_delta = session.adapter.extrude_with_topology_delta(face, extent.vector) except ValueError: # Keep the established executable profile result when a # selected IMPRINT region is valid as a face but cannot be # a standalone valid prism. Its later fuse may still be # valid. There is no complete builder witness in this # case, so the entire multi-region topology delta is # withheld below instead of mixing proven and guessed # source anchors. solid = session.adapter.extrude(face, extent.vector) else: topology_deltas.append((extent_index, topology_delta)) solids.append(solid) 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)) tool = None for solid in solids: tool = session.adapter.fuse(tool, solid) if tool is None: raise ValueError("extrude produced no solid") if len(topology_deltas) != len(solids): return tool, None relations: list[TopologyDeltaRelation] = [] for extent_index, topology_delta in topology_deltas: for relation in topology_delta.relations: # Both prism builders start on the source plane. Its two source # caps are internal to the fused two-sided result. The reverse # extent's far ``LastShape`` is the FeatureScript start cap; map # that exact builder handle before the registry checks final # membership. No source-plane or geometry-derived relation is # promoted to a CAP role. if exact_two_sided_prism and extent_index == 1 and relation.output_role == "extrude.end": relation = replace(relation, output_role="extrude.start") relations.append(relation) return tool, TopologyDelta( operation="extrude", relations=tuple(relations), ) def _apply_primary_tool( node: FeaturePlanNode, session: "ExecutionSession", tool: Any, *, cutting: bool, topology_delta: TopologyDelta | None = None, topology_anchors: list[TopologyRecord] | None = None, ) -> FeatureResult: """Apply a profile-derived tool while preserving only final-snapshot topology evidence.""" topology_predecessors: list[TopologyRecord] | None = None if cutting: if session.body is None: raise ValueError("cut feature has no body") members, member_cut_deltas = _cut_explicit_body_members(session, tool) if not members: session.clear_body() return session.result(node) tool_delta = topology_delta tool_anchors = list(topology_anchors or ()) retain_transient_tool = _can_register_primary_cut_tool_history( session, tool, tool_delta, topology_anchors, ) if len(session.body_members) == 1: member_id = next(iter(session.body_members)) body = members[member_id] cut_delta = member_cut_deltas[0] if len(member_cut_deltas) == 1 else None # The target-side boolean history is independent of the source # tool's construction history. A trimmed tool cannot support a # source-qualified section query, but its exact BRepAlgoAPI_Cut # Modified/Preserved facts can still prove a unique continuation # of the active target face or edge. topology_anchors = None if cut_delta is not None: if retain_transient_tool: try: topology_predecessors = session.register_transient_prism_tool( node.feature_id, tool, topology_delta=tool_delta, topology_anchors=tool_anchors, ) except ValueError: # Retain cut history as partial diagnostic evidence. The # absent transient source records prevent it from proving # a source-qualified section edge. topology_predecessors = None topology_delta = cut_delta else: topology_delta = None members = {member_id: body} else: if retain_transient_tool: try: topology_predecessors = session.register_transient_prism_tool( node.feature_id, tool, topology_delta=tool_delta, topology_anchors=tool_anchors, ) except ValueError: # A transient tool is optional evidence. Keep the exact # per-member target history when its tool snapshot cannot # be registered completely. topology_predecessors = None body = session.adapter.combine(None, next(iter(members.values()))) for member in list(members.values())[1:]: body = session.adapter.combine(body, member) topology_delta = _compose_member_cut_deltas(member_cut_deltas) topology_anchors = None elif node.params.get("result_mode") == "new_body": body = session.adapter.combine(session.body, tool) members = {**session.body_members, node.feature_id: tool} else: # An ADD can replace both the prior active solid and its direct prism # tool. Retain a selector relation only when both snapshots are # singular and OCC supplies the exact union history. The direct-prism # snapshot remains transient: its source anchors cannot be selected # until the union proves a complete successor in the active result. can_trace_add = ( session.body is not None and topology_delta is not None and topology_delta.operation == "extrude" and topology_delta.history_status == "proven" and bool(topology_delta.relations) and bool(topology_anchors) and len(session.adapter.body_solids(session.body)) == 1 and len(session.adapter.body_solids(tool)) == 1 ) if can_trace_add: fused, fuse_delta = session.adapter.fuse_with_topology_delta(session.body, tool) if fuse_delta is not None: topology_predecessors = session.register_transient_prism_tool( node.feature_id, tool, topology_delta=topology_delta, topology_anchors=list(topology_anchors or ()), ) body = fused topology_delta = fuse_delta topology_anchors = None else: body = fused topology_delta = None topology_anchors = None else: body = session.adapter.fuse(session.body, tool) members = {node.feature_id: body} session.register_body( node.feature_id, body, replay_node=node, body_members=members, topology_delta=topology_delta, topology_predecessors=topology_predecessors, topology_anchors=topology_anchors, ) return session.result(node) 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），没有闭合区域就无法生成实体。 support_face = session.sketch_attachment_faces.get(str(node.sketch_id)) external_anchor_edges = session.sketch_imprint_external_edges.get(str(node.sketch_id)) faces, source_anchor_specs = session.adapter.faces_for_sketch_with_source_anchors( selected_sketch, support_face=support_face, external_anchor_edges=external_anchor_edges, ) 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])] topology_delta: TopologyDelta | None = None topology_anchors: list[TopologyRecord] = [] profile = selected_sketch.get("profile") or {} contours = profile.get("contours") if profile.get("type") == "analytic_contours" else None direct_all_circle_profile = isinstance(contours, list) and bool(contours) and all( isinstance(contour, dict) and bool(contour.get("closed")) and len(contour.get("segments") or []) == 1 and (contour.get("segments") or [{}])[0].get("type") == "circle" for contour in contours ) planar_imprint_profile = profile.get("type") == "planar_imprint" # 3. 按特征类型生成子实体： if node.atomic_id.startswith("extrude_"): # 拉伸：先按终止条件（盲孔/贯穿/至面/双侧等）求出位移向量， # 再对每个面沿每个向量做拉伸，得到实体列表。up_to_surface 在 # profile 与目标面非均匀相交时（extent.trim_to 非空）改用裁剪 # 拉伸：穿透后与目标面求交，只保留可达部分（issue #5）。 tool, topology_delta = _extruded_tool( node, faces, _normal_from_sketch(selected_sketch), session, # A multi-region prism gets per-region builder history only when # every profile region has at least one exact source boundary. # This covers direct circles and the adapter's bounded IMPRINT # splitter path, without turning arbitrary multi-face profiles # into a different construction algorithm. record_multiface_prism_history=(direct_all_circle_profile or planar_imprint_profile) and len(source_anchor_specs) >= len(faces), use_operation_wide_prism_history=planar_imprint_profile and len(source_anchor_specs) >= len(faces), ) if topology_delta is not None: for index, spec in enumerate(source_anchor_specs): kind = spec.get("kind") value = spec.get("value") if kind not in {"edge", "vertex"} or value is None: continue source_entity = spec.get("source_entity") source_entities = tuple(spec.get("source_entities") or ()) if source_entity is None and not source_entities: continue topology_anchors.append(TopologyRecord( record_id=f"anchor:{node.feature_id}:{kind}:{index}", kind=kind, feature_id=node.feature_id, geometry={}, value=value, source_entity=source_entity if isinstance(source_entity, tuple) else None, source_entities=source_entities, )) 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 tool = None can_record_revolve_history = ( node.atomic_id == "revolve_add" and node.params.get("result_mode") == "new_body" and len(faces) == 1 ) for face in faces: if can_record_revolve_history: try: solid, topology_delta = session.adapter.revolve_with_topology_delta(face, angle, axis) except ValueError: # Retain the established executable revolve when OCC # cannot expose a complete builder-history witness. solid = session.adapter.revolve(face, angle, axis) topology_delta = None else: solid = session.adapter.revolve(face, angle, axis) tool = solid if tool is None else session.adapter.fuse(tool, solid) if tool is None: raise ValueError("revolve produced no solid") if topology_delta is not None: for index, spec in enumerate(source_anchor_specs): kind = spec.get("kind") value = spec.get("value") if kind not in {"edge", "vertex"} or value is None: continue source_entity = spec.get("source_entity") source_entities = tuple(spec.get("source_entities") or ()) if source_entity is None and not source_entities: continue topology_anchors.append(TopologyRecord( record_id=f"anchor:{node.feature_id}:{kind}:{index}", kind=kind, feature_id=node.feature_id, geometry={}, value=value, source_entity=source_entity if isinstance(source_entity, tuple) else None, source_entities=source_entities, )) return _apply_primary_tool( node, session, tool, cutting="cut" in node.atomic_id, topology_delta=topology_delta, topology_anchors=topology_anchors, ) 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 _pattern_instance_sources( node: FeaturePlanNode, session: "ExecutionSession", parameter: str = "pattern_instance_refs", ) -> list[str]: """Resolve CDSL pattern-instance refs to their internal body-member keys.""" resolved: list[str] = [] for reference in node.params.get(parameter) or (): if not isinstance(reference, dict): raise ValueError("pattern instance reference must be an object") pattern_id = str(reference.get("pattern_feature_id") or "") source_id = str(reference.get("source_feature_id") or "") instance = reference.get("instance_index") if not pattern_id or not source_id or not isinstance(instance, int): raise ValueError("pattern instance reference is incomplete") pattern = session.nodes.get(pattern_id) if pattern is None or pattern.atomic_id not in {"pattern_circular", "pattern_mirror"}: raise ValueError(f"pattern instance owner is unavailable: {pattern_id}") params = pattern.params if source_id not in {str(value) for value in params.get("source_feature_ids") or ()}: raise ValueError("pattern instance source is not selected by its pattern") count = int(params.get("pattern_count") or 0) excluded = {int(value) for value in params.get("excluded_instance_indices") or ()} if ( pattern.atomic_id == "pattern_mirror" and instance != 1 ) or ( pattern.atomic_id == "pattern_circular" and (instance < 1 or instance >= count or instance in excluded) ): raise ValueError("pattern instance is outside the pattern's surviving instances") member_id = pattern_instance_member_id(pattern_id, source_id, instance) if member_id not in session.body_members: raise ValueError(f"pattern instance body is unavailable: {pattern_id}/{source_id}/{instance}") if member_id not in resolved: resolved.append(member_id) return resolved def _transform_copy_sources( node: FeaturePlanNode, session: "ExecutionSession", parameter: str = "transform_copy_refs", ) -> list[str]: """Resolve source-qualified outputs of preceding multi-body COPY transforms.""" resolved: list[str] = [] for reference in node.params.get(parameter) or (): if not isinstance(reference, dict): raise ValueError("transform COPY reference must be an object") transform_id = str(reference.get("transform_feature_id") or "") source_id = str(reference.get("source_feature_id") or "") if not transform_id or not source_id: raise ValueError("transform COPY reference is incomplete") transform = session.nodes.get(transform_id) params = transform.params if transform is not None else {} sources = params.get("source_feature_ids") or [] if ( transform is None or transform.atomic_id != "transform_bodies" or not bool(params.get("make_copy")) or not isinstance(sources, list) or len(sources) < 2 or source_id not in {str(value) for value in sources} ): raise ValueError(f"transform COPY owner/source is unavailable: {transform_id}/{source_id}") member_id = transform_copy_member_id(transform_id, source_id) if member_id not in session.body_members: raise ValueError(f"transform COPY body is unavailable: {transform_id}/{source_id}") if member_id not in resolved: resolved.append(member_id) return resolved def _member_sources( node: FeaturePlanNode, session: "ExecutionSession", parameter: str, *, pattern_instance_parameter: str | None = None, allow_transform_copies: bool = False, transform_copy_parameter: str = "transform_copy_refs", ) -> list[str]: source_ids = [str(value) for value in node.params.get(parameter) or []] if pattern_instance_parameter is not None: source_ids.extend(_pattern_instance_sources(node, session, pattern_instance_parameter)) if allow_transform_copies: source_ids.extend(_transform_copy_sources(node, session, transform_copy_parameter)) if not source_ids: raise ValueError(f"{node.atomic_id} requires explicit {parameter}") missing = [feature_id for feature_id in source_ids if feature_id not in session.body_members] if missing: raise ValueError(f"{node.atomic_id} source bodies are unavailable: " + ", ".join(missing)) return source_ids 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") spatial = path.get("workplane") is None plane = None if spatial else PlaneSpec.from_mapping(path.get("workplane") or {}) segment = path.get("segment") or {} segments = path.get("segments") def local_point(value: Any) -> Vector3: if plane is None: raise ValueError("planar sweep path requires a workplane") if not isinstance(value, list) or len(value) != 2: raise ValueError("sweep path requires two-dimensional points") 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 local_vector(value: Any) -> Vector3: if plane is None: raise ValueError("planar sweep path requires a workplane") 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]))) if segments is not None: if segment: raise ValueError("sweep path cannot mix segment and segments") if not isinstance(segments, list) or len(segments) < 2: raise ValueError("sweep segmented path requires at least two segments") materialized: list[dict[str, Any]] = [] for index, source in enumerate(segments): if not isinstance(source, dict): raise ValueError("sweep path segment must be an object") kind = str(source.get("type") or "") if kind not in {"line", "arc", "bspline"}: raise ValueError(f"unsupported sweep path segment {kind!r}") target: dict[str, Any] = {"type": kind} if spatial: if kind in {"line", "arc"}: target["start_mm"] = source.get("start_mm") target["end_mm"] = source.get("end_mm") if kind == "arc": target["center_mm"] = source.get("center_mm") target["normal"] = source.get("normal") target["radius_mm"] = source.get("radius_mm") target["clockwise"] = bool(source.get("clockwise", False)) if kind == "bspline": target["points_mm"] = source.get("points_mm") if source.get("start_tangent_mm") is not None: target["start_tangent_mm"] = source.get("start_tangent_mm") if source.get("end_tangent_mm") is not None: target["end_tangent_mm"] = source.get("end_tangent_mm") if source.get("parameters") is not None: target["parameters"] = [float(value) for value in source.get("parameters") or []] if source.get("periodic") is not None: target["periodic"] = bool(source.get("periodic")) materialized.append(target) continue if kind in {"line", "arc"}: target["start_mm"] = local_point(source.get("start")) target["end_mm"] = local_point(source.get("end")) if kind == "arc": target["center_mm"] = local_point(source.get("center")) target["normal"] = plane.normal target["clockwise"] = bool(source.get("clockwise", False)) if kind == "bspline": points = source.get("points") if not isinstance(points, list) or len(points) < 2: raise ValueError("sweep B-spline path requires at least two interpolation points") target["points_mm"] = [local_point(value) for value in points] if source.get("start_tangent") is not None: target["start_tangent_mm"] = local_vector(source.get("start_tangent")) if source.get("end_tangent") is not None: target["end_tangent_mm"] = local_vector(source.get("end_tangent")) if source.get("parameters") is not None: target["parameters"] = [float(value) for value in source.get("parameters") or []] if source.get("periodic") is not None: target["periodic"] = bool(source.get("periodic")) materialized.append(target) return session.adapter.sweep_path_segments(materialized) if spatial: raise ValueError("spatial sweep path requires captured segments") 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 == "circle": return session.adapter.sweep_circle_path( local_point(segment.get("center")), plane.x_dir, plane.normal, radius_mm=float(segment["radius_mm"]), ) elif kind == "arc": return session.adapter.sweep_arc_path( local_point(segment.get("start")), local_point(segment.get("end")), local_point(segment.get("center")), plane.normal, radius_mm=float(segment["radius_mm"]), clockwise=bool(segment["clockwise"]), ) elif kind == "bspline": local_points = segment.get("points") or [] else: raise ValueError(f"unsupported sweep path segment {kind!r}") if len(local_points) < 2: raise ValueError("sweep path requires two-dimensional points") return session.adapter.sweep_path( [local_point(value) for value in local_points], start_tangent=local_vector(segment["start_tangent"]) if segment.get("start_tangent") is not None else None, end_tangent=local_vector(segment["end_tangent"]) if segment.get("end_tangent") is not None else None, parameters=[float(value) for value in segment.get("parameters") or []] or None, ) def _register_added_solid( session: "ExecutionSession", node: FeaturePlanNode, solid: Any, *, topology_delta: TopologyDelta | None = None, ) -> None: """Register an additive primitive solid (box/cyl/sphere/thread/gear/rack/bend). When ``node.params['result_mode'] == "new_body"`` the primitive is kept as an independent body member so that downstream ``boolean_bodies`` can reference it without pulling in the accumulated fuse history. The current body is replaced by a Compound that preserves both, matching the ``extrude_add_blind`` ``new_body`` semantics. Any other value (including missing) falls back to the legacy fuse-into-body behavior. """ if node.params.get("result_mode") == "new_body": combined = session.adapter.combine(session.body, solid) members = {**session.body_members, node.feature_id: solid} session.register_body( node.feature_id, combined, replay_node=node, body_members=members, topology_delta=topology_delta, ) return if session.body is None: session.register_body(node.feature_id, solid, replay_node=node, topology_delta=topology_delta) return if topology_delta is None: session.register_body(node.feature_id, session.adapter.fuse(session.body, solid), replay_node=node) return # Preserve primitive output roles only through the exact OCC fuse history. # The transient records provide source handles, not selectable snapshots. role_records = _direct_output_role_records(session, node, solid, topology_delta) fused, fuse_delta = session.adapter.fuse_with_topology_delta(session.body, solid) session.register_body( node.feature_id, fused, replay_node=node, topology_delta=fuse_delta, topology_predecessors=role_records, ) def _direct_output_role_records( session: "ExecutionSession", node: FeaturePlanNode, solid: Any, topology_delta: TopologyDelta, ) -> list[TopologyRecord]: """Attach only builder-proven output roles to a transient primitive snapshot.""" records = session.adapter.topology_records(solid, node.feature_id, f"transient:{node.feature_id}") result: list[TopologyRecord] = [] for record in records: roles = { relation.output_role for relation in topology_delta.relations if relation.output_role is not None and relation.kind == record.kind and any(session.topology._same_topology_value(record.value, value) for value in relation.result_values) } if roles: result.append(TopologyRecord( record_id=record.record_id, kind=record.kind, feature_id=record.feature_id, body_id=record.body_id, geometry=record.geometry, value=record.value, owner_feature_ids=(node.feature_id,), output_roles=tuple(sorted(roles)), )) return result def _host_plane(resolution: SelectorResolution, adapter: Any) -> PlaneSpec: if resolution.record is None: raise ValueError(resolution.diagnostic.message if resolution.diagnostic else "host face was not resolved") return adapter.planar_face_workplane(resolution.record.value) 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 _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: records = item.records or ((item.record,) if item.record is not None else ()) for record in records: if record.kind == "edge": edges.append(record.value) elif record.kind == "face": edges.extend(edge for edge in 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 = [ record for item in resolved for record in (item.records or ((item.record,) if item.record is not None else ())) ] 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: target = members[0] else: 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") target = members[member_index] target_feature_id = node.params.get("target_feature_id") if target_feature_id is not None: if not isinstance(target_feature_id, str) or not target_feature_id: raise ValueError("shell target_feature_id is invalid") declared = session.body_members.get(target_feature_id) if declared is None: raise ValueError("shell target body is no longer an independently selectable member") declared_solids = session.adapter.body_solids(declared) if len(declared_solids) != 1: raise ValueError("shell target body must resolve to exactly one active solid") if not declared_solids[0].is_same(target): raise ValueError("shell target body does not match the resolved face member") return target, [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