257 lines
16 KiB
Python
257 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
import tempfile
|
|
from typing import Any
|
|
|
|
|
|
def _score(expected: dict[str, Any], actual: dict[str, Any]) -> float | None:
|
|
scores: list[float] = []
|
|
reversed_plane_normal = False
|
|
for key in ("center_mm", "circle_center_mm", "start_mm", "end_mm", "normal", "axis_origin_mm", "axis_direction"):
|
|
if key in expected:
|
|
# plane_offset_mm 与平面方程绑定，必须使用记录平面方程时采用的
|
|
# plane_normal。face 的局部采样 normal 在 OCC 中可能与其相反。
|
|
use_plane_normal = key == "normal" and "plane_offset_mm" in expected and actual.get("plane_normal") is not None
|
|
right = actual.get("plane_normal") if use_plane_normal else actual.get(key)
|
|
left = expected[key]
|
|
if not isinstance(right, (list, tuple)) or len(left) != len(right): return None
|
|
delta = math.sqrt(sum((float(a) - float(b)) ** 2 for a, b in zip(left, right)))
|
|
if key in {"normal", "axis_direction"}:
|
|
# A source CAP_FACE identifies a geometric plane, not the OCC
|
|
# orientation of the resulting face. The two kernels may
|
|
# report the same cap with inverse normals, especially for a
|
|
# start cap. A rotational-face axis is likewise a geometric
|
|
# line, whose direction may be reversed by OCC.
|
|
reversed_delta = math.sqrt(sum((float(a) + float(b)) ** 2 for a, b in zip(left, right)))
|
|
if key == "normal" and use_plane_normal and reversed_delta < delta: reversed_plane_normal = True
|
|
delta = min(delta, reversed_delta)
|
|
scores.append(max(0.0, 1.0 - delta / 0.05))
|
|
for key in ("radius_mm", "plane_offset_mm"):
|
|
if key in expected:
|
|
try:
|
|
value = float(actual[key])
|
|
# 平面偏移是 normal · point。面法向反向时，同一几何平面的
|
|
# 有符号偏移也必须同步反号，不能只放宽 normal 的比较。
|
|
if key == "plane_offset_mm" and reversed_plane_normal: value = -value
|
|
delta = abs(float(expected[key]) - value)
|
|
except Exception: return None
|
|
scores.append(max(0.0, 1.0 - delta / 0.05))
|
|
if "minimum_area_mm2" in expected:
|
|
try:
|
|
if float(actual["area_mm2"]) + 1e-6 < float(expected["minimum_area_mm2"]): return None
|
|
except Exception: return None
|
|
if "bbox_mm" in expected:
|
|
actual_box = actual.get("bbox_mm")
|
|
if not isinstance(actual_box, (list, tuple)) or len(actual_box) != 6: return None
|
|
delta = max(abs(float(a) - float(b)) for a, b in zip(expected["bbox_mm"], actual_box))
|
|
scores.append(max(0.0, 1.0 - delta / 0.05))
|
|
return sum(scores) / len(scores) if scores else 0.0
|
|
|
|
|
|
def bind_selector(kind: str, owner_feature_id: str, geometry: dict[str, Any], records: list[dict[str, Any]], *, minimum_score: float = 0.8) -> dict[str, Any]:
|
|
candidates = []
|
|
for record in records:
|
|
owners = record.get("owner_feature_ids") or [record.get("feature_id")]
|
|
if record.get("kind") != kind or owner_feature_id not in owners: continue
|
|
score = _score(geometry, record.get("geometry") or {})
|
|
if score is not None and score >= minimum_score: candidates.append((score, record))
|
|
candidates.sort(key=lambda item: (-item[0], str(item[1].get("record_id"))))
|
|
if not candidates: raise ValueError("selector_not_found")
|
|
if len(candidates) > 1 and abs(candidates[0][0] - candidates[1][0]) <= 1e-9: raise ValueError("selector_ambiguous")
|
|
score, record = candidates[0]
|
|
return {"kind": kind, "owner_feature_id": owner_feature_id, "stable_id": record["record_id"], "snapshot_id": record["record_id"], "source": "runtime_snapshot", "confidence": round(score, 6), "geometry": record.get("geometry") or geometry}
|
|
|
|
|
|
def _dot(left: list[float], right: list[float]) -> float: return sum(float(a)*float(b) for a, b in zip(left, right))
|
|
|
|
|
|
def _circle_records(expected: dict[str, Any], records: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
center = expected.get("source_circle_center_mm"); normal = expected.get("source_plane_normal"); radius = float(expected.get("source_circle_radius_mm") or 0)
|
|
if not isinstance(center, list) or not isinstance(normal, list) or radius <= 0: return []
|
|
plane_offset = _dot(center, normal); matches = []
|
|
for record in records:
|
|
geometry = record.get("geometry") or {}
|
|
if record.get("kind") != "edge" or geometry.get("curve_type") != "circle": continue
|
|
points = [geometry.get("start_mm"), geometry.get("end_mm")]
|
|
if not all(isinstance(point, list) and len(point) == 3 for point in points): continue
|
|
if any(abs(_dot(point, normal) - plane_offset) > 0.05 for point in points): continue
|
|
radial = []
|
|
for point in points:
|
|
delta = [float(point[i])-float(center[i]) for i in range(3)]; axial = _dot(delta, normal)
|
|
radial.append(math.sqrt(max(0.0, sum(value*value for value in delta)-axial*axial)))
|
|
if all(abs(value-radius) <= max(0.05, radius*1e-4) for value in radial): matches.append(record)
|
|
return matches
|
|
|
|
|
|
def _binding_targets(selector: dict[str, Any]):
|
|
components = selector.get("intersection_of")
|
|
if isinstance(components, list):
|
|
for component in components:
|
|
if isinstance(component, dict): yield from _binding_targets(component)
|
|
return
|
|
yield selector
|
|
|
|
|
|
def _bound_selector(placeholder: dict[str, Any], records: list[dict[str, Any]]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
bound = []
|
|
for record in records:
|
|
owners = record.get("owner_feature_ids") or [record.get("feature_id")]
|
|
selector = {"kind": placeholder["kind"], "owner_feature_id": str(owners[0]), "stable_id": record["record_id"], "snapshot_id": record["record_id"], "source": "runtime_snapshot", "confidence": 1.0, "geometry": record.get("geometry") or {}}
|
|
if placeholder.get("binding_feature_id") is not None: selector["binding_feature_id"] = placeholder["binding_feature_id"]
|
|
if placeholder.get("owner_match_required"): selector["owner_match_required"] = True
|
|
bound.append(selector)
|
|
if placeholder.get("match_mode") == "all":
|
|
selector = dict(placeholder); selector["matched_selectors"] = bound
|
|
return selector, bound
|
|
return bound[0], bound
|
|
|
|
|
|
def bind_candidate_selectors(cdsl: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
"""Rebuild every selector-bearing prefix and bind against its active body."""
|
|
from engine.cdsl_engine.runtime import rebuild_cdsl
|
|
bound = deepcopy(cdsl); evidence = []
|
|
with tempfile.TemporaryDirectory(prefix="cadfs-bind-") as temporary:
|
|
for index, feature in enumerate(bound.get("features") or []):
|
|
roots = list(feature.get("selectors") or [])
|
|
for name in ("end_condition", "reverse_end_condition"):
|
|
condition = (feature.get("params") or {}).get(name)
|
|
reference = condition.get("reference") if isinstance(condition, dict) else None
|
|
if isinstance(reference, dict): roots.append(reference)
|
|
targets = [target for root in roots for target in _binding_targets(root)]
|
|
if not targets: continue
|
|
prefix_cache: dict[int, tuple[list[dict[str, Any]], str | None]] = {}
|
|
|
|
def prefix_records(binding_feature_id: str | None, owner_feature_id: str | None) -> list[dict[str, Any]]:
|
|
prefix_count = index
|
|
if binding_feature_id is not None:
|
|
binding_index = next((item_index for item_index, item in enumerate(bound["features"][:index]) if item["id"] == binding_feature_id), None)
|
|
if binding_index is None: raise ValueError(f"{feature['id']}: selector binding feature is missing or forward")
|
|
prefix_count = binding_index + 1
|
|
if prefix_count not in prefix_cache:
|
|
prefix = deepcopy(bound); prefix["features"] = bound["features"][:prefix_count]
|
|
if not prefix["features"]: raise ValueError(f"{feature['id']}: selector has no executable prefix")
|
|
report = rebuild_cdsl(prefix, Path(temporary) / f"prefix-{index}-{prefix_count}.step", strict=True)
|
|
body_id = next((item.get("body_id") for item in reversed(report.get("feature_results") or []) if item.get("body_id")), None)
|
|
prefix_cache[prefix_count] = (list(report.get("topology_records") or []), body_id)
|
|
all_records, body_id = prefix_cache[prefix_count]
|
|
return [
|
|
item for item in all_records
|
|
# Reference planes and axes are session context, not body
|
|
# topology. They must remain available while binding a mirror
|
|
# or extent selector against a body-bearing prefix.
|
|
if item.get("kind") in {"plane", "axis"}
|
|
or not body_id
|
|
or item.get("body_id") == body_id
|
|
or str(item.get("body_id") or "").startswith(f"{body_id}:")
|
|
]
|
|
|
|
resolved = []
|
|
for placeholder in targets:
|
|
records = prefix_records(placeholder.get("binding_feature_id"), placeholder.get("owner_feature_id"))
|
|
output_role = str(placeholder.get("output_role") or "").strip()
|
|
if output_role:
|
|
# Builder output roles are not geometry placeholders. They
|
|
# remain in the bound CDSL so runtime can resolve the
|
|
# current active B-rep face through exact kernel history.
|
|
# Replacing one with stable_id/geometry would mix evidence
|
|
# and make a stale snapshot appear durable.
|
|
owner = placeholder.get("owner_feature_id")
|
|
role_source = placeholder.get("output_role_source")
|
|
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
|
|
candidates = [
|
|
record for record in records
|
|
if record.get("kind") == placeholder.get("kind")
|
|
and owner in (record.get("owner_feature_ids") or [record.get("feature_id")])
|
|
and output_role in (record.get("output_roles") or [])
|
|
and (
|
|
role_source is None
|
|
or any(
|
|
item.get("output_role") == output_role
|
|
and item.get("owner_feature_id") == source_owner
|
|
and item.get("source_output_role") == source_role
|
|
for item in record.get("output_role_sources") or []
|
|
)
|
|
)
|
|
]
|
|
if len(candidates) != 1:
|
|
status = "not_found" if not candidates else "ambiguous"
|
|
raise ValueError(
|
|
f"{feature['id']}: selector_output_role_{status} after prefix rebuild"
|
|
)
|
|
resolved.append(candidates[0])
|
|
continue
|
|
geometry = placeholder.get("geometry") or {}
|
|
candidates = _circle_records(geometry, records) if geometry.get("source_circle_radius_mm") else []
|
|
if not candidates:
|
|
same_kind = [record for record in records if record.get("kind") == placeholder.get("kind")]
|
|
owner = placeholder.get("owner_feature_id")
|
|
owner_matches = [record for record in same_kind if owner in (record.get("owner_feature_ids") or [record.get("feature_id")])]
|
|
# Pattern instance provenance is stronger than an ordinary
|
|
# feature owner: an unresolved instance cannot fall back to
|
|
# an aggregate face with matching geometry. Ordinary source
|
|
# selectors retain their established geometry-binding path
|
|
# across topology-changing dress-up operations.
|
|
pool = owner_matches if placeholder.get("owner_match_required") else (owner_matches or same_kind)
|
|
if not geometry:
|
|
# Context selectors (notably a generated mirror plane)
|
|
# may have no geometric snapshot. Their owner-qualified
|
|
# singleton identity is sufficient and must not be scored
|
|
# as a zero-information geometric match.
|
|
if len(pool) != 1:
|
|
status = "not_found" if not pool else "ambiguous"
|
|
raise ValueError(f"{feature['id']}: selector_{status} after prefix rebuild")
|
|
candidates = [pool[0]]
|
|
else:
|
|
scored = [(score, record) for record in pool if (score := _score(geometry, record.get("geometry") or {})) is not None and score >= 0.8]
|
|
# A source owner is preferred for ordinary geometry
|
|
# selectors, but can be retained by an unrelated exact
|
|
# continuation after a dress-up. If none of that
|
|
# owner's active records satisfies the full geometry
|
|
# signature, bind against the current active body and
|
|
# still require a unique threshold-qualified match.
|
|
# Instance-qualified selectors never take this path.
|
|
if not scored and not placeholder.get("owner_match_required"):
|
|
scored = [
|
|
(score, record)
|
|
for record in same_kind
|
|
if (score := _score(geometry, record.get("geometry") or {})) is not None
|
|
and score >= 0.8
|
|
]
|
|
scored.sort(key=lambda value: (-value[0], str(value[1].get("record_id"))))
|
|
if scored:
|
|
if placeholder.get("match_mode") != "all" and len(scored) > 1 and abs(scored[0][0] - scored[1][0]) <= 1e-9:
|
|
raise ValueError(f"{feature['id']}: selector_ambiguous after prefix rebuild")
|
|
candidates = [item[1] for item in scored] if placeholder.get("match_mode") == "all" else [scored[0][1]]
|
|
if not candidates: raise ValueError(f"{feature['id']}: selector_not_found after prefix rebuild")
|
|
selector, bound_selectors = _bound_selector(placeholder, candidates)
|
|
placeholder.clear(); placeholder.update(selector)
|
|
resolved.extend(bound_selectors)
|
|
feature_selectors = feature.get("selectors") or []
|
|
def selector_key(selector: dict[str, Any]) -> tuple[Any, ...]:
|
|
stable_id = selector.get("stable_id")
|
|
if stable_id is not None:
|
|
return ("stable_id", str(stable_id))
|
|
source = selector.get("output_role_source")
|
|
return (
|
|
"output_role",
|
|
selector.get("owner_feature_id"),
|
|
selector.get("kind"),
|
|
selector.get("output_role"),
|
|
source.get("owner_feature_id") if isinstance(source, dict) else None,
|
|
source.get("output_role") if isinstance(source, dict) else None,
|
|
)
|
|
|
|
unique = {selector_key(selector): selector for selector in feature_selectors}
|
|
feature["selectors"] = list(unique.values())
|
|
if feature.get("atomic_id") == "pattern_mirror":
|
|
planes = [selector for selector in feature["selectors"] if selector.get("kind") == "plane"]
|
|
if len(planes) != 1:
|
|
raise ValueError(f"{feature['id']}: mirror plane binding is not unique")
|
|
feature.setdefault("params", {})["mirror_plane"] = planes[0]
|
|
evidence.append({"feature_id": feature["id"], "prefix_feature_count": index, "selectors": feature["selectors"], "resolved": resolved})
|
|
return bound, evidence
|