Files
cdsl-cad/backend/app/cad_agent/domain/verifier_registry.py
T
2026-09-01 14:10:23 +08:00

872 lines
51 KiB
Python

"""Deterministic acceptance-claim registry and closed-schema validation."""
from __future__ import annotations
from dataclasses import dataclass
from math import atan2, isclose, pi, sqrt
from typing import Any, Callable
from jsonschema import Draft202012Validator
from jsonschema.exceptions import SchemaError
ClaimResult = dict[str, Any]
ClaimEvaluator = Callable[[dict[str, Any], dict[str, Any]], ClaimResult]
def _closed_object(properties: dict[str, Any], required: list[str]) -> dict[str, Any]:
return {"type": "object", "properties": properties, "required": required, "additionalProperties": False}
def _vector(value: Any) -> tuple[float, float, float] | None:
if not isinstance(value, list) or len(value) != 3:
return None
try:
return tuple(float(item) for item in value) # type: ignore[return-value]
except (TypeError, ValueError):
return None
def _cylinder_axis_point(record: dict[str, Any]) -> tuple[float, float, float] | None:
"""Return an actual point on a cylinder's axis, never a surface centroid."""
geometry = record.get("geometry") if isinstance(record.get("geometry"), dict) else {}
return _vector(geometry.get("axis_origin_mm")) or _vector(geometry.get("center_mm"))
def _cylinder_axis_direction(record: dict[str, Any]) -> tuple[float, float, float] | None:
geometry = record.get("geometry") if isinstance(record.get("geometry"), dict) else {}
direction = _vector(geometry.get("axis_direction"))
if direction is None:
return None
length = sqrt(sum(component * component for component in direction))
return tuple(component / length for component in direction) if length > 1e-9 else None
def _cylinder_axis_span(record: dict[str, Any]) -> float | None:
"""Return the axial span of a cylinder face from its B-rep bounding box."""
interval = _cylinder_axis_interval(record)
return None if interval is None else interval[1] - interval[0]
def _cylinder_axis_interval(record: dict[str, Any]) -> tuple[float, float] | None:
"""Return the inclusive axial interval of a cylindrical face's bounds."""
geometry = record.get("geometry") if isinstance(record.get("geometry"), dict) else {}
bbox = geometry.get("bbox_mm")
direction = _cylinder_axis_direction(record)
if not isinstance(bbox, list) or len(bbox) != 6 or direction is None:
return None
try:
minimum = tuple(float(value) for value in bbox[:3])
maximum = tuple(float(value) for value in bbox[3:])
except (TypeError, ValueError):
return None
projections = [
sum(direction[index] * point[index] for index in range(3))
for point in (
(x, y, z)
for x in (minimum[0], maximum[0])
for y in (minimum[1], maximum[1])
for z in (minimum[2], maximum[2])
)
]
return min(projections), max(projections)
def _records(facts: dict[str, Any]) -> list[dict[str, Any]]:
topology = facts.get("topology") if isinstance(facts.get("topology"), dict) else {}
return [item for item in topology.get("records") or () if isinstance(item, dict)]
def _pass(evidence: dict[str, Any]) -> ClaimResult:
return {"status": "pass", "evidence": evidence}
def _fail(evidence: dict[str, Any]) -> ClaimResult:
return {"status": "fail", "evidence": evidence}
def _pending(reason: str) -> ClaimResult:
return {"status": "pending", "evidence": {"reason": reason}}
def _solid_count(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
health = facts.get("health") if isinstance(facts.get("health"), dict) else {}
actual = health.get("solid_count")
if not isinstance(actual, int):
return _pending("rebuild report has no solid_count")
return _pass({"solid_count": actual}) if actual == expected["value"] else _fail({"expected": expected["value"], "actual": actual})
def _volume_decreased(_expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
"""Prove that a subtractive operation removed a measurable amount of material."""
health = facts.get("health") if isinstance(facts.get("health"), dict) else {}
parent_health = facts.get("parent_health") if isinstance(facts.get("parent_health"), dict) else {}
parent_volume = parent_health.get("volume_mm3")
candidate_volume = health.get("volume_mm3")
if not isinstance(parent_volume, (int, float)) or not isinstance(candidate_volume, (int, float)):
return _pending("parent or candidate rebuild report has no measurable volume")
removed = float(parent_volume) - float(candidate_volume)
evidence = {
"parent_volume_mm3": float(parent_volume),
"candidate_volume_mm3": float(candidate_volume),
"removed_volume_mm3": removed,
}
return _pass(evidence) if removed > 1e-6 else _fail(evidence)
def _bbox(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
health = facts.get("health") if isinstance(facts.get("health"), dict) else {}
dimensions = ((health.get("bbox_mm") or {}).get("dimensions") if isinstance(health.get("bbox_mm"), dict) else None)
axis = {"x": 0, "y": 1, "z": 2}[expected["axis"]]
if not isinstance(dimensions, list) or len(dimensions) != 3:
return _pending("rebuild report has no bounding-box dimensions")
actual = float(dimensions[axis])
tolerance = float(expected["tolerance_mm"])
return _pass({"axis": expected["axis"], "actual_mm": actual, "tolerance_mm": tolerance}) if abs(actual - expected["value"]) <= tolerance else _fail({"axis": expected["axis"], "expected_mm": expected["value"], "actual_mm": actual, "tolerance_mm": tolerance})
def _bbox_rank(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
"""Measure a principal extent without assuming the author's world axes."""
health = facts.get("health") if isinstance(facts.get("health"), dict) else {}
dimensions = ((health.get("bbox_mm") or {}).get("dimensions") if isinstance(health.get("bbox_mm"), dict) else None)
if not isinstance(dimensions, list) or len(dimensions) != 3:
return _pending("rebuild report has no bounding-box dimensions")
try:
ranked = sorted(float(value) for value in dimensions)
except (TypeError, ValueError):
return _pending("rebuild report has invalid bounding-box dimensions")
index = {"minimum": 0, "median": 1, "maximum": 2}[expected["rank"]]
actual = ranked[index]
tolerance = float(expected["tolerance_mm"])
return _pass({"rank": expected["rank"], "actual_mm": actual, "tolerance_mm": tolerance}) if abs(actual - expected["value"]) <= tolerance else _fail({"rank": expected["rank"], "expected_mm": expected["value"], "actual_mm": actual, "tolerance_mm": tolerance})
def _matching_cylinders(expected: dict[str, Any], facts: dict[str, Any]) -> list[dict[str, Any]]:
results: list[dict[str, Any]] = []
diameter = float(expected["diameter_mm"])
tolerance = float(expected.get("tolerance_mm", 0.01))
for record in _records(facts):
geometry = record.get("geometry") if isinstance(record.get("geometry"), dict) else {}
if geometry.get("surface_type") != "cylinder":
continue
# A cylinder bounded by two opposing planes can be either an external
# wall or an internal bore. Modern topology snapshots make this
# distinction explicit from B-rep face orientation. Do not accept an
# unknown/export-incomplete cylinder as a bore: that would reintroduce
# the false positive this verifier exists to prevent.
if geometry.get("cylinder_role") != "inner":
continue
radius = geometry.get("radius_mm")
if isinstance(radius, (int, float)) and abs(float(radius) * 2 - diameter) <= tolerance:
results.append(record)
by_id = {
str(record.get("record_id") or ""): record
for record in _records(facts)
if isinstance(record.get("record_id"), str)
}
return [
_merged_inner_cylindrical_shell(group, by_id)
for group in _cylindrical_shells(results, tolerance)
]
def _matching_outer_cylinders(expected: dict[str, Any], facts: dict[str, Any]) -> list[dict[str, Any]]:
results: list[dict[str, Any]] = []
diameter = float(expected["diameter_mm"])
tolerance = float(expected.get("tolerance_mm", 0.01))
for record in _records(facts):
geometry = record.get("geometry") if isinstance(record.get("geometry"), dict) else {}
if geometry.get("surface_type") != "cylinder" or geometry.get("cylinder_role") != "outer":
continue
radius = geometry.get("radius_mm")
if isinstance(radius, (int, float)) and abs(float(radius) * 2 - diameter) <= tolerance:
results.append(record)
return results
def _same_cylindrical_shell(left: dict[str, Any], right: dict[str, Any], tolerance: float) -> bool:
"""Identify B-rep patches that represent one continuous cylindrical shell.
OpenCascade can split an extruded circular wall into four periodic-face
patches. Those patches have the same axis/radius and axial interval, but
are not four physical outer diameters. Conversely, equal-diameter shaft
sections separated along their axis must remain distinct claim instances.
"""
left_axis = _cylinder_axis_direction(left)
right_axis = _cylinder_axis_direction(right)
left_origin = _cylinder_axis_point(left)
right_origin = _cylinder_axis_point(right)
left_interval = _cylinder_axis_interval(left)
right_interval = _cylinder_axis_interval(right)
if None in {left_axis, right_axis, left_origin, right_origin, left_interval, right_interval}:
return False
assert left_axis is not None and right_axis is not None
assert left_origin is not None and right_origin is not None
assert left_interval is not None and right_interval is not None
alignment = abs(sum(left_axis[index] * right_axis[index] for index in range(3)))
if alignment < 1.0 - 1e-6:
return False
delta = tuple(right_origin[index] - left_origin[index] for index in range(3))
axial = sum(delta[index] * left_axis[index] for index in range(3))
radial_offset = sqrt(sum((delta[index] - axial * left_axis[index]) ** 2 for index in range(3)))
if radial_offset > tolerance:
return False
return left_interval[0] <= right_interval[1] + tolerance and right_interval[0] <= left_interval[1] + tolerance
def _cylindrical_shells(records: list[dict[str, Any]], tolerance: float) -> list[list[dict[str, Any]]]:
"""Group periodic B-rep cylinder patches into physical cylindrical shells."""
groups: list[list[dict[str, Any]]] = []
for record in records:
overlapping = [index for index, group in enumerate(groups) if any(_same_cylindrical_shell(record, member, tolerance) for member in group)]
if not overlapping:
groups.append([record])
continue
first = overlapping[0]
groups[first].append(record)
for index in reversed(overlapping[1:]):
groups[first].extend(groups.pop(index))
return groups
def _merged_inner_cylindrical_shell(
members: list[dict[str, Any]],
records_by_id: dict[str, dict[str, Any]],
) -> dict[str, Any]:
"""Build one verifier-facing bore record from periodic inner-face patches.
Extruding an analytic circle can produce several cylindrical B-rep faces.
Individual patches touch only one end plane, while the union represents a
through bore. Keep topology records raw for selectors, but aggregate the
measured surface for acceptance claims.
"""
representative = members[0]
geometry = representative.get("geometry") if isinstance(representative.get("geometry"), dict) else {}
merged_geometry = dict(geometry)
member_ids = [str(member.get("record_id") or "") for member in members if member.get("record_id")]
connected_plane_ids = list(dict.fromkeys(
str(plane_id)
for member in members
for plane_id in ((member.get("geometry") or {}).get("connected_plane_ids") or ())
if isinstance(plane_id, str)
))
merged_geometry["member_record_ids"] = member_ids
merged_geometry["connected_plane_ids"] = connected_plane_ids
axis = _cylinder_axis_direction(representative)
merged_geometry["through"] = bool(geometry.get("through")) or _planes_prove_through(
connected_plane_ids,
axis,
records_by_id,
)
return {**representative, "geometry": merged_geometry}
def _planes_prove_through(
plane_ids: list[str],
axis: tuple[float, float, float] | None,
records_by_id: dict[str, dict[str, Any]],
) -> bool:
if axis is None:
return False
planes = [
records_by_id[record_id].get("geometry")
for record_id in plane_ids
if isinstance(records_by_id.get(record_id), dict)
and isinstance(records_by_id[record_id].get("geometry"), dict)
and records_by_id[record_id]["geometry"].get("surface_type") == "plane"
]
for first_index, first in enumerate(planes):
if not isinstance(first, dict):
continue
first_normal = _vector(first.get("normal"))
if first_normal is None:
continue
for second in planes[first_index + 1:]:
if not isinstance(second, dict):
continue
second_normal = _vector(second.get("normal"))
if second_normal is None:
continue
opposite = sum(first_normal[index] * second_normal[index] for index in range(3)) <= -0.99
axial = all(abs(sum(normal[index] * axis[index] for index in range(3))) >= 0.99 for normal in (first_normal, second_normal))
if opposite and axial:
return True
return False
def _outer_cylinder_shells(expected: dict[str, Any], facts: dict[str, Any]) -> list[list[dict[str, Any]]]:
"""Coalesce periodic B-rep patches into physical outer-cylinder shells."""
matched = _matching_outer_cylinders(expected, facts)
tolerance = float(expected.get("tolerance_mm", 0.01))
return _cylindrical_shells(matched, tolerance)
def _matching_cones(expected: dict[str, Any], facts: dict[str, Any]) -> list[dict[str, Any]]:
results: list[dict[str, Any]] = []
small = float(expected["small_diameter_mm"])
large = float(expected["large_diameter_mm"])
included_angle = float(expected["included_angle_deg"])
tolerance = float(expected.get("tolerance_mm", 0.01))
for record in _records(facts):
geometry = record.get("geometry") if isinstance(record.get("geometry"), dict) else {}
if geometry.get("surface_type") != "cone" or geometry.get("cylinder_role") != "inner":
continue
radii = geometry.get("boundary_radii_mm")
angle = geometry.get("semi_angle_deg")
if not isinstance(radii, list) or not isinstance(angle, (int, float)):
continue
values = [float(value) for value in radii if isinstance(value, (int, float))]
if not values:
continue
if (
abs(min(values) * 2 - small) <= tolerance
and abs(max(values) * 2 - large) <= tolerance
and abs(abs(float(angle)) * 2 - included_angle) <= 1e-4
):
results.append(record)
return results
def _cylindrical_bore(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
matched = _matching_cylinders(expected, facts)
count = int(expected.get("count", 1))
if not _records(facts):
return _pending("topology is unavailable")
if not matched:
return _pending("the target bore has not been introduced at this checkpoint")
if len(matched) < count:
return _pending("the target bore pattern is incomplete at this checkpoint")
return _pass({"matched_cylindrical_faces": [item.get("record_id") for item in matched]}) if len(matched) == count else _fail({"expected_count": count, "actual_count": len(matched), "diameter_mm": expected["diameter_mm"]})
def _through_cylindrical_bore(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
matched = _matching_cylinders(expected, facts)
if not _records(facts):
return _pending("topology is unavailable")
if not matched:
return _pending("the target bore has not been introduced at this checkpoint")
# A bore is through only when the B-rep exporter established that its
# cylindrical face links two oppositely oriented plane faces. The flag is
# derived from shared topology edges, not from its requested depth.
through = [item for item in matched if bool((item.get("geometry") or {}).get("through"))]
count = int(expected.get("count", 1))
if len(matched) < count and len(through) == len(matched):
return _pending("the target through-bore pattern is incomplete at this checkpoint")
return _pass({"through_bores": [item.get("record_id") for item in through]}) if len(matched) == count and len(through) == count else _fail({"expected_count": count, "actual_count": len(matched), "actual_through_count": len(through), "diameter_mm": expected["diameter_mm"]})
def _cylindrical_bore_depth(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
matched = _matching_cylinders(expected, facts)
count = int(expected.get("count", 1))
if not _records(facts):
return _pending("topology is unavailable")
if not matched:
return _pending("the target bore has not been introduced at this checkpoint")
if len(matched) < count:
return _pending("the target bore pattern is incomplete at this checkpoint")
if len(matched) != count:
return _fail({"expected_count": count, "actual_count": len(matched), "diameter_mm": expected["diameter_mm"]})
spans = [_cylinder_axis_span(record) for record in matched]
if any(span is None for span in spans):
return _pending("target bore faces have no measurable axial span")
depth = float(expected["depth_mm"])
tolerance = float(expected["tolerance_mm"])
actual = [float(span) for span in spans if span is not None]
return _pass({"diameter_mm": expected["diameter_mm"], "depths_mm": actual, "tolerance_mm": tolerance}) if all(abs(value - depth) <= tolerance for value in actual) else _fail({"diameter_mm": expected["diameter_mm"], "expected_depth_mm": depth, "actual_depths_mm": actual, "tolerance_mm": tolerance})
def _conical_bore(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
matched = _matching_cones(expected, facts)
count = int(expected.get("count", 1))
if not _records(facts):
return _pending("topology is unavailable")
if not matched:
return _pending("the target conical bore has not been introduced at this checkpoint")
if len(matched) < count:
return _pending("the target conical-bore pattern is incomplete at this checkpoint")
return _pass({"conical_bores": [item.get("record_id") for item in matched], "small_diameter_mm": expected["small_diameter_mm"], "large_diameter_mm": expected["large_diameter_mm"], "included_angle_deg": expected["included_angle_deg"]}) if len(matched) == count else _fail({"expected_count": count, "actual_count": len(matched), "small_diameter_mm": expected["small_diameter_mm"], "large_diameter_mm": expected["large_diameter_mm"], "included_angle_deg": expected["included_angle_deg"]})
def _outer_cylindrical_surface(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
shells = _outer_cylinder_shells(expected, facts)
count = int(expected.get("count", 1))
if not _records(facts):
return _pending("topology is unavailable")
if not shells:
return _pending("the target external cylindrical surface has not been introduced at this checkpoint")
if len(shells) < count:
return _pending("the target external cylindrical-surface pattern is incomplete at this checkpoint")
if len(shells) != count:
return _fail({"expected_count": count, "actual_count": len(shells), "diameter_mm": expected["diameter_mm"]})
axial_span = expected.get("axial_span_mm")
evidence: dict[str, Any] = {
"external_cylindrical_faces": [item.get("record_id") for shell in shells for item in shell],
"external_cylindrical_surface_groups": [[item.get("record_id") for item in shell] for shell in shells],
"diameter_mm": expected["diameter_mm"],
}
if axial_span is None:
return _pass(evidence)
intervals = [_cylinder_axis_interval(shell[0]) for shell in shells]
if any(interval is None for interval in intervals):
return _pending("target external cylindrical faces have no measurable axial span")
actual_spans = []
for shell in shells:
shell_intervals = [_cylinder_axis_interval(record) for record in shell]
if any(interval is None for interval in shell_intervals):
return _pending("target external cylindrical faces have no measurable axial span")
actual_spans.append(
max(interval[1] for interval in shell_intervals if interval is not None)
- min(interval[0] for interval in shell_intervals if interval is not None)
)
tolerance = float(expected["tolerance_mm"])
evidence.update({"axial_spans_mm": actual_spans, "tolerance_mm": tolerance})
return _pass(evidence) if all(abs(span - float(axial_span)) <= tolerance for span in actual_spans) else _fail({"diameter_mm": expected["diameter_mm"], "expected_axial_span_mm": axial_span, "actual_axial_spans_mm": actual_spans, "tolerance_mm": tolerance})
def _hole_pattern(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
# The same topology evidence supports the count/diameter component; the
# runtime can add per-cylinder centres to prove PCD and angular spacing.
matched = _matching_cylinders(expected, facts)
count = int(expected["count"])
if not _records(facts):
return _pending("topology is unavailable")
if not matched:
return _pending("the target hole pattern has not been introduced at this checkpoint")
centers = [_cylinder_axis_point(item) for item in matched]
centers = [center for center in centers if center is not None]
if len(matched) < count:
return _pending("the target hole pattern is incomplete at this checkpoint")
if len(matched) != count or len(centers) != count:
return _fail({"expected_count": count, "actual_count": len(matched), "centred_count": len(centers), "diameter_mm": expected["diameter_mm"]})
radius = float(expected["pitch_radius_mm"])
tolerance = float(expected["tolerance_mm"])
selected = centers
centroid = tuple(sum(point[index] for point in selected) / count for index in range(3))
radial = [sqrt(sum((point[index] - centroid[index]) ** 2 for index in range(3))) for point in selected]
if not all(abs(value - radius) <= tolerance for value in radial):
return _fail({"expected_pitch_radius_mm": radius, "measured_radii_mm": radial, "tolerance_mm": tolerance})
central_bore_diameter = expected.get("concentric_bore_diameter_mm")
if central_bore_diameter is not None:
central = _matching_cylinders({"diameter_mm": central_bore_diameter, "tolerance_mm": tolerance}, facts)
central_axes = [_cylinder_axis_point(record) for record in central if bool((record.get("geometry") or {}).get("through"))]
central_axes = [axis for axis in central_axes if axis is not None]
if not central_axes:
return _pending("the required concentric reference bore is not available")
if len(central_axes) != 1:
return _fail({"reason": "concentric reference bore is ambiguous", "diameter_mm": central_bore_diameter, "count": len(central_axes)})
axis = central_axes[0]
normal = _cylinder_axis_direction(central[0])
if normal is None:
return _pending("the required concentric reference bore has no measurable axis")
delta = tuple(centroid[index] - axis[index] for index in range(3))
axial = sum(delta[index] * normal[index] for index in range(3))
concentric_error = sqrt(sum((delta[index] - axial * normal[index]) ** 2 for index in range(3)))
if concentric_error > tolerance:
return _fail({"expected_concentric_bore_diameter_mm": central_bore_diameter, "pattern_centroid_mm": centroid, "reference_axis_point_mm": axis, "concentric_error_mm": concentric_error, "tolerance_mm": tolerance})
angular = _equal_angular_spacing(selected, centroid, radius, tolerance)
if angular is None:
return _fail({"reason": "hole centres are degenerate or not evenly spaced", "centres_mm": selected, "tolerance_mm": tolerance})
evidence = {"count": count, "pitch_radius_mm": radius, "centres_mm": selected, "angular_spacing_rad": angular}
if central_bore_diameter is not None:
evidence["concentric_bore_diameter_mm"] = central_bore_diameter
return _pass(evidence)
def _equal_angular_spacing(
centres: list[tuple[float, float, float]],
centroid: tuple[float, float, float],
radius: float,
tolerance: float,
) -> list[float] | None:
"""Return cyclic angular increments only for a genuinely even circle."""
vectors = [tuple(point[index] - centroid[index] for index in range(3)) for point in centres]
base_length = sqrt(sum(value * value for value in vectors[0]))
if base_length == 0:
return None
normal: tuple[float, float, float] | None = None
for vector in vectors[1:]:
cross = (
vectors[0][1] * vector[2] - vectors[0][2] * vector[1],
vectors[0][2] * vector[0] - vectors[0][0] * vector[2],
vectors[0][0] * vector[1] - vectors[0][1] * vector[0],
)
length = sqrt(sum(value * value for value in cross))
if length > 1e-9:
normal = tuple(value / length for value in cross)
break
if normal is None:
# Two diametrically opposed holes are evenly spaced exactly when both
# are at the requested radius around their midpoint.
return [pi, pi] if len(centres) == 2 else None
basis_x = tuple(value / base_length for value in vectors[0])
basis_y = (
normal[1] * basis_x[2] - normal[2] * basis_x[1],
normal[2] * basis_x[0] - normal[0] * basis_x[2],
normal[0] * basis_x[1] - normal[1] * basis_x[0],
)
angles = sorted(
atan2(
sum(vector[index] * basis_y[index] for index in range(3)),
sum(vector[index] * basis_x[index] for index in range(3)),
)
for vector in vectors
)
increments = [
(angles[(index + 1) % len(angles)] - angles[index]) % (2 * pi)
for index in range(len(angles))
]
target = 2 * pi / len(centres)
return increments if all(abs(increment - target) * radius <= tolerance for increment in increments) else None
def _collinear_bore_chain(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
"""Prove a coplanar bore-centre chain in either direction.
``axis_origin_mm`` may be either end of a through bore, depending on the
host-face orientation used to create that feature. Project each axis line
onto the common plane normal to the bore direction before comparing its
centre location; otherwise equivalent top- and bottom-hosted holes appear
to be offset by the part thickness.
"""
matched = [
record for record in _matching_cylinders(expected, facts)
if bool((record.get("geometry") or {}).get("through"))
]
expected_distances = [float(value) for value in expected["adjacent_distances_mm"]]
count = len(expected_distances) + 1
tolerance = float(expected["tolerance_mm"])
centres = [_cylinder_axis_point(record) for record in matched]
centres = [centre for centre in centres if centre is not None]
if not _records(facts):
return _pending("topology is unavailable")
if not matched:
return _pending("the target bore chain has not been introduced at this checkpoint")
# A checkpoint may deliberately add this chain across more than one
# action. Fewer already-through matching bores are therefore unfinished
# future work, while a matching bore that is not through is a real
# violation of this claim.
if len(centres) < count and len(centres) == len(matched):
return _pending("the target bore chain is incomplete at this checkpoint")
if len(centres) != count:
return _fail({"expected_count": count, "actual_count": len(centres), "diameter_mm": expected["diameter_mm"]})
directions = [_cylinder_axis_direction(record) for record in matched]
if all(direction is not None for direction in directions):
reference_direction = directions[0]
assert reference_direction is not None
alignments = [
abs(sum(reference_direction[index] * direction[index] for index in range(3)))
for direction in directions
if direction is not None
]
if any(alignment < 1.0 - 1e-6 for alignment in alignments):
return _fail({"reason": "bore axes are not parallel", "axis_alignments": alignments})
# This is the unique point where each parallel bore axis intersects
# the plane through the world origin normal to ``reference_direction``.
# It is invariant to choosing either axial endpoint as axis_origin_mm.
centres = [
tuple(
centre[index]
- sum(centre[component] * reference_direction[component] for component in range(3))
* reference_direction[index]
for index in range(3)
)
for centre in centres
]
origin = centres[0]
endpoint = max(centres[1:], key=lambda centre: sum((centre[index] - origin[index]) ** 2 for index in range(3)))
direction = tuple(endpoint[index] - origin[index] for index in range(3))
direction_length = sqrt(sum(value * value for value in direction))
if direction_length <= tolerance:
return _fail({"reason": "bore centres do not define a non-zero chain axis"})
unit = tuple(value / direction_length for value in direction)
projections: list[tuple[float, tuple[float, float, float]]] = []
for centre in centres:
relative = tuple(centre[index] - origin[index] for index in range(3))
projection = sum(relative[index] * unit[index] for index in range(3))
residual = sqrt(sum((relative[index] - projection * unit[index]) ** 2 for index in range(3)))
if residual > tolerance:
return _fail({"reason": "bore centres are not collinear", "residual_mm": residual, "tolerance_mm": tolerance})
projections.append((projection, centre))
projections.sort(key=lambda item: item[0])
actual_distances = [projections[index + 1][0] - projections[index][0] for index in range(count - 1)]
matches_forward = all(
abs(actual - target) <= tolerance
for actual, target in zip(actual_distances, expected_distances, strict=True)
)
matches_reverse = all(
abs(actual - target) <= tolerance
for actual, target in zip(actual_distances, reversed(expected_distances), strict=True)
)
if not matches_forward and not matches_reverse:
return _fail({"expected_adjacent_distances_mm": expected_distances, "actual_adjacent_distances_mm": actual_distances, "tolerance_mm": tolerance})
return _pass({
"centres_mm": [centre for _projection, centre in projections],
"adjacent_distances_mm": actual_distances,
"expected_orientation": "forward" if matches_forward else "reversed",
"tolerance_mm": tolerance,
})
def _coaxial_through_bore_group(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
matched = _matching_cylinders(expected, facts)
count = int(expected["count"])
if not _records(facts):
return _pending("topology is unavailable")
if not matched:
return _pending("the target bore has not been introduced at this checkpoint")
through = [record for record in matched if bool((record.get("geometry") or {}).get("through"))]
if len(matched) < count and len(through) == len(matched):
return _pending("the target coaxial bore group is incomplete at this checkpoint")
if len(matched) != count or len(through) != count:
return _fail({"expected_count": count, "actual_count": len(matched), "actual_through_count": len(through), "diameter_mm": expected["diameter_mm"]})
points = [_cylinder_axis_point(record) for record in through]
directions = [_cylinder_axis_direction(record) for record in through]
if any(point is None for point in points) or any(direction is None for direction in directions):
return _pending("target bore axes are unavailable")
point_values = [point for point in points if point is not None]
direction_values = [direction for direction in directions if direction is not None]
tolerance = float(expected["tolerance_mm"])
reference_point, reference_direction = point_values[0], direction_values[0]
deviations: list[dict[str, Any]] = []
for record, point, direction in zip(through[1:], point_values[1:], direction_values[1:], strict=True):
dot = abs(sum(reference_direction[index] * direction[index] for index in range(3)))
offset = tuple(point[index] - reference_point[index] for index in range(3))
distance = sqrt(sum((offset[index] - sum(offset[component] * reference_direction[component] for component in range(3)) * reference_direction[index]) ** 2 for index in range(3)))
deviations.append({"record_id": record.get("record_id"), "axis_dot": dot, "axis_distance_mm": distance})
return _pass({"diameter_mm": expected["diameter_mm"], "axes": deviations, "tolerance_mm": tolerance}) if all(item["axis_dot"] >= 1 - 1e-6 and item["axis_distance_mm"] <= tolerance for item in deviations) else _fail({"diameter_mm": expected["diameter_mm"], "axes": deviations, "tolerance_mm": tolerance})
def _orthogonal_intersecting_through_bores(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
first = _matching_cylinders({"diameter_mm": expected["first_diameter_mm"], "tolerance_mm": expected["tolerance_mm"]}, facts)
second = _matching_cylinders({"diameter_mm": expected["second_diameter_mm"], "tolerance_mm": expected["tolerance_mm"]}, facts)
if not _records(facts):
return _pending("topology is unavailable")
if not first or not second:
return _pending("one or both target bores have not been introduced at this checkpoint")
first_through = [record for record in first if bool((record.get("geometry") or {}).get("through"))]
second_through = [record for record in second if bool((record.get("geometry") or {}).get("through"))]
if len(first) != 1 or len(second) != 1 or len(first_through) != 1 or len(second_through) != 1:
return _fail({"first_diameter_mm": expected["first_diameter_mm"], "first_count": len(first), "first_through_count": len(first_through), "second_diameter_mm": expected["second_diameter_mm"], "second_count": len(second), "second_through_count": len(second_through)})
first_point, second_point = _cylinder_axis_point(first_through[0]), _cylinder_axis_point(second_through[0])
first_axis, second_axis = _cylinder_axis_direction(first_through[0]), _cylinder_axis_direction(second_through[0])
if first_point is None or second_point is None or first_axis is None or second_axis is None:
return _pending("target bore axes are unavailable")
expected_first_axis = {"x": (1.0, 0.0, 0.0), "y": (0.0, 1.0, 0.0), "z": (0.0, 0.0, 1.0)}[expected["first_axis"]]
expected_second_axis = {"x": (1.0, 0.0, 0.0), "y": (0.0, 1.0, 0.0), "z": (0.0, 0.0, 1.0)}[expected["second_axis"]]
first_axis_alignment = abs(sum(first_axis[index] * expected_first_axis[index] for index in range(3)))
second_axis_alignment = abs(sum(second_axis[index] * expected_second_axis[index] for index in range(3)))
dot = sum(first_axis[index] * second_axis[index] for index in range(3))
denominator = 1 - dot * dot
if denominator <= 1e-9:
return _fail({"reason": "bore axes are parallel", "axis_dot": dot})
offset = tuple(first_point[index] - second_point[index] for index in range(3))
first_parameter = (dot * sum(second_axis[index] * offset[index] for index in range(3)) - sum(first_axis[index] * offset[index] for index in range(3))) / denominator
second_parameter = (sum(second_axis[index] * offset[index] for index in range(3)) - dot * sum(first_axis[index] * offset[index] for index in range(3))) / denominator
closest_first = tuple(first_point[index] + first_parameter * first_axis[index] for index in range(3))
closest_second = tuple(second_point[index] + second_parameter * second_axis[index] for index in range(3))
separation = sqrt(sum((closest_first[index] - closest_second[index]) ** 2 for index in range(3)))
tolerance = float(expected["tolerance_mm"])
evidence = {
"axis_dot": dot,
"axis_intersection_separation_mm": separation,
"first_axis": expected["first_axis"],
"first_axis_alignment": first_axis_alignment,
"second_axis": expected["second_axis"],
"second_axis_alignment": second_axis_alignment,
"tolerance_mm": tolerance,
}
return _pass(evidence) if abs(dot) <= 1e-6 and separation <= tolerance and first_axis_alignment >= 1 - 1e-6 and second_axis_alignment >= 1 - 1e-6 else _fail(evidence)
def _rectangular_corner_bore_pattern(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
"""Prove four through bores lie at one equal offset from a rectangular plate's edges."""
matched = _matching_cylinders(expected, facts)
count = int(expected["count"])
if not _records(facts):
return _pending("topology is unavailable")
if not matched:
return _pending("the target bore has not been introduced at this checkpoint")
through = [record for record in matched if bool((record.get("geometry") or {}).get("through"))]
if len(matched) < count and len(through) == len(matched):
return _pending("the target corner-bore pattern is incomplete at this checkpoint")
if len(matched) != count or len(through) != count:
return _fail({"expected_count": count, "actual_count": len(matched), "actual_through_count": len(through), "diameter_mm": expected["diameter_mm"]})
health = facts.get("health") if isinstance(facts.get("health"), dict) else {}
bbox = health.get("bbox_mm") if isinstance(health.get("bbox_mm"), dict) else {}
minimum, maximum = bbox.get("min"), bbox.get("max")
if not isinstance(minimum, list) or not isinstance(maximum, list) or len(minimum) != 3 or len(maximum) != 3:
return _pending("rebuild report has no bounding-box extrema")
normal = _cylinder_axis_direction(through[0])
centres = [_cylinder_axis_point(record) for record in through]
if normal is None or any(centre is None for centre in centres):
return _pending("target bore axes are unavailable")
normal_axis = max(range(3), key=lambda index: abs(normal[index]))
if abs(normal[normal_axis]) < 0.99:
return _pending("corner-bore pattern is not aligned to a measurable cardinal plate normal")
plane_axes = [index for index in range(3) if index != normal_axis]
try:
lower = [float(value) for value in minimum]
upper = [float(value) for value in maximum]
except (TypeError, ValueError):
return _pending("rebuild report has invalid bounding-box extrema")
offset = float(expected["edge_offset_mm"])
tolerance = float(expected["tolerance_mm"])
centre_values = [centre for centre in centres if centre is not None]
positions: list[tuple[int, int]] = []
for centre in centre_values:
position: list[int] = []
for axis in plane_axes:
low_error = abs((centre[axis] - lower[axis]) - offset)
high_error = abs((upper[axis] - centre[axis]) - offset)
if low_error <= tolerance:
position.append(0)
elif high_error <= tolerance:
position.append(1)
else:
return _fail({"reason": "bore centre has the wrong edge offset", "centre_mm": centre, "axis": ["x", "y", "z"][axis], "low_offset_mm": centre[axis] - lower[axis], "high_offset_mm": upper[axis] - centre[axis], "expected_edge_offset_mm": offset, "tolerance_mm": tolerance})
positions.append((position[0], position[1]))
expected_positions = {(0, 0), (0, 1), (1, 0), (1, 1)}
return _pass({"diameter_mm": expected["diameter_mm"], "edge_offset_mm": offset, "normal_axis": ["x", "y", "z"][normal_axis], "corner_positions": positions, "tolerance_mm": tolerance}) if set(positions) == expected_positions and len(set(positions)) == count else _fail({"reason": "bores do not occupy all rectangular corners", "corner_positions": positions, "expected_positions": sorted(expected_positions)})
def _coaxial(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
ids = set(expected["record_ids"])
matching = [record for record in _records(facts) if str(record.get("record_id") or "") in ids]
if len(matching) != 2:
return _pending("referenced topology records are not both available")
axes = [_vector((record.get("geometry") or {}).get("axis_direction")) for record in matching]
if not all(axes):
return _pending("referenced records have no measurable axes")
a, b = axes # type: ignore[misc]
dot = abs(sum(a[index] * b[index] for index in range(3)))
return _pass({"axis_dot": dot}) if isclose(dot, 1.0, abs_tol=float(expected["tolerance"])) else _fail({"axis_dot": dot, "tolerance": expected["tolerance"]})
def _coplanar(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
ids = set(expected["record_ids"])
matching = [record for record in _records(facts) if str(record.get("record_id") or "") in ids]
if len(matching) != 2:
return _pending("referenced topology records are not both available")
geometry = [(record.get("geometry") or {}) for record in matching]
normals = [_vector(value.get("plane_normal") or value.get("normal")) for value in geometry]
offsets = [value.get("plane_offset_mm") for value in geometry]
if not all(normals) or not all(isinstance(value, (int, float)) for value in offsets):
return _pending("referenced records have no measurable planes")
dot = abs(sum(normals[0][index] * normals[1][index] for index in range(3))) # type: ignore[index]
offset = abs(float(offsets[0]) - float(offsets[1]))
tolerance = float(expected["tolerance_mm"])
return _pass({"normal_dot": dot, "offset_delta_mm": offset}) if isclose(dot, 1.0, abs_tol=1e-5) and offset <= tolerance else _fail({"normal_dot": dot, "offset_delta_mm": offset, "tolerance_mm": tolerance})
@dataclass(frozen=True, slots=True)
class ClaimDefinition:
claim_kind: str
expected_schema: dict[str, Any]
required_artifacts: tuple[str, ...]
evaluator: ClaimEvaluator
deterministic: bool = True
class VerifierRegistry:
def __init__(self, definitions: tuple[ClaimDefinition, ...]) -> None:
self._definitions = {item.claim_kind: item for item in definitions}
if len(self._definitions) != len(definitions):
raise ValueError("Verifier claim_kind values must be unique")
for definition in definitions:
if not _is_closed_schema(definition.expected_schema):
raise ValueError(f"Verifier schema is not closed: {definition.claim_kind}")
try:
Draft202012Validator.check_schema(definition.expected_schema)
except SchemaError as error:
raise ValueError(f"Invalid verifier schema: {definition.claim_kind}") from error
@property
def claim_kinds(self) -> tuple[str, ...]:
return tuple(sorted(self._definitions))
def definition(self, claim_kind: str) -> ClaimDefinition:
try:
return self._definitions[claim_kind]
except KeyError as error:
raise ValueError(f"VERIFIER_UNAVAILABLE: {claim_kind}") from error
def expected_one_of_schema(self) -> dict[str, Any]:
return {
"oneOf": [
_closed_object({"claim_kind": {"const": item.claim_kind}, "expected": item.expected_schema}, ["claim_kind", "expected"])
for item in self._definitions.values()
]
}
def validate_expected(self, claim_kind: str, expected: dict[str, Any]) -> list[dict[str, str]]:
definition = self.definition(claim_kind)
validator = Draft202012Validator(definition.expected_schema)
return [
{"path": "/" + "/".join(str(part) for part in error.absolute_path), "message": error.message}
for error in sorted(validator.iter_errors(expected), key=lambda item: (list(item.absolute_path), item.message))
]
def evaluate(self, claim_kind: str, expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult:
errors = self.validate_expected(claim_kind, expected)
if errors:
return {"status": "unavailable", "evidence": {"schema_errors": errors}}
return self.definition(claim_kind).evaluator(expected, facts)
def _is_closed_schema(schema: Any) -> bool:
if not isinstance(schema, dict):
return False
if schema.get("type") == "object" and schema.get("additionalProperties") is not False:
return False
for key in ("properties", "$defs", "definitions"):
values = schema.get(key)
if isinstance(values, dict) and not all(_is_closed_schema(value) for value in values.values()):
return False
for key in ("items", "additionalItems"):
if key in schema and isinstance(schema[key], dict) and not _is_closed_schema(schema[key]):
return False
for key in ("oneOf", "anyOf", "allOf"):
if key in schema and (not isinstance(schema[key], list) or not all(_is_closed_schema(value) for value in schema[key])):
return False
return True
def default_registry() -> VerifierRegistry:
# Acceptance tolerances are author-selected but never author-unbounded.
# A broad value could otherwise turn an exact dimensions claim into a
# vacuous pass. A tenth of a millimetre is a deliberately generous upper bound for
# this CAD protocol; requests needing looser acceptance must be made
# explicit through a separate verifier rather than weakening all claims.
positive = {"type": "number", "exclusiveMinimum": 0, "maximum": 100_000}
tolerance = {"type": "number", "minimum": 0, "maximum": 0.1}
alignment_tolerance = {"type": "number", "minimum": 0, "maximum": 0.01}
cylindrical = _closed_object({"diameter_mm": positive, "count": {"type": "integer", "minimum": 1}, "tolerance_mm": tolerance}, ["diameter_mm"])
outer_cylindrical = _closed_object({"diameter_mm": positive, "count": {"type": "integer", "minimum": 1}, "axial_span_mm": positive, "tolerance_mm": tolerance}, ["diameter_mm"])
bore_depth = _closed_object({"diameter_mm": positive, "depth_mm": positive, "count": {"type": "integer", "minimum": 1}, "tolerance_mm": tolerance}, ["diameter_mm", "depth_mm", "tolerance_mm"])
conical_bore = _closed_object({"small_diameter_mm": positive, "large_diameter_mm": positive, "included_angle_deg": {"type": "number", "exclusiveMinimum": 0, "maximum": 179.999}, "count": {"type": "integer", "minimum": 1}, "tolerance_mm": tolerance}, ["small_diameter_mm", "large_diameter_mm", "included_angle_deg", "tolerance_mm"])
bore_chain = _closed_object({
"diameter_mm": positive,
"adjacent_distances_mm": {"type": "array", "items": positive, "minItems": 1, "maxItems": 63},
"tolerance_mm": tolerance,
}, ["diameter_mm", "adjacent_distances_mm", "tolerance_mm"])
return VerifierRegistry((
ClaimDefinition("solid_count_equals", _closed_object({"value": {"type": "integer", "minimum": 1}}, ["value"]), ("rebuild_report",), _solid_count),
ClaimDefinition("volume_decreased", _closed_object({}, []), ("rebuild_report", "parent_rebuild_report"), _volume_decreased),
ClaimDefinition("single_connected_body", _closed_object({}, []), ("rebuild_report", "topology"), lambda _expected, facts: _solid_count({"value": 1}, facts)),
ClaimDefinition("bbox_dimension_mm", _closed_object({"axis": {"enum": ["x", "y", "z"]}, "value": positive, "tolerance_mm": tolerance}, ["axis", "value", "tolerance_mm"]), ("rebuild_report",), _bbox),
ClaimDefinition("bbox_rank_dimension_mm", _closed_object({"rank": {"enum": ["minimum", "median", "maximum"]}, "value": positive, "tolerance_mm": tolerance}, ["rank", "value", "tolerance_mm"]), ("rebuild_report",), _bbox_rank),
ClaimDefinition("cylindrical_bore", cylindrical, ("topology",), _cylindrical_bore),
ClaimDefinition("through_cylindrical_bore", cylindrical, ("topology",), _through_cylindrical_bore),
ClaimDefinition("cylindrical_bore_depth", bore_depth, ("topology",), _cylindrical_bore_depth),
ClaimDefinition("conical_bore", conical_bore, ("topology",), _conical_bore),
ClaimDefinition("outer_cylindrical_surface", outer_cylindrical, ("topology",), _outer_cylindrical_surface),
ClaimDefinition("circular_hole_pattern", _closed_object({"count": {"type": "integer", "minimum": 2}, "diameter_mm": positive, "pitch_radius_mm": positive, "concentric_bore_diameter_mm": positive, "tolerance_mm": tolerance}, ["count", "diameter_mm", "pitch_radius_mm", "tolerance_mm"]), ("topology",), _hole_pattern),
ClaimDefinition("collinear_through_bore_chain", bore_chain, ("topology",), _collinear_bore_chain),
ClaimDefinition("coaxial_through_bore_group", _closed_object({"diameter_mm": positive, "count": {"type": "integer", "minimum": 2, "maximum": 16}, "tolerance_mm": alignment_tolerance}, ["diameter_mm", "count", "tolerance_mm"]), ("topology",), _coaxial_through_bore_group),
ClaimDefinition("orthogonal_intersecting_through_bores", _closed_object({"first_diameter_mm": positive, "second_diameter_mm": positive, "first_axis": {"enum": ["x", "y", "z"]}, "second_axis": {"enum": ["x", "y", "z"]}, "tolerance_mm": alignment_tolerance}, ["first_diameter_mm", "second_diameter_mm", "first_axis", "second_axis", "tolerance_mm"]), ("topology",), _orthogonal_intersecting_through_bores),
ClaimDefinition("rectangular_corner_through_bore_pattern", _closed_object({"diameter_mm": positive, "count": {"const": 4}, "edge_offset_mm": positive, "tolerance_mm": tolerance}, ["diameter_mm", "count", "edge_offset_mm", "tolerance_mm"]), ("rebuild_report", "topology"), _rectangular_corner_bore_pattern),
ClaimDefinition("coaxial", _closed_object({"record_ids": {"type": "array", "items": {"type": "string", "minLength": 1}, "minItems": 2, "maxItems": 2, "uniqueItems": True}, "tolerance": alignment_tolerance}, ["record_ids", "tolerance"]), ("topology",), _coaxial),
ClaimDefinition("coplanar", _closed_object({"record_ids": {"type": "array", "items": {"type": "string", "minLength": 1}, "minItems": 2, "maxItems": 2, "uniqueItems": True}, "tolerance_mm": tolerance}, ["record_ids", "tolerance_mm"]), ("topology",), _coplanar),
ClaimDefinition("visual", _closed_object({"description": {"type": "string", "minLength": 1, "maxLength": 360}}, ["description"]), ("render",), lambda _expected, _facts: _pending("requires independent visual review"), deterministic=False),
))