"""Deterministic acceptance-claim registry and closed-schema validation.""" from __future__ import annotations from dataclasses import dataclass from copy import deepcopy 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] _DEFAULT_TOLERANCE_MM = 0.1 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. B-rep face orientation is not a physical property of a cylindrical shell. In particular, a two-sided extrusion can return its two half-walls with opposite axis directions. Use one canonical direction for an undirected cylinder axis so those halves share the same coordinate interval. """ 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 dominant_index = max(range(3), key=lambda index: abs(direction[index])) if direction[dominant_index] < 0: direction = tuple(-component for component in direction) 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.get("tolerance_mm", _DEFAULT_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 _concentric_bore_to_outer_cylinder(expected: dict[str, Any], facts: dict[str, Any]) -> ClaimResult: """Verify one bore axis coincides with one external cylindrical shell. This is distinct from ``coaxial_through_bore_group``, which compares two or more inner bores. A centered flange bore has only one inner cylinder, so treating its outer wall as a second bore makes the old verifier permanently pending. """ if not _records(facts): return _pending("topology is unavailable") tolerance = float(expected["tolerance_mm"]) bore = _matching_cylinders({ "diameter_mm": expected["bore_diameter_mm"], "tolerance_mm": tolerance, }, facts) outer_shells = _outer_cylinder_shells({ "diameter_mm": expected["outer_diameter_mm"], "tolerance_mm": tolerance, }, facts) if not bore: return _pending("the target bore has not been introduced at this checkpoint") if not outer_shells: return _pending("the target outer cylindrical surface has not been introduced at this checkpoint") if len(bore) != 1 or len(outer_shells) != 1: return _fail({ "expected_bore_count": 1, "actual_bore_count": len(bore), "expected_outer_count": 1, "actual_outer_count": len(outer_shells), "bore_diameter_mm": expected["bore_diameter_mm"], "outer_diameter_mm": expected["outer_diameter_mm"], }) bore_point = _cylinder_axis_point(bore[0]) bore_axis = _cylinder_axis_direction(bore[0]) outer_point = _cylinder_axis_point(outer_shells[0][0]) outer_axis = _cylinder_axis_direction(outer_shells[0][0]) if None in {bore_point, bore_axis, outer_point, outer_axis}: return _pending("the bore or outer-cylinder axis is not measurable") assert bore_point is not None and bore_axis is not None assert outer_point is not None and outer_axis is not None axis_dot = abs(sum(bore_axis[index] * outer_axis[index] for index in range(3))) delta = tuple(bore_point[index] - outer_point[index] for index in range(3)) axial = sum(delta[index] * outer_axis[index] for index in range(3)) radial_distance = sqrt(sum((delta[index] - axial * outer_axis[index]) ** 2 for index in range(3))) evidence = { "bore_record_id": bore[0].get("record_id"), "outer_record_ids": [record.get("record_id") for record in outer_shells[0]], "axis_dot": axis_dot, "axis_distance_mm": radial_distance, "tolerance_mm": tolerance, } return _pass(evidence) if axis_dot >= 1 - 1e-6 and radial_distance <= tolerance else _fail(evidence) 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, *, exclude_claim_kinds: set[str] | frozenset[str] | tuple[str, ...] = ()) -> dict[str, Any]: excluded = set(exclude_claim_kinds) return { "oneOf": [ _closed_object({"claim_kind": {"const": item.claim_kind}, "expected": item.expected_schema}, ["claim_kind", "expected"]) for item in self._definitions.values() if item.claim_kind not in excluded ] } 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 normalize_expected(self, claim_kind: str, expected: dict[str, Any]) -> dict[str, Any]: """Apply protocol defaults before a compiled contract is frozen. These defaults describe verifier mechanics, never user geometry. The outer-cylinder verifier can match a diameter without a tolerance, but measuring its optional axial span needs one. Persist the default so the resulting contract is complete and independently reproducible. """ normalized = deepcopy(expected) if claim_kind == "outer_cylindrical_surface" and "axial_span_mm" in normalized: normalized.setdefault("tolerance_mm", _DEFAULT_TOLERANCE_MM) return normalized 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("concentric_bore_to_outer_cylinder", _closed_object({"bore_diameter_mm": positive, "outer_diameter_mm": positive, "tolerance_mm": alignment_tolerance}, ["bore_diameter_mm", "outer_diameter_mm", "tolerance_mm"]), ("topology",), _concentric_bore_to_outer_cylinder), 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), ))