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).
This commit is contained in:
@@ -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,
|
||||
)
|
||||
@@ -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
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user