155 lines
6.5 KiB
Python
155 lines
6.5 KiB
Python
"""Derived-metrics backfill for CDSL documents (training-grade, server-owned).
|
|
|
|
The metrics describe the *reconstructed geometry* of a document and are
|
|
written exclusively by the server after a successful rebuild. Models never
|
|
author them; the fields are excluded from validation ordering (backfilled
|
|
after validation) so they can never influence acceptance.
|
|
|
|
Counting conventions:
|
|
- ``hole_count`` : inner cylindrical walls on the final body (cylinder faces
|
|
whose ``cylinder_role`` is ``inner``). Countersink/counterbore bores add
|
|
their own walls; no coaxial clustering is applied in v1.
|
|
- ``fillet_count``: torus faces (rolling-ball blends) on the final body.
|
|
- ``face_type_distribution``: final-body faces by ``surface_type``.
|
|
- ``per_feature`` : prefix re-execution sampling. Prefix ``i`` re-runs the
|
|
first ``i`` features; sampling stops at the first non-executable prefix.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
import tempfile
|
|
from typing import Any, Callable
|
|
|
|
from app.services.engine_service import topology_snapshot
|
|
|
|
_METRIC_SNAPSHOT_KEYS = (
|
|
"volume_mm3", "surface_area_mm2", "bbox_mm", "solid_count",
|
|
"face_count", "edge_count", "vertex_count", "hole_count",
|
|
"fillet_count", "face_type_distribution",
|
|
)
|
|
_DELTA_KEYS = ("volume_mm3", "surface_area_mm2", "face_count", "hole_count")
|
|
|
|
#: Prefix sampling above this feature count degrades to final-only metrics,
|
|
#: keeping the O(n^2) re-execution cost bounded on very large documents.
|
|
MAX_PER_FEATURE_SAMPLES = 64
|
|
|
|
|
|
def _face_stats(topology: dict[str, Any]) -> dict[str, Any]:
|
|
faces = [
|
|
record for record in (topology.get("records") or [])
|
|
if isinstance(record, dict) and record.get("kind") == "face"
|
|
]
|
|
distribution: Counter[str] = Counter()
|
|
hole_walls = 0
|
|
torus = 0
|
|
for face in faces:
|
|
geometry = face.get("geometry") or {}
|
|
surface_type = str(geometry.get("surface_type") or "unknown")
|
|
distribution[surface_type] += 1
|
|
if surface_type == "cylinder" and geometry.get("cylinder_role") == "inner":
|
|
hole_walls += 1
|
|
if surface_type == "torus":
|
|
torus += 1
|
|
return {
|
|
"face_type_distribution": dict(sorted(distribution.items())),
|
|
"hole_count": hole_walls,
|
|
"fillet_count": torus,
|
|
}
|
|
|
|
|
|
def _snapshot_from_engine_result(engine_result: dict[str, Any], topology: dict[str, Any]) -> dict[str, Any]:
|
|
records = [r for r in (topology.get("records") or []) if isinstance(r, dict)]
|
|
faces = [r for r in records if r.get("kind") == "face"]
|
|
area = sum(float((f.get("geometry") or {}).get("area_mm2") or 0.0) for f in faces)
|
|
snapshot = {
|
|
"volume_mm3": float(engine_result.get("volume_mm3") or 0.0),
|
|
"surface_area_mm2": area or float(engine_result.get("surface_area_mm2") or 0.0),
|
|
"bbox_mm": [float(value) for value in (engine_result.get("bbox_mm") or {}).get("min", [])]
|
|
+ [float(value) for value in (engine_result.get("bbox_mm") or {}).get("max", [])],
|
|
"solid_count": int(engine_result.get("solid_count") or 0),
|
|
"face_count": len(faces),
|
|
"edge_count": sum(1 for r in records if r.get("kind") == "edge"),
|
|
"vertex_count": sum(1 for r in records if r.get("kind") == "vertex"),
|
|
}
|
|
snapshot.update(_face_stats(topology))
|
|
return {key: snapshot[key] for key in _METRIC_SNAPSHOT_KEYS}
|
|
|
|
|
|
def _delta(after: dict[str, Any], before: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"volume_mm3": float(after["volume_mm3"]) - float(before["volume_mm3"]),
|
|
"surface_area_mm2": float(after["surface_area_mm2"]) - float(before["surface_area_mm2"]),
|
|
"face_count": int(after["face_count"]) - int(before["face_count"]),
|
|
"hole_count": int(after["hole_count"]) - int(before["hole_count"]),
|
|
}
|
|
|
|
|
|
def derive_metrics(
|
|
cdsl: dict[str, Any],
|
|
engine_result: dict[str, Any],
|
|
topology: dict[str, Any],
|
|
*,
|
|
engine_build: str = "",
|
|
prefix_runner: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Build the ``meta.derived_metrics`` payload for a successfully rebuilt document.
|
|
|
|
``prefix_runner`` re-executes a prefix of the document's features and
|
|
returns the same shape as ``engine_result``; when omitted, ``per_feature``
|
|
degrades to status entries without geometric snapshots.
|
|
"""
|
|
features = cdsl.get("features") or []
|
|
final = _snapshot_from_engine_result(engine_result, topology)
|
|
zero = {
|
|
"volume_mm3": 0.0, "surface_area_mm2": 0.0,
|
|
"face_count": 0, "hole_count": 0,
|
|
}
|
|
|
|
per_feature: list[dict[str, Any]] = []
|
|
if prefix_runner is not None and 1 <= len(features) <= MAX_PER_FEATURE_SAMPLES:
|
|
previous = zero
|
|
for index in range(1, len(features) + 1):
|
|
prefix_doc = deepcopy(cdsl)
|
|
prefix_doc["features"] = deepcopy(features[:index])
|
|
try:
|
|
prefix_result = prefix_runner(prefix_doc)
|
|
topology_prefix = topology_snapshot(prefix_result)
|
|
after = _snapshot_from_engine_result(prefix_result, topology_prefix)
|
|
status = str(next(
|
|
(fr.get("status") for fr in prefix_result.get("feature_results") or []
|
|
if fr.get("feature_id") == features[index - 1].get("id")),
|
|
"executed",
|
|
))
|
|
except Exception:
|
|
break # prefix chain broken; keep samples collected so far
|
|
per_feature.append({
|
|
"feature_id": str(features[index - 1].get("id") or ""),
|
|
"atomic_id": str(features[index - 1].get("atomic_id") or ""),
|
|
"status": status,
|
|
"after": after,
|
|
"delta": _delta(after, previous),
|
|
})
|
|
previous = after
|
|
|
|
histogram = Counter(str(feature.get("atomic_id") or "") for feature in features)
|
|
metrics: dict[str, Any] = {
|
|
"engine": str(engine_result.get("engine") or ""),
|
|
"final": final,
|
|
"per_feature": per_feature,
|
|
"atomic_histogram": dict(sorted(histogram.items())),
|
|
}
|
|
if engine_build:
|
|
metrics["engine_build"] = engine_build
|
|
return metrics
|
|
|
|
|
|
def make_prefix_runner(run_cdsl_only: Callable[[dict[str, Any], Path], dict[str, Any]]):
|
|
"""Wrap an engine ``run_cdsl_only`` entry point for prefix sampling."""
|
|
def runner(prefix_doc: dict[str, Any]) -> dict[str, Any]:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
return run_cdsl_only(prefix_doc, Path(tmp) / "prefix.step")
|
|
return runner
|