122 lines
5.9 KiB
Python
122 lines
5.9 KiB
Python
"""Loft and sweep executors (loft_add / loft_add_with_cap_face / sweep_add)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from ..registry import atomic_executor
|
|
from ..topology import FeaturePlanNode, FeatureResult, TopologyRecord
|
|
from .common import _apply_primary_tool, _sweep_path
|
|
|
|
if TYPE_CHECKING: # pragma: no cover - import for type checkers only
|
|
from ..session import ExecutionSession
|
|
|
|
|
|
def _execute_loft_add(node: FeaturePlanNode, session: "ExecutionSession") -> FeatureResult:
|
|
# 放样截面不占用 feature.sketch_id；按有序 profile_sketch_ids 取已解析
|
|
# 草图，并由 adapter 统一校验单闭环、无内环等内核输入约束。
|
|
profile_ids = node.params.get("profile_sketch_ids") or []
|
|
profiles: list[dict[str, Any]] = []
|
|
for sketch_id in profile_ids:
|
|
sketch = session.sketches.get(str(sketch_id))
|
|
if sketch is None:
|
|
raise ValueError(f"loft profile sketch {sketch_id!r} is not resolved")
|
|
profiles.append(sketch)
|
|
solid, topology_delta = session.adapter.loft_with_topology_delta(profiles)
|
|
body = session.adapter.fuse(session.body, solid)
|
|
# Fusing a loft into an existing body replaces its subshapes through a
|
|
# different builder. Only an initial direct loft can expose this builder's
|
|
# cap evidence for the final B-rep snapshot.
|
|
if session.body is not None:
|
|
topology_delta = None
|
|
session.register_body(node.feature_id, body, replay_node=node, topology_delta=topology_delta)
|
|
return session.result(node)
|
|
|
|
|
|
@atomic_executor("loft_add")
|
|
def _loft_executor(node: FeaturePlanNode, session: "ExecutionSession", sketch: dict[str, Any] | None) -> FeatureResult:
|
|
del sketch
|
|
return _execute_loft_add(node, session)
|
|
|
|
|
|
def _execute_loft_add_with_cap_face(node: FeaturePlanNode, session: "ExecutionSession") -> FeatureResult:
|
|
resolved = [session.resolve(selector) for selector in node.selectors]
|
|
failed = next((item for item in resolved if item.status != "resolved"), None)
|
|
if failed or len(resolved) != 1 or resolved[0].record is None or resolved[0].record.kind != "face":
|
|
raise ValueError(failed.diagnostic.message if failed and failed.diagnostic else "cap-face loft selector is unresolved")
|
|
profile_ids = node.params.get("profile_sketch_ids") or []
|
|
profiles: list[dict[str, Any]] = []
|
|
for sketch_id in profile_ids:
|
|
sketch = session.sketches.get(str(sketch_id))
|
|
if sketch is None:
|
|
raise ValueError(f"loft profile sketch {sketch_id!r} is not resolved")
|
|
profiles.append(sketch)
|
|
solid = session.adapter.loft_with_cap_face(resolved[0].record.value, profiles)
|
|
session.register_body(node.feature_id, session.adapter.fuse(session.body, solid), replay_node=node)
|
|
return session.result(node)
|
|
|
|
|
|
@atomic_executor("loft_add_with_cap_face")
|
|
def _loft_cap_face_executor(node: FeaturePlanNode, session: "ExecutionSession", sketch: dict[str, Any] | None) -> FeatureResult:
|
|
del sketch
|
|
return _execute_loft_add_with_cap_face(node, session)
|
|
|
|
|
|
def _execute_sweep_add(node: FeaturePlanNode, session: "ExecutionSession", sketch: dict[str, Any] | None = None) -> FeatureResult:
|
|
profile = sketch or session.sketches.get(str(node.sketch_id))
|
|
if profile is None:
|
|
raise ValueError("sweep has no resolved profile sketch")
|
|
faces, source_anchor_specs = session.adapter.faces_for_sketch_with_source_anchors(profile)
|
|
if len(faces) != 1:
|
|
raise ValueError("sweep requires exactly one closed profile region")
|
|
solid, topology_delta = session.adapter.sweep_with_topology_delta(
|
|
faces[0], _sweep_path(node, session),
|
|
is_frenet=bool(node.params.get("is_frenet", False)),
|
|
)
|
|
if node.atomic_id == "sweep_cut":
|
|
# The PipeShell is a transient cutting tool. Its own builder history
|
|
# cannot become selector provenance after the BRepAlgoAPI_Cut; only
|
|
# the cut's exact target-side delta may survive registration.
|
|
return _apply_primary_tool(node, session, solid, cutting=True)
|
|
is_new_body = node.params.get("result_mode") == "new_body"
|
|
body = session.adapter.combine(session.body, solid) if is_new_body else session.adapter.fuse(session.body, solid)
|
|
# A union rebuilds topology, so the pipe-shell builder cannot prove the
|
|
# final aggregate's relations. The independent-body path retains its exact
|
|
# subshape identity and may expose evidence for the new member.
|
|
if session.body is not None and not is_new_body:
|
|
topology_delta = None
|
|
topology_anchors: list[TopologyRecord] = []
|
|
if topology_delta is not None:
|
|
for index, spec in enumerate(source_anchor_specs):
|
|
kind = spec.get("kind")
|
|
if kind not in {"edge", "vertex"} or spec.get("value") is None:
|
|
continue
|
|
source_entity = spec.get("source_entity")
|
|
source_entities = tuple(spec.get("source_entities") or ())
|
|
if not isinstance(source_entity, tuple) and not source_entities:
|
|
continue
|
|
topology_anchors.append(TopologyRecord(
|
|
record_id=f"anchor:{node.feature_id}:{kind}:{index}",
|
|
kind=kind,
|
|
feature_id=node.feature_id,
|
|
geometry={},
|
|
value=spec["value"],
|
|
source_entity=source_entity if isinstance(source_entity, tuple) else None,
|
|
source_entities=source_entities,
|
|
))
|
|
session.register_body(
|
|
node.feature_id, body, replay_node=node, topology_delta=topology_delta,
|
|
topology_anchors=topology_anchors,
|
|
)
|
|
return session.result(node)
|
|
|
|
|
|
@atomic_executor("sweep_add")
|
|
def _sweep_executor(node: FeaturePlanNode, session: "ExecutionSession", sketch: dict[str, Any] | None) -> FeatureResult:
|
|
return _execute_sweep_add(node, session, sketch)
|
|
|
|
|
|
@atomic_executor("sweep_cut")
|
|
def _sweep_cut_executor(node: FeaturePlanNode, session: "ExecutionSession", sketch: dict[str, Any] | None) -> FeatureResult:
|
|
return _execute_sweep_add(node, session, sketch)
|