"""Pattern executors (pattern_linear / pattern_mirror / pattern_circular). Instances replay their source features with transformed parameters rather than copying the current body. NEW-body sources can additionally be instanced as rigid body-graph copies, keeping each instance independently addressable for later COPY/DELETE queries. """ from __future__ import annotations import math from copy import deepcopy from typing import TYPE_CHECKING, Any, Callable from ..capabilities import pattern_transform_blocker from ..pattern_transform import ( _box_circular_is_exact, _mirrored_node, _mirrored_sketch, _normal_is_coordinate_axis, _pattern_operation_node, _rotated_node, _rotated_sketch, _translated_node, _translated_sketch, ) from ..registry import atomic_executor, execute_node from ..specs import AxisSpec, PlaneSpec, Vector3, pattern_instance_member_id, vector_add, vector_dot, vector_scale, vector_subtract, vector_unit from ..topology import FeaturePlanNode, FeatureResult, TopologyDelta, TopologyDeltaRelation, TopologyRecord, TopologyRegistry if TYPE_CHECKING: # pragma: no cover - import for type checkers only from ..session import ExecutionSession ExecutorFunction = Callable[[FeaturePlanNode, "ExecutionSession", dict[str, Any] | None], FeatureResult] def _execute_linear_pattern( node: FeaturePlanNode, session: "ExecutionSession", execute: ExecutorFunction, ) -> 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) @atomic_executor("pattern_linear") 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 _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") source_ids = [str(value) for value in node.params.get("source_feature_ids") or ()] if ( source_ids and all(source_id in session.body_members for source_id in source_ids) and all( (source := session.nodes.get(source_id)) is not None and source.params.get("result_mode") == "new_body" for source_id in source_ids ) ): # Only a direct NEW body has a standalone source identity after a # mirror. A hole, dress-up, or ordinary additive source is merely an # aggregate successor and must use the feature-replay path below. # Keeping this condition identical to capability preflight prevents a # downstream COPY body query from selecting an arbitrary aggregate. members = dict(session.body_members) body = session.body for source_id in source_ids: mirrored = session.adapter.mirror(session.body_members[source_id], resolution.record.value) members[pattern_instance_member_id(node.feature_id, source_id, 1)] = mirrored body = session.adapter.fuse(body, mirrored) if body is None: raise ValueError("mirror pattern produced no body") session.register_body(node.feature_id, body, replay_node=node, body_members=members) return session.result(node) 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) @atomic_executor("pattern_mirror") def _mirror_pattern_executor(node: FeaturePlanNode, session: "ExecutionSession", sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_mirror_pattern(node, session) def _circular_source_is_axisymmetric(node: FeaturePlanNode, session: "ExecutionSession", axis: AxisSpec) -> bool: """Whether rotating a direct circular extrusion creates no new geometry.""" if node.atomic_id not in {"extrude_add_blind", "extrude_add_two_sided"}: return False sketch = session.sketches.get(str(node.sketch_id)) if sketch is None: return False profile = sketch.get("profile") or {} circle = profile if profile.get("type") == "circle" else None if circle is None: contours = profile.get("contours") or [] segments = (contours[0] or {}).get("segments") if len(contours) == 1 else [] circle = segments[0] if isinstance(segments, list) and len(segments) == 1 and segments[0].get("type") == "circle" else None center = (circle or {}).get("center") if not isinstance(center, list) or len(center) != 2: return False try: plane = PlaneSpec.from_mapping(sketch.get("workplane") or {}) except (TypeError, ValueError): return False if abs(vector_dot(plane.normal, axis.direction)) < 1 - 1e-7: return False world_center = vector_add( plane.origin_mm, vector_add(vector_scale(plane.x_dir, float(center[0])), vector_scale(plane.y_dir, float(center[1]))), ) offset = vector_subtract(world_center, axis.origin_mm) radial = vector_subtract(offset, vector_scale(axis.direction, vector_dot(offset, axis.direction))) return math.sqrt(vector_dot(radial, radial)) <= 1e-6 def _advance_copy_topology_records( records: list[TopologyRecord], topology_delta: TopologyDelta | None, ) -> list[TopologyRecord]: """Carry COPY provenance through one exact adapter-history operation. Pattern copies are separate CDSL results even when their solids fuse into a single final body. The temporary records here are never selector candidates themselves. They only retain instance ownership while opaque OCC history proves a unique subshape continuation to the final snapshot. """ if topology_delta is None: return [] advanced: list[TopologyRecord] = [] for record in records: values: list[Any] = [] for relation in topology_delta.relations: if ( relation.kind != record.kind or relation.event not in {"preserved", "modified"} or not TopologyRegistry._same_topology_value(record.value, relation.source_value) ): continue for value in relation.result_values: if not any(TopologyRegistry._same_topology_value(value, known) for known in values): values.append(value) # A split/merge has no unique COPY owner in the present selector # contract. Keep the executable model, but do not make a claim that a # later COPY selector can bind one arbitrary descendant. if len(values) != 1: continue advanced.append(TopologyRecord( record_id=record.record_id, kind=record.kind, feature_id=record.feature_id, body_id=record.body_id, geometry=dict(record.geometry), value=values[0], owner_feature_ids=record.owners, output_roles=record.output_roles, output_role_sources=record.output_role_sources, )) return advanced def _copy_snapshot_topology_delta(records: list[TopologyRecord]) -> TopologyDelta | None: """Bridge traced final COPY handles into the one registered body snapshot.""" if not records: return None return TopologyDelta( operation="pattern_circular_copy_snapshot", relations=tuple( # ``record.value`` has already passed through every transform/fuse # builder in this pattern and is an actual final-B-rep handle. The # identity relation merely connects that evidence to the fresh # adapter snapshot; it is not a geometric rebinding shortcut. TopologyDeltaRelation("preserved", record.kind, record.value, (record.value,)) for record in records ), ) def _has_usable_pattern_body(session: "ExecutionSession", body: Any | None) -> bool: """Reject a formally valid but empty OCC boolean result before publishing it.""" if body is None or not session.adapter.body_solids(body): return False try: return abs(float(body.volume)) > 1e-12 except (AttributeError, TypeError, ValueError): return False def _execute_circular_pattern( node: FeaturePlanNode, session: "ExecutionSession", execute: ExecutorFunction, ) -> 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") source_ids = [source.feature_id for source in sources] pre_pattern_members = dict(session.body_members) if operation_mode == "add" and all(source_id in session.body_members for source_id in source_ids): # A pattern over explicit NEW/kept body members has a stronger contract # than replay: each copy is an independently addressable rigid image of # the named source member. Keep the instance keys in the body graph so # a later CADFS COPY(BODY) transform/delete can name exactly one copy. members = dict(session.body_members) body = session.body traced_copy_records: list[TopologyRecord] = [] for instance in range(1, count): if instance in excluded: continue angle_deg = sweep_angle_deg * instance / count transform = { "type": "rotation", "axis": {"origin_mm": list(axis.origin_mm), "direction": list(axis.direction)}, "angle_deg": angle_deg, } for source_id in source_ids: member_id = pattern_instance_member_id(node.feature_id, source_id, instance) owner_id = f"{node.feature_id}.c{instance}.{source_id}" source_body = session.body_members[source_id] copy, transform_delta = session.adapter.transform_with_topology_delta(source_body, transform) source_records = session.adapter.topology_records( source_body, owner_id, f"body:{node.feature_id}:copy:{instance}:{source_id}:source", ) copy_records = _advance_copy_topology_records(source_records, transform_delta) members[member_id] = copy body, fuse_delta = session.adapter.fuse_with_topology_delta(body, copy) traced_copy_records = _advance_copy_topology_records( [*traced_copy_records, *copy_records], fuse_delta, ) if _has_usable_pattern_body(session, body): session.register_body( node.feature_id, body, replay_node=node, body_members=members, topology_delta=_copy_snapshot_topology_delta(traced_copy_records), topology_predecessors=traced_copy_records, ) return session.result(node) # An OCC boolean may report IsDone/valid for an empty result when a # copied fused body contains coincident internal topology. The normal # pattern contract can replay the source feature contribution instead; # it is the only sound fallback because it keeps source operation, # sketch frame, and body lifecycle semantics intact. 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: # 与阵列轴同心、法向平行的圆形实体拉伸在任意环形实例中均与 # 原实体完全重合。重复执行它会把同一 B-rep 再次交给 OCC fuse， # 后续非轴对称 source 可能因此丢失已生成的实体分支。 if _circular_source_is_axisymmetric(source, session, axis): continue 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") # Replaying a fused sole-body source may be more robust than copying its # full aggregate B-rep (for example, when a rotationally invariant base # would otherwise be unioned with itself). If that replay still has one # physical body, the direct source remains a proven alias of the current # member. Preserve it for a following parts-scoped operation such as # shell; do not extend this alias across multi-body patterns or multiple # source members. members = {node.feature_id: session.body} if ( len(source_ids) == 1 and len(pre_pattern_members) == 1 and source_ids[0] in pre_pattern_members and _has_usable_pattern_body(session, session.body) and len(session.adapter.body_solids(session.body)) == 1 ): members[source_ids[0]] = session.body session.register_body(node.feature_id, session.body, replay_node=node, body_members=members) return session.result(node) @atomic_executor("pattern_circular") def _circular_pattern_executor(node: FeaturePlanNode, session: "ExecutionSession", sketch: dict[str, Any] | None) -> FeatureResult: del sketch return _execute_circular_pattern(node, session, execute_node)