280 lines
16 KiB
Python
280 lines
16 KiB
Python
"""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 cylinder_with_topology_delta(self, radius_mm: float, height_mm: float, axis: AxisSpec | None = None) -> tuple[Any, TopologyDelta]: ...
|
||
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 and not item.records)]
|
||
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 [
|
||
record
|
||
for item in resolved
|
||
for record in (item.records or ((item.record,) if item.record is not None else ()))
|
||
]
|
||
|
||
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
|