From 871070c440a47d69734841117502291fa56c85e1 Mon Sep 17 00:00:00 2001 From: ganjihong Date: Wed, 9 Sep 2026 13:13:17 +0800 Subject: [PATCH] refactor(cdsl_engine): extract session/extents/pattern_transform from runtime Phase 2 of the decoupling refactor (behavior-preserving move): - runtime_base.py: RuntimeExecutionError, FeatureExecutionError, ExtentVector - session.py: GeometryAdapter protocol + ExecutionSession - extents.py: end-condition planning (_extent_vectors family) - pattern_transform.py: translate/mirror/rotate replay parameter algebra - runtime.py: keeps executors + registry + entry points; re-exports all moved names (incl. test-referenced privates) for import stability No behavior change; verified against baseline (zero new failures). --- backend/engine/cdsl_engine/extents.py | 231 ++++ .../engine/cdsl_engine/pattern_transform.py | 509 ++++++++ backend/engine/cdsl_engine/runtime.py | 1040 +---------------- backend/engine/cdsl_engine/runtime_base.py | 46 + backend/engine/cdsl_engine/session.py | 274 +++++ 5 files changed, 1108 insertions(+), 992 deletions(-) create mode 100644 backend/engine/cdsl_engine/extents.py create mode 100644 backend/engine/cdsl_engine/pattern_transform.py create mode 100644 backend/engine/cdsl_engine/runtime_base.py create mode 100644 backend/engine/cdsl_engine/session.py diff --git a/backend/engine/cdsl_engine/extents.py b/backend/engine/cdsl_engine/extents.py new file mode 100644 index 00000000..1df0a516 --- /dev/null +++ b/backend/engine/cdsl_engine/extents.py @@ -0,0 +1,231 @@ +"""Extrusion/termination-condition planning for the session runtime. + +These helpers turn a CDSL feature's end condition (blind, mid-plane, +through-all, up-to-surface, ...) into one or more :class:`ExtentVector` +displacements. They depend on the session only through its adapter and +selector resolution, never on executors. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from .runtime_base import ExtentVector, FeatureExecutionError +from .specs import PlaneSpec, Vector3, vector_dot, vector_scale, vector_subtract, vector_unit +from .topology import FeaturePlanNode + +if TYPE_CHECKING: # pragma: no cover - import for type checkers only + from .session import ExecutionSession + + +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]: + return _extent_vectors_from_normal( + node, faces, vector_unit(_normal_from_sketch(sketch), field_name="sketch normal"), session, + ) + + +def _extent_vectors_from_normal( + node: FeaturePlanNode, + faces: list[Any], + profile_normal: Vector3, + session: "ExecutionSession", +) -> list[ExtentVector]: + """Resolve extents from an explicit profile normal. + + A derived profile can be an actual B-rep face rather than a sketch. Its + outward normal is just as authoritative as a sketch workplane normal, so + both profile sources share the same bounded extent semantics. + """ + params = node.params + normal = vector_unit(profile_normal, field_name="profile 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"} or bool(params.get("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, + ) diff --git a/backend/engine/cdsl_engine/pattern_transform.py b/backend/engine/cdsl_engine/pattern_transform.py new file mode 100644 index 00000000..bb9c832b --- /dev/null +++ b/backend/engine/cdsl_engine/pattern_transform.py @@ -0,0 +1,509 @@ +"""Parametric transforms for pattern replay (translate / mirror / rotate). + +A pattern instance re-executes its source feature with every absolute +coordinate parameter transformed (sketch, host frame, axis, positions, +center, nested mirror-plane references). These helpers own that parameter +algebra; the pattern executors only decide which transform to apply. +""" + +from __future__ import annotations + +import math +from copy import deepcopy +from typing import TYPE_CHECKING, Any, Callable + +from .specs import AxisSpec, PlaneSpec, Vector3, vector_cross, vector_dot, vector_scale, vector_subtract +from .topology import FeaturePlanNode + +if TYPE_CHECKING: # pragma: no cover - import for type checkers only + from .session import ExecutionSession + + +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 _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 _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 _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 diff --git a/backend/engine/cdsl_engine/runtime.py b/backend/engine/cdsl_engine/runtime.py index aa7cb729..5706d560 100644 --- a/backend/engine/cdsl_engine/runtime.py +++ b/backend/engine/cdsl_engine/runtime.py @@ -1,23 +1,63 @@ -"""Session-based CDSL execution with atomic executor registry.""" +"""Session-based CDSL execution with atomic executor registry. + +The runtime was split into focused modules (behavior-preserving move): + +- ``runtime_base``: shared error types and the ``ExtentVector`` value. +- ``session``: ``ExecutionSession`` and the ``GeometryAdapter`` protocol. +- ``extents``: end-condition planning. +- ``pattern_transform``: translate/mirror/rotate parameter algebra for replay. + +This module keeps the executor registry, the per-atomic executors, and the +``analyze_cdsl`` / ``rebuild_cdsl`` entry points. The moved names are +re-exported so every historical ``cdsl_engine.runtime`` import keeps working. +""" 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 .build123d_adapter import Build123dGeometryAdapter # noqa: F401 (historical re-export) from .capabilities import CapabilityAnalyzer, pattern_transform_blocker, sketch_ids_required_by_contract -from .runtime_types import ( - AxisSpec, BendSpec, CapabilityResult, FeaturePlanNode, FeatureResult, HoleSpec, PlaneSpec, - GearSpec, RackSpec, ThreadSpec, TopologyDelta, TopologyDeltaRelation, Vector3, - RuntimeDiagnostic, SelectorResolution, TopologyRecord, TopologyRegistry, +from .extents import ( + _extent_reference, + _extent_vectors, + _extent_vectors_from_normal, + _normal_from_sketch, + _side_extent_vectors, + _targeted_extent_vector, +) +from .pattern_transform import ( + _box_circular_is_exact, + _coordinate_axis_direction, + _mirrored_node, + _mirrored_sketch, + _normal_is_coordinate_axis, + _owner_plane_frame, + _pattern_operation_node, + _reflect_point, + _rotated_node, + _rotated_point, + _rotated_sketch, + _rotated_vector, + _transformed_loft_profiles, + _translated_node, + _translated_sketch, +) +from .runtime_base import ExtentVector, FeatureExecutionError, RuntimeExecutionError +from .session import ExecutionSession, GeometryAdapter +from .sketch_solver import CORE_SHAPE_GENERATORS, resolve_required_sketches +from .specs import ( + AxisSpec, BendSpec, GearSpec, HoleSpec, PlaneSpec, RackSpec, ThreadSpec, Vector3, pattern_instance_member_id, transform_copy_member_id, vector_add, vector_cross, vector_dot, vector_scale, vector_subtract, vector_unit, ) -from .sketch_solver import CORE_SHAPE_GENERATORS, resolve_required_sketches +from .topology import ( + CapabilityResult, FeaturePlanNode, FeatureResult, RuntimeDiagnostic, + SelectorResolution, TopologyDelta, TopologyDeltaRelation, TopologyRecord, TopologyRegistry, +) ALL_ATOMIC_IDS = frozenset({ @@ -34,39 +74,6 @@ ALL_ATOMIC_IDS = frozenset({ }) -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 @@ -74,467 +81,6 @@ class AtomicExecutor(Protocol): 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_topology_delta(self, sketches: list[dict[str, Any]]) -> tuple[Any, TopologyDelta | None]: ... - 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_with_topology_delta(self, section: Any, spine: Any, *, inner_wires: list[Any] | None = None, make_solid: bool = True, is_frenet: bool = False, transition: Any = None) -> tuple[Any, TopologyDelta | None]: ... - def sweep_path(self, points: list[Vector3], *, start_tangent: Vector3 | None = None, end_tangent: Vector3 | None = None, parameters: list[float] | None = None) -> Any: ... - def face_normal(self, face: Any) -> Vector3: ... - def extrude(self, face: Any, direction: Vector3) -> Any: ... - def extrude_with_topology_delta(self, face: Any, direction: Vector3) -> tuple[Any, TopologyDelta]: ... - def extrude_taper_with_topology_delta(self, face: Any, direction: Vector3, taper_deg: float) -> tuple[Any, TopologyDelta | None]: ... - 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 intersect_with_topology_delta(self, left: Any, right: Any) -> tuple[Any, TopologyDelta | None]: ... - def transform(self, body: Any, transform: dict[str, Any]) -> Any: ... - def transform_with_topology_delta(self, body: Any, transform: dict[str, Any]) -> tuple[Any, TopologyDelta]: ... - def fuse(self, body: Any | None, solid: Any) -> Any: ... - def fuse_with_topology_delta(self, body: Any | None, solid: Any) -> tuple[Any, TopologyDelta | None]: ... - def combine(self, body: Any | None, solid: Any) -> Any: ... - def cut(self, body: Any, tool: Any) -> Any: ... - def cut_with_topology_delta(self, body: Any, tool: Any) -> tuple[Any, TopologyDelta | None]: ... - def sphere(self, radius_mm: float, center_mm: Vector3) -> Any: ... - def thread_solid(self, spec: ThreadSpec) -> Any: ... - def bend_solid(self, spec: BendSpec) -> Any: ... - def gear_solid(self, spec: GearSpec) -> Any: ... - def rack_solid(self, spec: RackSpec) -> 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 fillet_with_topology_delta(self, body: Any, radius_mm: float, edges: list[Any]) -> tuple[Any, TopologyDelta | None]: ... - 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 chamfer_with_topology_delta(self, body: Any, distance_mm: float, distance_2_mm: float | None, edges: list[Any], face: Any | None = None) -> tuple[Any, TopologyDelta | None]: ... - 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 shell_with_topology_delta(self, body: Any, faces: list[Any], thickness_mm: float, *, inward: bool = True) -> tuple[Any, TopologyDelta]: ... - 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, - topology_delta: TopologyDelta | None = None, - topology_predecessors: list[TopologyRecord] | 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), - topology_delta=topology_delta, - additional_predecessors=topology_predecessors or (), - ) - 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, topology_delta=topology_delta, - additional_predecessors=topology_predecessors or (), - ) - 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 clear_body(self) -> None: - """Clear the active solid after an explicit deleteBodies result.""" - self.body = None - self.body_id = None - self.body_members = {} - - 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]: - return _extent_vectors_from_normal( - node, faces, vector_unit(_normal_from_sketch(sketch), field_name="sketch normal"), session, - ) - - -def _extent_vectors_from_normal( - node: FeaturePlanNode, - faces: list[Any], - profile_normal: Vector3, - session: ExecutionSession, -) -> list[ExtentVector]: - """Resolve extents from an explicit profile normal. - - A derived profile can be an actual B-rep face rather than a sketch. Its - outward normal is just as authoritative as a sketch workplane normal, so - both profile sources share the same bounded extent semantics. - """ - params = node.params - normal = vector_unit(profile_normal, field_name="profile 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"} or bool(params.get("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: @@ -1553,143 +1099,6 @@ def _execute_chamfer(node: FeaturePlanNode, session: ExecutionSession) -> Featur 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）执行入口：沿两个方向按数量与间距重放源特征形成阵列。 @@ -1727,171 +1136,6 @@ def _execute_linear_pattern(node: FeaturePlanNode, session: ExecutionSession, ex 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) @@ -1950,183 +1194,6 @@ def _execute_mirror_pattern(node: FeaturePlanNode, session: ExecutionSession) -> 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 _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"}: @@ -2354,17 +1421,6 @@ def _execute_circular_pattern(node: FeaturePlanNode, session: ExecutionSession, 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) diff --git a/backend/engine/cdsl_engine/runtime_base.py b/backend/engine/cdsl_engine/runtime_base.py new file mode 100644 index 00000000..5d2449b0 --- /dev/null +++ b/backend/engine/cdsl_engine/runtime_base.py @@ -0,0 +1,46 @@ +"""Shared runtime error types and extent planning values. + +These primitives sit below both ``session`` (execution state) and the +executor modules, so geometry-independent helpers can raise the same stable +execution errors without importing the session or any executor. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .topology import RuntimeDiagnostic, Vector3 + + +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 diff --git a/backend/engine/cdsl_engine/session.py b/backend/engine/cdsl_engine/session.py new file mode 100644 index 00000000..9dc99bd4 --- /dev/null +++ b/backend/engine/cdsl_engine/session.py @@ -0,0 +1,274 @@ +"""Execution session state and the geometry adapter boundary. + +``ExecutionSession`` owns the active body, body-member graph, replay +definitions, and selector-resolution evidence. ``GeometryAdapter`` is the +kernel-facing protocol the session consumes; geometry values stay opaque so a +different B-rep backend can replace build123d without touching the runtime. +""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Any, Protocol + +from .build123d_adapter import Build123dGeometryAdapter +from .runtime_base import FeatureExecutionError +from .specs import AxisSpec, BendSpec, GearSpec, HoleSpec, PlaneSpec, RackSpec, ThreadSpec, Vector3 +from .topology import ( + FeaturePlanNode, + FeatureResult, + RuntimeDiagnostic, + SelectorResolution, + TopologyDelta, + TopologyRecord, + TopologyRegistry, +) + + +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_topology_delta(self, sketches: list[dict[str, Any]]) -> tuple[Any, TopologyDelta | None]: ... + 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_with_topology_delta(self, section: Any, spine: Any, *, inner_wires: list[Any] | None = None, make_solid: bool = True, is_frenet: bool = False, transition: Any = None) -> tuple[Any, TopologyDelta | None]: ... + def sweep_path(self, points: list[Vector3], *, start_tangent: Vector3 | None = None, end_tangent: Vector3 | None = None, parameters: list[float] | None = None) -> Any: ... + def face_normal(self, face: Any) -> Vector3: ... + def extrude(self, face: Any, direction: Vector3) -> Any: ... + def extrude_with_topology_delta(self, face: Any, direction: Vector3) -> tuple[Any, TopologyDelta]: ... + def extrude_taper_with_topology_delta(self, face: Any, direction: Vector3, taper_deg: float) -> tuple[Any, TopologyDelta | None]: ... + 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 intersect_with_topology_delta(self, left: Any, right: Any) -> tuple[Any, TopologyDelta | None]: ... + def transform(self, body: Any, transform: dict[str, Any]) -> Any: ... + def transform_with_topology_delta(self, body: Any, transform: dict[str, Any]) -> tuple[Any, TopologyDelta]: ... + def fuse(self, body: Any | None, solid: Any) -> Any: ... + def fuse_with_topology_delta(self, body: Any | None, solid: Any) -> tuple[Any, TopologyDelta | None]: ... + def combine(self, body: Any | None, solid: Any) -> Any: ... + def cut(self, body: Any, tool: Any) -> Any: ... + def cut_with_topology_delta(self, body: Any, tool: Any) -> tuple[Any, TopologyDelta | None]: ... + def sphere(self, radius_mm: float, center_mm: Vector3) -> Any: ... + def thread_solid(self, spec: ThreadSpec) -> Any: ... + def bend_solid(self, spec: BendSpec) -> Any: ... + def gear_solid(self, spec: GearSpec) -> Any: ... + def rack_solid(self, spec: RackSpec) -> 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 fillet_with_topology_delta(self, body: Any, radius_mm: float, edges: list[Any]) -> tuple[Any, TopologyDelta | None]: ... + 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 chamfer_with_topology_delta(self, body: Any, distance_mm: float, distance_2_mm: float | None, edges: list[Any], face: Any | None = None) -> tuple[Any, TopologyDelta | None]: ... + 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 shell_with_topology_delta(self, body: Any, faces: list[Any], thickness_mm: float, *, inward: bool = True) -> tuple[Any, TopologyDelta]: ... + 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, + topology_delta: TopologyDelta | None = None, + topology_predecessors: list[TopologyRecord] | 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), + topology_delta=topology_delta, + additional_predecessors=topology_predecessors or (), + ) + 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, topology_delta=topology_delta, + additional_predecessors=topology_predecessors or (), + ) + 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 clear_body(self) -> None: + """Clear the active solid after an explicit deleteBodies result.""" + self.body = None + self.body_id = None + self.body_members = {} + + 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