ganjihong
3fb08423da
Phase 1 of the decoupling refactor (behavior-preserving move): - specs.py: vector math, plane/axis helpers, parametric feature specs - topology.py: diagnostics, planning contracts, TopologyRegistry - runtime_types.py: compatibility shim re-exporting all public names No behavior change; all historical import paths keep working.
1011 lines
48 KiB
Python
1011 lines
48 KiB
Python
"""Runtime-neutral CDSL diagnostics, planning, and topology contracts.
|
||
|
||
This module deliberately has no build123d dependency. The planner and
|
||
selector resolver can therefore be used by validation, batch reporting, and
|
||
any geometry adapter without importing OCC objects.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from math import sqrt
|
||
from typing import Any, Iterable
|
||
|
||
from .specs import AxisSpec, PlaneSpec, Vector3, _length, _vector3, normalize_selector_geometry
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RuntimeDiagnostic:
|
||
code: str
|
||
message: str
|
||
feature_id: str | None = None
|
||
detail: dict[str, Any] = field(default_factory=dict)
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
output: dict[str, Any] = {"code": self.code, "message": self.message}
|
||
if self.feature_id is not None:
|
||
output["feature_id"] = self.feature_id
|
||
if self.detail:
|
||
output["detail"] = self.detail
|
||
return output
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CapabilityResult:
|
||
feature_id: str
|
||
atomic_id: str
|
||
resolved_status: str
|
||
required_capabilities: tuple[str, ...] = ()
|
||
blockers: tuple[RuntimeDiagnostic, ...] = ()
|
||
|
||
@property
|
||
def executable(self) -> bool:
|
||
return self.resolved_status == "executable"
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"feature_id": self.feature_id,
|
||
"atomic_id": self.atomic_id,
|
||
"resolved_status": self.resolved_status,
|
||
"required_capabilities": list(self.required_capabilities),
|
||
"blockers": [blocker.as_dict() for blocker in self.blockers],
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class FeaturePlanNode:
|
||
feature_id: str
|
||
atomic_id: str
|
||
name: str | None
|
||
depends_on: tuple[str, ...]
|
||
params: dict[str, Any]
|
||
selectors: tuple[dict[str, Any], ...]
|
||
sketch_id: str | None
|
||
declared_status: str | None
|
||
source_feature: dict[str, Any]
|
||
|
||
|
||
@dataclass
|
||
class FeatureResult:
|
||
feature_id: str
|
||
atomic_id: str
|
||
status: str
|
||
body_id: str | None = None
|
||
surface_id: str | None = None
|
||
context: PlaneSpec | AxisSpec | None = None
|
||
replay_definition: dict[str, Any] | None = None
|
||
diagnostics: list[RuntimeDiagnostic] = field(default_factory=list)
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
output: dict[str, Any] = {
|
||
"feature_id": self.feature_id,
|
||
"atomic_id": self.atomic_id,
|
||
"status": self.status,
|
||
"diagnostics": [diagnostic.as_dict() for diagnostic in self.diagnostics],
|
||
}
|
||
if self.body_id is not None:
|
||
output["body_id"] = self.body_id
|
||
if self.surface_id is not None:
|
||
output["surface_id"] = self.surface_id
|
||
if self.context is not None:
|
||
output["context"] = self.context.as_dict()
|
||
if self.replay_definition is not None:
|
||
output["replay_definition"] = self.replay_definition
|
||
return output
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TopologyRecord:
|
||
"""Runtime-side signature of a topology item or context object.
|
||
|
||
``feature_id`` identifies the feature that produced this *snapshot*.
|
||
``owner_feature_ids`` is durable semantic provenance for an unchanged
|
||
current B-rep item. Boolean and dress-up operations replace OCC objects,
|
||
so keeping these concepts separate prevents a later mutation from making
|
||
every surviving face appear to be owned by that mutation.
|
||
"""
|
||
|
||
record_id: str
|
||
kind: str
|
||
feature_id: str
|
||
body_id: str | None = None
|
||
geometry: dict[str, Any] = field(default_factory=dict)
|
||
value: Any = None
|
||
owner_feature_ids: tuple[str, ...] = ()
|
||
# Builder-produced roles describe a particular result subshape. They are
|
||
# deliberately separate from the geometric signature: equal geometry does
|
||
# not prove that two faces have the same feature-output meaning.
|
||
output_roles: tuple[str, ...] = ()
|
||
# Generated roles may carry the direct feature-output role that the kernel
|
||
# operation transformed. This is semantic provenance, not a stable-id
|
||
# shortcut: the resolver still requires the exact active result snapshot.
|
||
output_role_sources: tuple[tuple[str, str, str], ...] = ()
|
||
|
||
@property
|
||
def owners(self) -> tuple[str, ...]:
|
||
"""Return durable provenance, retaining compatibility for contexts."""
|
||
return self.owner_feature_ids or (self.feature_id,)
|
||
|
||
def public_dict(self) -> dict[str, Any]:
|
||
output: dict[str, Any] = {
|
||
"record_id": self.record_id,
|
||
"kind": self.kind,
|
||
"feature_id": self.feature_id,
|
||
"geometry": self.geometry,
|
||
}
|
||
if self.body_id is not None:
|
||
output["body_id"] = self.body_id
|
||
if self.owner_feature_ids:
|
||
output["owner_feature_ids"] = list(self.owner_feature_ids)
|
||
if self.output_roles:
|
||
output["output_roles"] = list(self.output_roles)
|
||
if self.output_role_sources:
|
||
output["output_role_sources"] = [
|
||
{"output_role": role, "owner_feature_id": owner, "source_output_role": source_role}
|
||
for role, owner, source_role in self.output_role_sources
|
||
]
|
||
return output
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TopologyDeltaRelation:
|
||
"""One opaque kernel-history relationship for a topology subshape.
|
||
|
||
Geometry adapters retain ownership of the values in this structure. They
|
||
are intentionally opaque to the runtime: a build123d/OCC adapter may use
|
||
``TopoDS_Shape`` values while another adapter can use its native handles.
|
||
The registry only asks whether a handle is exactly the same topology item;
|
||
it never uses this contract to score nearby geometry.
|
||
"""
|
||
|
||
event: str
|
||
kind: str
|
||
source_value: Any
|
||
result_values: tuple[Any, ...] = ()
|
||
output_role: str | None = None
|
||
|
||
def __post_init__(self) -> None:
|
||
if self.event not in {"preserved", "modified", "generated", "deleted"}:
|
||
raise ValueError(f"unsupported topology delta event {self.event!r}")
|
||
if self.kind not in {"face", "edge", "vertex"}:
|
||
raise ValueError(f"unsupported topology delta kind {self.kind!r}")
|
||
if self.event == "deleted" and (self.result_values or self.output_role is not None):
|
||
raise ValueError("deleted topology delta relations cannot have result values or an output role")
|
||
if self.output_role is not None and (not isinstance(self.output_role, str) or not self.output_role):
|
||
raise ValueError("topology delta output_role must be a non-empty string when provided")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TopologyDelta:
|
||
"""Kernel-backed topology history for one adapter operation.
|
||
|
||
``operation`` is evidence only. The runtime transfers durable provenance
|
||
solely from a unique exact relationship, never from an operation name or a
|
||
geometric resemblance.
|
||
"""
|
||
|
||
operation: str
|
||
relations: tuple[TopologyDeltaRelation, ...] = ()
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SelectorResolution:
|
||
selector: dict[str, Any]
|
||
status: str
|
||
record: TopologyRecord | None = None
|
||
candidates: tuple[dict[str, Any], ...] = ()
|
||
diagnostic: RuntimeDiagnostic | None = None
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
output = {
|
||
"selector": self.selector,
|
||
"status": self.status,
|
||
"candidates": list(self.candidates),
|
||
}
|
||
if self.record is not None:
|
||
output["record"] = self.record.public_dict()
|
||
output["selected"] = self.record.public_dict()
|
||
score = next(
|
||
(candidate.get("score") for candidate in self.candidates if candidate.get("record_id") == self.record.record_id),
|
||
None,
|
||
)
|
||
if score is not None:
|
||
output["score"] = score
|
||
if self.diagnostic is not None:
|
||
output["diagnostic"] = self.diagnostic.as_dict()
|
||
return output
|
||
|
||
|
||
class TopologyRegistry:
|
||
"""Feature-scoped context/topology registry with explainable matching."""
|
||
|
||
def __init__(self) -> None:
|
||
self._records: list[TopologyRecord] = []
|
||
self._by_feature: dict[str, list[TopologyRecord]] = {}
|
||
self._active_body_id: str | None = None
|
||
self._topology_deltas: list[dict[str, Any]] = []
|
||
# #8 selector 持久性：old_record_id -> [new_record_id]。fillet/chamfer
|
||
# 会把一条直线边拆分为若干段（中间直段 + 两端圆弧），旧边不再与任何
|
||
# 新边几何等价；这里记录"位置轨迹延续"的直段后继，使后续 selector 的
|
||
# stable_id 引用可以解析到 active body 内的新形态。
|
||
self._successors: dict[str, list[str]] = {}
|
||
|
||
def register(self, record: TopologyRecord) -> None:
|
||
self._records.append(record)
|
||
self._by_feature.setdefault(record.feature_id, []).append(record)
|
||
|
||
def records_for_feature(self, feature_id: str) -> tuple[TopologyRecord, ...]:
|
||
return tuple(self._by_feature.get(feature_id, ()))
|
||
|
||
def records(self) -> tuple[TopologyRecord, ...]:
|
||
return tuple(self._records)
|
||
|
||
def topology_deltas(self) -> tuple[dict[str, Any], ...]:
|
||
"""Return serializable evidence derived from exact adapter history."""
|
||
return tuple(self._topology_deltas)
|
||
|
||
def register_context(self, feature_id: str, context: PlaneSpec | AxisSpec) -> TopologyRecord:
|
||
kind = "plane" if isinstance(context, PlaneSpec) else "axis"
|
||
record = TopologyRecord(
|
||
record_id=f"{feature_id}:{kind}",
|
||
kind=kind,
|
||
feature_id=feature_id,
|
||
geometry=context.as_dict(),
|
||
value=context,
|
||
)
|
||
self.register(record)
|
||
return record
|
||
|
||
def replace_body_topology(
|
||
self, feature_id: str, body_id: str, records: Iterable[TopologyRecord],
|
||
*, active_body_id: str | None = None, topology_delta: TopologyDelta | None = None,
|
||
additional_predecessors: Iterable[TopologyRecord] = (),
|
||
) -> None:
|
||
self.replace_body_topologies(
|
||
feature_id, [(body_id, records)], active_body_id=active_body_id, topology_delta=topology_delta,
|
||
additional_predecessors=additional_predecessors,
|
||
)
|
||
|
||
def replace_body_topologies(
|
||
self, feature_id: str, bodies: Iterable[tuple[str, Iterable[TopologyRecord]]],
|
||
*, active_body_id: str | None = None, topology_delta: TopologyDelta | None = None,
|
||
additional_predecessors: Iterable[TopologyRecord] = (),
|
||
) -> None:
|
||
"""Record a fresh B-rep snapshot after a feature mutates the body.
|
||
|
||
OCC topology object identity is invalidated by most body mutations.
|
||
We therefore keep old objects out of active selector resolution but
|
||
carry their semantic owners forward when, and only when, one current
|
||
object has one geometrically equivalent predecessor. A changed or
|
||
split object intentionally becomes owned by this feature instead of
|
||
being guessed as belonging to an older one.
|
||
|
||
``active_body_id`` names the whole-body group when ``body_id`` is a
|
||
member of a multi-solid body (issue #7): the group id keeps the next
|
||
mutation's predecessor lookup scoped to every solid of the previous
|
||
body, while each member keeps its own ``body:{feature}:{index}`` id.
|
||
"""
|
||
active_previous = [
|
||
record for record in self._records
|
||
if self._active_body_id is not None and record.body_id is not None
|
||
and (
|
||
record.body_id == self._active_body_id
|
||
or record.body_id.startswith(f"{self._active_body_id}:")
|
||
)
|
||
]
|
||
# A pattern COPY can carry a chain of exact transform/boolean builder
|
||
# histories before its final aggregate snapshot is registered. These
|
||
# temporary records are valid predecessors only for that documented
|
||
# kernel-history bridge. They are deliberately excluded from geometric
|
||
# fallback matching: an equal-looking final face never proves that it
|
||
# belongs to one particular copy instance.
|
||
transient_previous = list(additional_predecessors)
|
||
previous = [*active_previous, *transient_previous]
|
||
# 同一 source feature 的 pattern copy 可以产生完全相同的几何面。它们
|
||
# 必须保留为多个实例，不能在跨 body 的全局 predecessor 匹配中互相消费。
|
||
# pattern 的 Compound 成员顺序是稳定的：已有实例以同一 member index
|
||
# 延续，新增实例只会出现在末尾。按该 index 限定后继匹配。
|
||
current = [(body_id, list(records)) for body_id, records in bodies]
|
||
current_records = [record for _body_id, records in current for record in records]
|
||
(
|
||
exact_predecessors,
|
||
exact_successors,
|
||
kernel_covered_predecessors,
|
||
exact_output_roles,
|
||
exact_output_role_sources,
|
||
delta_evidence,
|
||
) = self._exact_delta_links(
|
||
topology_delta, previous, current_records,
|
||
)
|
||
previous_member_ids = {
|
||
suffix for record in active_previous
|
||
for suffix in [str(record.body_id).rsplit(":", 1)[-1]]
|
||
if suffix.isdigit()
|
||
}
|
||
use_member_indexes = len(current) > 1 and previous_member_ids
|
||
consumed_predecessors: set[str] = set()
|
||
registered: list[TopologyRecord] = []
|
||
for body_id, records in current:
|
||
member_id = str(body_id).rsplit(":", 1)[-1]
|
||
local_predecessors = [
|
||
record for record in active_previous
|
||
if not use_member_indexes or str(record.body_id).rsplit(":", 1)[-1] == member_id
|
||
]
|
||
for record in records:
|
||
exact_predecessor_id = exact_predecessors.get(record.record_id)
|
||
# Member order is a useful isolation boundary for geometric
|
||
# fallback matching, especially for coincident pattern
|
||
# copies. It is not a provenance boundary when a kernel
|
||
# builder explicitly relates one source subshape to one
|
||
# result subshape: boolean/delete lifecycle can remove an
|
||
# earlier member and shift a surviving source to another
|
||
# member index. A unique OCC continuation remains exact
|
||
# evidence across that index change.
|
||
predecessor = next(
|
||
(prior for prior in previous if prior.record_id == exact_predecessor_id),
|
||
None,
|
||
)
|
||
if predecessor is None and record.record_id not in exact_predecessors:
|
||
predecessor = self._unique_equivalent_predecessor(record, local_predecessors, consumed_predecessors)
|
||
owners = predecessor.owners if predecessor is not None else (feature_id,)
|
||
output_roles = set(record.output_roles)
|
||
output_roles.update(exact_output_roles.get(record.record_id, ()))
|
||
output_role_sources = set(record.output_role_sources)
|
||
output_role_sources.update(exact_output_role_sources.get(record.record_id, ()))
|
||
# A feature-output role can survive a later operation only
|
||
# through the same unique kernel continuation used for owner
|
||
# provenance. Geometry equivalence alone never carries it.
|
||
if predecessor is not None and record.record_id in exact_predecessors:
|
||
output_roles.update(predecessor.output_roles)
|
||
if predecessor is not None and record.record_id not in exact_predecessors:
|
||
consumed_predecessors.add(predecessor.record_id)
|
||
registered.append(TopologyRecord(
|
||
record_id=record.record_id,
|
||
kind=record.kind,
|
||
feature_id=feature_id,
|
||
body_id=body_id,
|
||
geometry=dict(record.geometry),
|
||
value=record.value,
|
||
owner_feature_ids=owners,
|
||
output_roles=tuple(sorted(output_roles)),
|
||
output_role_sources=tuple(sorted(output_role_sources)),
|
||
))
|
||
for record in registered:
|
||
self.register(record)
|
||
for predecessor_id, successor_ids in exact_successors.items():
|
||
known = self._successors.setdefault(predecessor_id, [])
|
||
for successor_id in successor_ids:
|
||
if successor_id not in known:
|
||
known.append(successor_id)
|
||
if delta_evidence is not None:
|
||
self._topology_deltas.append({
|
||
"feature_id": feature_id,
|
||
"operation": topology_delta.operation,
|
||
"relations": delta_evidence,
|
||
})
|
||
# #8 selector 持久性：被消费（拆分成段）的旧边记录演化后继，供后续
|
||
# selector 的 stable_id 引用解析到 active body 内的新形态。多条演化
|
||
# 候选时只登记"漂移显著最小"的那条（例如底面边圆角后既有缩短的直段
|
||
# 也有圆角过渡带的新边，前者的端点与原边重合、漂移更小）；漂移并列
|
||
# （如竖直边被完整消费成两条等距直段）属于本质歧义，保守不登记。
|
||
for prior in previous:
|
||
if (
|
||
prior.record_id in consumed_predecessors
|
||
or prior.record_id in exact_successors
|
||
or prior.record_id in kernel_covered_predecessors
|
||
):
|
||
continue
|
||
candidates = sorted(
|
||
(
|
||
(self._evolved_drift(prior, record), record.record_id)
|
||
for record in registered if self._evolved_equivalent(prior, record)
|
||
),
|
||
key=lambda item: item[0],
|
||
)
|
||
if not candidates:
|
||
continue
|
||
best, second = candidates[0], (candidates[1] if len(candidates) > 1 else None)
|
||
if second is None or (second[0] - best[0]) > max(0.5, 0.2 * best[0]):
|
||
self._successors[prior.record_id] = [best[1]]
|
||
self._active_body_id = active_body_id or body_id
|
||
|
||
@staticmethod
|
||
def _same_topology_value(left: Any, right: Any) -> bool:
|
||
"""Compare adapter handles only through their exact topology identity."""
|
||
left_value = getattr(left, "wrapped", left)
|
||
right_value = getattr(right, "wrapped", right)
|
||
if left_value is right_value:
|
||
return True
|
||
for candidate, other in ((left_value, right_value), (right_value, left_value)):
|
||
for method_name in ("IsSame", "is_same"):
|
||
method = getattr(candidate, method_name, None)
|
||
if callable(method):
|
||
try:
|
||
return bool(method(other))
|
||
except (AttributeError, TypeError, ValueError):
|
||
continue
|
||
return False
|
||
|
||
@classmethod
|
||
def _exact_delta_links(
|
||
cls,
|
||
topology_delta: TopologyDelta | None,
|
||
previous: list[TopologyRecord],
|
||
current: list[TopologyRecord],
|
||
) -> tuple[
|
||
dict[str, str],
|
||
dict[str, list[str]],
|
||
set[str],
|
||
dict[str, tuple[str, ...]],
|
||
dict[str, tuple[tuple[str, str, str], ...]],
|
||
list[dict[str, Any]] | None,
|
||
]:
|
||
"""Bind opaque kernel history to snapshots without geometric guessing.
|
||
|
||
Ownership transfer is intentionally limited to a single source item and
|
||
a single output item. Split/merge history remains useful evidence, but
|
||
has no unique owner continuation until a later operation-specific
|
||
contract can express it.
|
||
"""
|
||
if topology_delta is None:
|
||
return {}, {}, set(), {}, {}, None
|
||
candidate_sources: dict[str, set[str]] = {}
|
||
kernel_covered_predecessors: set[str] = set()
|
||
output_roles: dict[str, set[str]] = {}
|
||
output_role_sources: dict[str, set[tuple[str, str, str]]] = {}
|
||
relation_links: list[tuple[str, str] | None] = []
|
||
evidence: list[dict[str, Any]] = []
|
||
for relation in topology_delta.relations:
|
||
sources = [
|
||
record for record in previous
|
||
if record.kind == relation.kind and cls._same_topology_value(record.value, relation.source_value)
|
||
]
|
||
outputs = [
|
||
record for record in current
|
||
if record.kind == relation.kind
|
||
and any(cls._same_topology_value(record.value, value) for value in relation.result_values)
|
||
]
|
||
item = {
|
||
"event": relation.event,
|
||
"kind": relation.kind,
|
||
"source_record_ids": [record.record_id for record in sources],
|
||
"result_record_ids": [record.record_id for record in outputs],
|
||
"proof": "kernel_history",
|
||
}
|
||
if relation.output_role is not None:
|
||
item["output_role"] = relation.output_role
|
||
role_is_unique = len(relation.result_values) == 1 and len(outputs) == 1
|
||
item["output_role_status"] = (
|
||
"unique_result_snapshot" if role_is_unique else "non_unique_or_missing_result_snapshot"
|
||
)
|
||
if role_is_unique:
|
||
output_roles.setdefault(outputs[0].record_id, set()).add(relation.output_role)
|
||
for source in sources:
|
||
for source_role in source.output_roles:
|
||
for source_owner in source.owners:
|
||
output_role_sources.setdefault(outputs[0].record_id, set()).add(
|
||
(relation.output_role, source_owner, source_role)
|
||
)
|
||
if len(sources) == 1:
|
||
kernel_covered_predecessors.add(sources[0].record_id)
|
||
can_transfer = (
|
||
relation.event in {"preserved", "modified"}
|
||
and len(relation.result_values) == 1
|
||
and len(sources) == 1
|
||
and len(outputs) == 1
|
||
)
|
||
if can_transfer:
|
||
source_id, result_id = sources[0].record_id, outputs[0].record_id
|
||
candidate_sources.setdefault(result_id, set()).add(source_id)
|
||
relation_links.append((source_id, result_id))
|
||
else:
|
||
relation_links.append(None)
|
||
evidence.append(item)
|
||
predecessors = {
|
||
result_id: next(iter(source_ids))
|
||
for result_id, source_ids in candidate_sources.items()
|
||
if len(source_ids) == 1
|
||
}
|
||
successors: dict[str, list[str]] = {}
|
||
for result_id, source_id in predecessors.items():
|
||
successors.setdefault(source_id, []).append(result_id)
|
||
for item, relation, link in zip(evidence, topology_delta.relations, relation_links):
|
||
if link is not None:
|
||
_source_id, result_id = link
|
||
item["status"] = (
|
||
"unique_exact_continuation"
|
||
if len(candidate_sources[result_id]) == 1 else "ambiguous_exact_continuation"
|
||
)
|
||
elif relation.event in {"preserved", "modified"}:
|
||
item["status"] = "non_unique_or_incomplete"
|
||
else:
|
||
item["status"] = "recorded_without_owner_transfer"
|
||
return (
|
||
predecessors,
|
||
successors,
|
||
kernel_covered_predecessors,
|
||
{record_id: tuple(sorted(roles)) for record_id, roles in output_roles.items()},
|
||
{record_id: tuple(sorted(sources)) for record_id, sources in output_role_sources.items()},
|
||
evidence,
|
||
)
|
||
|
||
@staticmethod
|
||
def _numbers_equal(left: Any, right: Any, *, tolerance: float = 1e-6) -> bool:
|
||
try:
|
||
return abs(float(left) - float(right)) <= tolerance
|
||
except (TypeError, ValueError):
|
||
return False
|
||
|
||
@classmethod
|
||
def _vectors_equal(cls, left: Any, right: Any, *, tolerance: float = 1e-6) -> bool:
|
||
try:
|
||
first = _vector3(left, field_name="prior topology geometry")
|
||
second = _vector3(right, field_name="current topology geometry")
|
||
except ValueError:
|
||
return False
|
||
return all(abs(a - b) <= tolerance for a, b in zip(first, second))
|
||
|
||
@classmethod
|
||
def _geometry_equivalent(cls, prior: TopologyRecord, current: TopologyRecord) -> bool:
|
||
"""Check a complete, orientation-aware snapshot signature.
|
||
|
||
This is intentionally much stricter than selector scoring. Selector
|
||
scoring may compare partial source evidence; provenance transfer must
|
||
never manufacture ownership from a merely similar candidate.
|
||
"""
|
||
if prior.kind != current.kind:
|
||
return False
|
||
left, right = prior.geometry, current.geometry
|
||
for key in ("surface_type", "curve_type"):
|
||
if left.get(key) != right.get(key):
|
||
return False
|
||
for key in ("bbox_mm", "center_mm", "normal", "plane_normal"):
|
||
if key in left or key in right:
|
||
if key not in left or key not in right:
|
||
return False
|
||
left_value, right_value = left[key], right[key]
|
||
if key == "bbox_mm":
|
||
if not isinstance(left_value, (list, tuple)) or not isinstance(right_value, (list, tuple)):
|
||
return False
|
||
if len(left_value) != 6 or len(right_value) != 6:
|
||
return False
|
||
if not all(cls._numbers_equal(a, b) for a, b in zip(left_value, right_value)):
|
||
return False
|
||
elif not cls._vectors_equal(left_value, right_value):
|
||
return False
|
||
for key in ("area_mm2", "length_mm", "plane_offset_mm"):
|
||
if key in left or key in right:
|
||
if key not in left or key not in right or not cls._numbers_equal(left[key], right[key]):
|
||
return False
|
||
for key in ("adjacency_signature", "adjacent_face_count", "incident_edge_count"):
|
||
if key in left or key in right:
|
||
if key not in left or key not in right or left[key] != right[key]:
|
||
return False
|
||
left_start, left_end = left.get("start_mm"), left.get("end_mm")
|
||
right_start, right_end = right.get("start_mm"), right.get("end_mm")
|
||
if any(value is not None for value in (left_start, left_end, right_start, right_end)):
|
||
if None in (left_start, left_end, right_start, right_end):
|
||
return False
|
||
same_direction = cls._vectors_equal(left_start, right_start) and cls._vectors_equal(left_end, right_end)
|
||
reverse_direction = cls._vectors_equal(left_start, right_end) and cls._vectors_equal(left_end, right_start)
|
||
if not same_direction and not reverse_direction:
|
||
return False
|
||
return True
|
||
|
||
@classmethod
|
||
def _unique_equivalent_predecessor(
|
||
cls,
|
||
current: TopologyRecord,
|
||
predecessors: Iterable[TopologyRecord],
|
||
consumed_predecessors: set[str],
|
||
) -> TopologyRecord | None:
|
||
matches = [
|
||
record for record in predecessors
|
||
if record.record_id not in consumed_predecessors and cls._geometry_equivalent(record, current)
|
||
]
|
||
return matches[0] if len(matches) == 1 else None
|
||
|
||
@staticmethod
|
||
def _evolved_drift(prior: TopologyRecord, current: TopologyRecord) -> float | None:
|
||
"""Endpoint drift between direction-aligned straight edges.
|
||
|
||
Returns the minimum total endpoint drift (mm) when the two edges are
|
||
collinear straight lines (either orientation), otherwise ``None``.
|
||
"""
|
||
if prior.kind != current.kind:
|
||
return None
|
||
left, right = prior.geometry, current.geometry
|
||
if left.get("curve_type") != "line" or right.get("curve_type") != "line":
|
||
return None
|
||
if None in (left.get("start_mm"), left.get("end_mm"), right.get("start_mm"), right.get("end_mm")):
|
||
return None
|
||
|
||
def _delta(a: tuple[float, float, float], b: tuple[float, float, float]) -> tuple[float, float, float]:
|
||
return (b[0] - a[0], b[1] - a[1], b[2] - a[2])
|
||
|
||
def _dist(a: tuple[float, float, float], b: tuple[float, float, float]) -> float:
|
||
return sqrt(sum((a[i] - b[i]) ** 2 for i in range(3)))
|
||
|
||
left_dir = _delta(left["start_mm"], left["end_mm"])
|
||
right_dir = _delta(right["start_mm"], right["end_mm"])
|
||
if _length(left_dir) <= 1e-9 or _length(right_dir) <= 1e-9:
|
||
return None
|
||
cross = (
|
||
left_dir[1] * right_dir[2] - left_dir[2] * right_dir[1],
|
||
left_dir[2] * right_dir[0] - left_dir[0] * right_dir[2],
|
||
left_dir[0] * right_dir[1] - left_dir[1] * right_dir[0],
|
||
)
|
||
if _length(cross) / (_length(left_dir) * _length(right_dir)) > 1e-3:
|
||
return None
|
||
same_order = _dist(left["start_mm"], right["start_mm"]) + _dist(left["end_mm"], right["end_mm"])
|
||
reversed_order = _dist(left["start_mm"], right["end_mm"]) + _dist(left["end_mm"], right["start_mm"])
|
||
return min(same_order, reversed_order)
|
||
|
||
@classmethod
|
||
def _evolved_equivalent(cls, prior: TopologyRecord, current: TopologyRecord, *, drift_mm: float = 5.0) -> bool:
|
||
"""Loose "position trajectory" equivalence used for evolved successors.
|
||
|
||
Unlike ``_geometry_equivalent`` (strict, anti-false-positive provenance),
|
||
this deliberately tolerates small endpoint drift: fillet/chamfer split a
|
||
straight edge into segments (a middle straight run plus end arcs). The
|
||
straight run keeps the same direction and stays within ``drift_mm`` of the
|
||
original edge, so it can serve as the edge's evolved successor. Uniqueness
|
||
is enforced by the caller (only a single best candidate is recorded).
|
||
"""
|
||
drift = cls._evolved_drift(prior, current)
|
||
return drift is not None and drift <= drift_mm
|
||
|
||
@staticmethod
|
||
def _vector_score(expected: Any, actual: Any, tolerance: float = 1e-4) -> float | None:
|
||
try:
|
||
left = _vector3(expected, field_name="selector geometry")
|
||
right = _vector3(actual, field_name="record geometry")
|
||
except ValueError:
|
||
return None
|
||
error = _length(tuple(a - b for a, b in zip(left, right)))
|
||
return max(0.0, 1.0 - error / tolerance)
|
||
|
||
@classmethod
|
||
def _geometry_score(cls, selector_geometry: dict[str, Any], record_geometry: dict[str, Any]) -> float | None:
|
||
if not selector_geometry:
|
||
return 0.0
|
||
scores: list[float] = []
|
||
for key in ("center_mm", "circle_center_mm", "normal", "origin_mm", "direction", "plane_normal", "start_mm", "end_mm"):
|
||
if key in selector_geometry:
|
||
score = cls._vector_score(selector_geometry[key], record_geometry.get(key))
|
||
if score is None:
|
||
return None
|
||
scores.append(score)
|
||
for key in ("surface_type", "curve_type"):
|
||
if key in selector_geometry:
|
||
if record_geometry.get(key) != selector_geometry[key]:
|
||
# #8 selector 持久性：fillet/chamfer 会把直线边演化为圆弧、
|
||
# 平面演化为柱面，但被选中拓扑的位置锚定（bbox/center/端点）
|
||
# 不变。曲线/曲面类型变化不再一票否决，而是记低分：位置完全
|
||
# 重合的候选（同一条边的形态演化）仍可胜出；位置不重合的
|
||
# 相邻边会被 0 分项拉低，仍被 minimum_score 挡住。
|
||
scores.append(0.5)
|
||
else:
|
||
scores.append(1.0)
|
||
if "bbox_mm" in selector_geometry:
|
||
expected = selector_geometry["bbox_mm"]
|
||
actual = record_geometry.get("bbox_mm")
|
||
if not isinstance(expected, list) or not isinstance(actual, list) or len(expected) != len(actual):
|
||
return None
|
||
delta = max(abs(float(a) - float(b)) for a, b in zip(expected, actual))
|
||
scores.append(max(0.0, 1.0 - delta / 1e-4))
|
||
if "plane_offset_mm" in selector_geometry:
|
||
try:
|
||
delta = abs(float(selector_geometry["plane_offset_mm"]) - float(record_geometry.get("plane_offset_mm")))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
scores.append(max(0.0, 1.0 - delta / 1e-4))
|
||
if "radius_mm" in selector_geometry:
|
||
try:
|
||
delta = abs(float(selector_geometry["radius_mm"]) - float(record_geometry.get("radius_mm")))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
scores.append(max(0.0, 1.0 - delta / 1e-4))
|
||
if "area_mm2" in selector_geometry:
|
||
try:
|
||
expected_area = float(selector_geometry["area_mm2"])
|
||
actual_area = float(record_geometry.get("area_mm2"))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
relative_delta = abs(expected_area - actual_area) / max(abs(expected_area), 1e-9)
|
||
scores.append(max(0.0, 1.0 - relative_delta / 1e-4))
|
||
return sum(scores) / len(scores) if scores else 0.0
|
||
|
||
def resolve(
|
||
self,
|
||
selector: dict[str, Any],
|
||
*,
|
||
minimum_score: float = 0.8,
|
||
active_body_id: str | None = None,
|
||
) -> SelectorResolution:
|
||
kind = selector.get("kind")
|
||
owner = selector.get("owner_feature_id")
|
||
candidates = [record for record in self._records if record.kind == kind]
|
||
if active_body_id and kind in {"face", "edge", "vertex", "body"}:
|
||
# #7 multi-body：记录 body_id 可能是 body:{feature}:{index}（多体
|
||
# 成员），用前缀匹配把整个主体的记录纳入候选，同时保证旧 body 的
|
||
# 记录（不同 feature 前缀）不会泄漏进来。
|
||
candidates = [
|
||
record for record in candidates
|
||
if record.body_id == active_body_id
|
||
or (record.body_id is not None and record.body_id.startswith(f"{active_body_id}:"))
|
||
]
|
||
if owner:
|
||
candidates = [record for record in candidates if owner in record.owners]
|
||
if kind == "plane" and selector.get("frame") is not None:
|
||
# #6 pattern 引用重解析：pattern 重放 source（pattern_mirror）时，
|
||
# mirror_plane 的 plane 引用由运行时随实例变换后内联为显式 frame
|
||
# （_translated_node / _mirrored_node），这里直接构造 PlaneSpec，
|
||
# 不再走 stable_id / 几何匹配，避免解析到未随实例变换的原始面。
|
||
try:
|
||
plane = PlaneSpec.from_mapping(selector.get("frame") or {})
|
||
except (TypeError, ValueError):
|
||
plane = None
|
||
if plane is not None:
|
||
record = TopologyRecord(
|
||
record_id=f"inline:{id(plane)}",
|
||
kind="plane",
|
||
feature_id=str(owner or "inline"),
|
||
geometry=plane.as_dict(),
|
||
value=plane,
|
||
)
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="resolved",
|
||
record=record,
|
||
candidates=({"score": 1.0, **record.public_dict()},),
|
||
)
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_frame_incomplete",
|
||
message="An inline plane frame requires origin_mm, x_dir and normal",
|
||
detail={"frame": selector.get("frame")},
|
||
),
|
||
)
|
||
geometry = normalize_selector_geometry(selector.get("geometry"))
|
||
if selector.get("snapshot_id") and not owner:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_owner_required",
|
||
message="A snapshot selector requires owner_feature_id",
|
||
detail={"minimum_score": minimum_score},
|
||
),
|
||
)
|
||
output_role = str(selector.get("output_role") or "").strip()
|
||
if output_role:
|
||
if not owner:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_output_role_owner_required",
|
||
message="A feature output role selector requires owner_feature_id",
|
||
detail={"output_role": output_role},
|
||
),
|
||
)
|
||
if active_body_id is None:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_output_role_active_body_required",
|
||
message="A feature output role selector requires an active body snapshot",
|
||
detail={"output_role": output_role},
|
||
),
|
||
)
|
||
if any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id")):
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_output_role_mixed_evidence",
|
||
message="A feature output role selector cannot mix stable or geometry evidence",
|
||
detail={"output_role": output_role},
|
||
),
|
||
)
|
||
role_candidates = [record for record in candidates if output_role in record.output_roles]
|
||
role_source = selector.get("output_role_source")
|
||
if role_source is not None:
|
||
source_owner = role_source.get("owner_feature_id") if isinstance(role_source, dict) else None
|
||
source_role = role_source.get("output_role") if isinstance(role_source, dict) else None
|
||
if not isinstance(source_owner, str) or not isinstance(source_role, str):
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_output_role_source_invalid",
|
||
message="An output role selector source requires owner_feature_id and output_role",
|
||
),
|
||
)
|
||
if output_role != "shell.offset_face" or source_role not in {"extrude.start", "extrude.end"}:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_output_role_source_unsupported",
|
||
message="Output role sources are currently supported only for shell.offset_face from an extrusion cap",
|
||
),
|
||
)
|
||
role_candidates = [
|
||
record for record in role_candidates
|
||
if (output_role, source_owner, source_role) in record.output_role_sources
|
||
]
|
||
public_candidates = tuple(
|
||
{"score": 1.0, **record.public_dict()} for record in role_candidates
|
||
)
|
||
if len(role_candidates) == 1:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="resolved",
|
||
record=role_candidates[0],
|
||
candidates=public_candidates,
|
||
)
|
||
if len(role_candidates) > 1:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="ambiguous",
|
||
candidates=public_candidates,
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_output_role_ambiguous",
|
||
message="More than one active topology record has the requested output role",
|
||
detail={"output_role": output_role, "candidate_count": len(role_candidates)},
|
||
),
|
||
)
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_output_role_not_found",
|
||
message="No active topology record has the requested output role",
|
||
detail={"output_role": output_role, "candidate_count": 0},
|
||
),
|
||
)
|
||
stable_id = str(selector.get("stable_id") or "").strip()
|
||
if stable_id:
|
||
# #8 selector 持久性：stable_id 是跨 body 演化的持久标识符，精确
|
||
# 匹配在 active body 过滤之前对整个记录集（kind + owner 过滤）执行。
|
||
# 命中已过期（旧 body）的记录时，经演化后继映射解析到 active body
|
||
# 内的新形态（fillet/chamfer 拆段后的直段后继）；无后继则回落到
|
||
# 几何打分流程。
|
||
stable_records = [
|
||
record for record in self._records
|
||
if record.kind == kind and (not owner or owner in record.owners)
|
||
]
|
||
exact = [record for record in stable_records if record.record_id == stable_id]
|
||
if len(exact) == 1:
|
||
record = exact[0]
|
||
is_active = active_body_id is None or (
|
||
record.body_id == active_body_id
|
||
or (record.body_id is not None and record.body_id.startswith(f"{active_body_id}:"))
|
||
)
|
||
if not is_active:
|
||
successors = [
|
||
candidate for candidate in stable_records
|
||
if candidate.record_id in self._successors.get(record.record_id, ())
|
||
and (
|
||
candidate.body_id == active_body_id
|
||
or (candidate.body_id is not None and active_body_id and candidate.body_id.startswith(f"{active_body_id}:"))
|
||
)
|
||
]
|
||
if len(successors) == 1:
|
||
record = successors[0]
|
||
is_active = True
|
||
elif len(successors) > 1:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="ambiguous",
|
||
candidates=tuple({"score": 1.0, **candidate.public_dict()} for candidate in successors),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_ambiguous",
|
||
message="More than one evolved successor record satisfies the stable_id",
|
||
detail={"stable_id": stable_id, "candidate_count": len(successors)},
|
||
),
|
||
)
|
||
if is_active:
|
||
# A stable ID is only a lookup accelerator for snapshot-aware
|
||
# selectors. It cannot revive a B-rep entity whose geometric
|
||
# signature changed after an upstream rebuild.
|
||
if selector.get("snapshot_id"):
|
||
score = self._geometry_score(geometry, record.geometry) if geometry else None
|
||
if score is None or score < minimum_score:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=({"score": round(float(score or 0), 6), **record.public_dict()},),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_geometry_mismatch",
|
||
message="The stable selector record no longer matches its geometry signature",
|
||
detail={"stable_id": stable_id, "score": score, "minimum_score": minimum_score},
|
||
),
|
||
)
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="resolved",
|
||
record=record,
|
||
candidates=({"score": round(float(score), 6) if selector.get("snapshot_id") else 1.0, **record.public_dict()},),
|
||
)
|
||
if not geometry:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_stable_id_inactive",
|
||
message="The stable selector record is not active and has no geometry signature for rebinding",
|
||
detail={"stable_id": stable_id},
|
||
),
|
||
)
|
||
if len(exact) > 1:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="ambiguous",
|
||
candidates=tuple({"score": 1.0, **record.public_dict()} for record in exact),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_ambiguous",
|
||
message="More than one runtime topology record has the requested stable_id",
|
||
detail={"stable_id": stable_id, "candidate_count": len(exact)},
|
||
),
|
||
)
|
||
if selector.get("snapshot_id") and not geometry:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=(),
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_geometry_mismatch",
|
||
message="A snapshot selector requires a geometry signature",
|
||
detail={"minimum_score": minimum_score},
|
||
),
|
||
)
|
||
scored: list[tuple[float, TopologyRecord]] = []
|
||
for candidate in candidates:
|
||
# An owner-qualified context selector is deterministic when it has
|
||
# a single runtime candidate even if its source stable_id cannot
|
||
# survive the SolidWorks -> OCC boundary.
|
||
score = 1.0 if not geometry else self._geometry_score(geometry, candidate.geometry)
|
||
if score is not None:
|
||
scored.append((score, candidate))
|
||
scored.sort(key=lambda item: (-item[0], item[1].record_id))
|
||
public_candidates = tuple({"score": round(score, 6), **record.public_dict()} for score, record in scored)
|
||
if not scored or scored[0][0] < minimum_score:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="not_found",
|
||
candidates=public_candidates,
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_not_found",
|
||
message="No runtime topology record satisfies the selector",
|
||
detail={"candidate_count": len(scored), "minimum_score": minimum_score},
|
||
),
|
||
)
|
||
best_score, best_record = scored[0]
|
||
if len(scored) > 1 and abs(scored[1][0] - best_score) <= 1e-9:
|
||
return SelectorResolution(
|
||
selector=selector,
|
||
status="ambiguous",
|
||
candidates=public_candidates,
|
||
diagnostic=RuntimeDiagnostic(
|
||
code="selector_ambiguous",
|
||
message="More than one runtime topology record has the best selector score",
|
||
detail={"best_score": best_score, "candidate_count": len(scored)},
|
||
),
|
||
)
|
||
return SelectorResolution(selector=selector, status="resolved", record=best_record, candidates=public_candidates)
|