diff --git a/docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md b/docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md index 006325b..013fb44 100644 --- a/docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md +++ b/docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md @@ -142,6 +142,18 @@ real authority. with zero bbox-centre collateral; the generic geometry point mask is rejected because it would erase accepted object evidence. Navigation, safety and command authority remain false. +- [x] Add ADR 0024 and the executable A5 contracts: + `missioncore.track-geometry/v1`, + `missioncore.track-geometry-frame/v1`, + `missioncore.track-geometry-binding/v1` and + `missioncore.point-slab/v1`. +- [x] Enforce camera-owned semantic identity, lossless unique frame-local + source indices, one point owner, explicit current/held/persistent and + current/held/persistent metric bases, retained reason codes, no fake + range/class, no missing-point free-space inference and no runtime authority. +- [x] Verify strict round-trip plus adversarial duplicate ownership, unknown + owner, camera-only/conflict range, geometry-only semantics, held/persistent + ownership, unavailable source and authority mutation cases. The frozen A3 engineering and human generations cover all 486 items and retains the exact A2 materialization/review-pack and 42-sheet evidence @@ -154,13 +166,14 @@ geometry and 21 visible missed class-bearing objects. Earlier generations and drafts remain historical and are not rewritten or presented as the current product workflow. -A3 and A4 are complete. E31 accepts only the recorded RAVNOVES00 diagnostic -binding and does not claim that host arrival is hardware firing time. The -factory KB4 identity is exact, but a measured calibration-target residual and -physical body/mount dimensions are unavailable. The accepted profile is -therefore source-session scoped and cannot transfer to another mount. A5 -`TrackGeometry v1` is the next critical-path implementation; E32 publication -must bind both that contract and the accepted E31 profile. +A3, A4 and the A5 contract gate are complete. E31 accepts only the recorded +RAVNOVES00 diagnostic binding and does not claim that host arrival is hardware +firing time. The factory KB4 identity is exact, but a measured +calibration-target residual and physical body/mount dimensions are +unavailable. The accepted profile is therefore source-session scoped and +cannot transfer to another mount. A6/E32 full replay is the next critical-path +implementation and must bind every published frame to both `TrackGeometry v1` +and the accepted E31 profile. ### A3 residual and human-exception policy diff --git a/docs/adr/0024-track-geometry-and-point-ownership.md b/docs/adr/0024-track-geometry-and-point-ownership.md new file mode 100644 index 0000000..db64091 --- /dev/null +++ b/docs/adr/0024-track-geometry-and-point-ownership.md @@ -0,0 +1,85 @@ +# ADR 0024: own camera semantics and frame-local geometry explicitly + +Date: 2026-07-27 +Status: accepted + +## Context + +E29 proved a camera-first diagnostic fusion shape, E30 froze its engineering +audit, and E31 accepted one exact source-time/calibration/self-mask profile. +Those results still publish experiment-specific dictionaries. Reusing them +directly in E32 would leave several architectural ambiguities: + +- whether a semantic class came from camera evidence or geometry; +- whether two tracks can claim the same LiDAR source point; +- whether a range is current, held from an earlier frame or supplied by a + persistent model; +- whether camera-only, conflict and unknown states may silently acquire metric + geometry; +- whether missing endpoints can be treated as free space. + +A universal runtime ontology is not justified. The required boundary is local +to the perception product and can remain an executable versioned contract. + +## Decision + +1. Mission Core owns `missioncore.track-geometry/v1` as the provider-neutral + object geometry contract. +2. `TrackGeometrySourceBinding` pins every frame to the exact source pack, + session, representation capability profile, E31 qualification result, + calibration identity, coordinate frame and time basis. +3. Camera tracks alone carry semantic track ID, label and image bbox. + Geometry-only products carry no semantic class or camera track ID. +4. Evidence state is explicit and closed: + `agree`, `single-source-camera`, `conflict`, `unknown` or + `single-source-geometry`. +5. Currentness is independent and closed: `current`, `held` or `persistent`. + Held values reference an earlier frame and persistent values reference a + separate model. Neither owns current source points. +6. Metric basis is explicit: `current-points`, `held-last-current`, + `persistent-model` or `unavailable`. +7. Camera-only, conflict and current unknown states cannot publish a range. + Current `agree` and current geometry-only products require accepted source + points and a positive range. +8. `missioncore.point-slab/v1` is the sole container for accepted current + metric points. Each row retains its lossless frame-local source index, + map-frame XYZ and exactly one owner index. +9. Source indices in a slab are unique and bounded by the source-frame point + count. This makes double ownership unrepresentable. +10. Candidate, rejected, held and persistent points are not inserted into the + current accepted slab. Their reason/provenance remains on the geometry + product or a separately bound model. +11. Missing source evidence remains unknown. The contract fixes + `absence_of_points_means_free=false`. +12. TrackGeometry has no command, navigation or safety authority. + +## Consequences + +- E32 can translate E29/E31 evidence into one stable frame contract without + changing camera/geometry ownership. +- A duplicate source-point claim, fake range, invented class or current/held + conflation fails at construction and deserialization. +- The contract supports future native scans and simulation providers through + their own representation and E31-equivalent source binding; it does not + upgrade historical K1 evidence. +- PointSlab currently defines the semantic array shape and strict JSON + round-trip. E32 may add a digest-bound binary artifact encoding without + changing ownership semantics. +- Planner-facing occupancy remains a later A8 product with its own admission; + TrackGeometry does not imply free space or traversability. + +## Implementation + +- `src/k1link/compute/track_geometry.py` +- `src/k1link/compute/sensor_representation.py` +- `tests/test_track_geometry.py` +- `tests/test_sensor_representation.py` +- `docs/16_ARCHITECTURE_AUDIT_EXECUTION_ROADMAP.md` + +## References + +- ADR 0018: LiDAR evidence before models +- ADR 0021: dataset gateway representation boundary +- ADR 0023: sensor representation capabilities and free-space admission +- LAB E29: camera-first semantics with independent local-surface geometry +- LAB E31: source-time, calibration and self-mask qualification diff --git a/src/k1link/compute/__init__.py b/src/k1link/compute/__init__.py index c31c16f..73d9ac1 100644 --- a/src/k1link/compute/__init__.py +++ b/src/k1link/compute/__init__.py @@ -202,6 +202,21 @@ from .results import ( RecordedPerceptionResult, validate_recorded_perception_result, ) +from .track_geometry import ( + POINT_SLAB_SCHEMA, + TRACK_GEOMETRY_BINDING_SCHEMA, + TRACK_GEOMETRY_FRAME_SCHEMA, + TRACK_GEOMETRY_SCHEMA, + PointSlab, + TrackGeometry, + TrackGeometryContractError, + TrackGeometryCurrentness, + TrackGeometryEvidenceState, + TrackGeometryFrame, + TrackGeometryMetricBasis, + TrackGeometryOwnerKind, + TrackGeometrySourceBinding, +) from .tracked_fusion_qualification import ( TrackedFusionQualificationResult, validate_tracked_fusion_qualification_result, @@ -302,6 +317,7 @@ __all__ = [ "PATCHWORKPP_SOURCE_COMMIT", "PATCHWORKPP_SOURCE_TAG", "PATCHWORKPP_SOURCE_URL", + "POINT_SLAB_SCHEMA", "PatchworkPPGroundSegmenter", "QueueSnapshot", "RecordedCalibratedFusion", @@ -363,6 +379,18 @@ __all__ = [ "TrackingQualificationResult", "TrackedFusionQualificationResult", "TELEMETRY_SCHEMA", + "TRACK_GEOMETRY_BINDING_SCHEMA", + "TRACK_GEOMETRY_FRAME_SCHEMA", + "TRACK_GEOMETRY_SCHEMA", + "PointSlab", + "TrackGeometry", + "TrackGeometryContractError", + "TrackGeometryCurrentness", + "TrackGeometryEvidenceState", + "TrackGeometryFrame", + "TrackGeometryMetricBasis", + "TrackGeometryOwnerKind", + "TrackGeometrySourceBinding", "WORLD_STATE_SCHEMA", "WorldStateProjector", "classify_health", diff --git a/src/k1link/compute/track_geometry.py b/src/k1link/compute/track_geometry.py new file mode 100644 index 0000000..cb41221 --- /dev/null +++ b/src/k1link/compute/track_geometry.py @@ -0,0 +1,901 @@ +"""Provider-neutral TrackGeometry v1 and frame-local PointSlab contract. + +The contract preserves camera-owned semantic identity, lossless source-point +ownership and explicit current/held/persistent separation. It never infers a +semantic class, metric range, free space or runtime authority from missing +evidence. +""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass +from enum import StrEnum +from typing import Final, cast + +import numpy as np +import numpy.typing as npt + +TRACK_GEOMETRY_SCHEMA: Final = "missioncore.track-geometry/v1" +TRACK_GEOMETRY_FRAME_SCHEMA: Final = "missioncore.track-geometry-frame/v1" +TRACK_GEOMETRY_BINDING_SCHEMA: Final = "missioncore.track-geometry-binding/v1" +POINT_SLAB_SCHEMA: Final = "missioncore.point-slab/v1" + +_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$") +_SOURCE_PACK_ID = re.compile(r"^e10-lidar-pack-[a-f0-9]{64}$") +_E31_RESULT_ID = re.compile(r"^e31-source-qualification-[a-f0-9]{64}$") +_SHA256 = re.compile(r"^[a-f0-9]{64}$") + +Float32Array = npt.NDArray[np.float32] +Int64Array = npt.NDArray[np.int64] +UInt32Array = npt.NDArray[np.uint32] + + +class TrackGeometryContractError(ValueError): + """A TrackGeometry v1 value violates source or ownership semantics.""" + + +class TrackGeometryOwnerKind(StrEnum): + CAMERA_TRACK = "camera-track" + GEOMETRY_CLUSTER = "geometry-cluster" + + +class TrackGeometryEvidenceState(StrEnum): + AGREE = "agree" + CAMERA_ONLY = "single-source-camera" + CONFLICT = "conflict" + UNKNOWN = "unknown" + GEOMETRY_ONLY = "single-source-geometry" + + +class TrackGeometryCurrentness(StrEnum): + CURRENT = "current" + HELD = "held" + PERSISTENT = "persistent" + + +class TrackGeometryMetricBasis(StrEnum): + CURRENT_POINTS = "current-points" + HELD_LAST_CURRENT = "held-last-current" + PERSISTENT_MODEL = "persistent-model" + UNAVAILABLE = "unavailable" + + +@dataclass(frozen=True, slots=True) +class TrackGeometrySourceBinding: + """Bind one frame contract to exact source, representation and E31 truth.""" + + source_pack_id: str + source_session_id: str + representation_profile_id: str + e31_qualification_id: str + calibration_sha256: str + coordinate_frame: str + time_basis: str + selected_offset_ms: int + + def __post_init__(self) -> None: + if _SOURCE_PACK_ID.fullmatch(self.source_pack_id) is None: + raise TrackGeometryContractError("source pack id is invalid") + _safe_identifier(self.source_session_id, "source session id") + _safe_identifier( + self.representation_profile_id, + "representation profile id", + ) + if _E31_RESULT_ID.fullmatch(self.e31_qualification_id) is None: + raise TrackGeometryContractError("E31 qualification id is invalid") + if _SHA256.fullmatch(self.calibration_sha256) is None: + raise TrackGeometryContractError("calibration identity is invalid") + _safe_identifier(self.coordinate_frame, "coordinate frame") + _safe_identifier(self.time_basis, "time basis") + if ( + not isinstance(self.selected_offset_ms, int) + or isinstance(self.selected_offset_ms, bool) + or abs(self.selected_offset_ms) > 1000 + ): + raise TrackGeometryContractError("selected offset is invalid") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": TRACK_GEOMETRY_BINDING_SCHEMA, + "source_pack_id": self.source_pack_id, + "source_session_id": self.source_session_id, + "representation_profile_id": self.representation_profile_id, + "e31_qualification_id": self.e31_qualification_id, + "calibration_sha256": self.calibration_sha256, + "coordinate_frame": self.coordinate_frame, + "time_basis": self.time_basis, + "selected_offset_ms": self.selected_offset_ms, + "authority": _authority(), + } + + @classmethod + def from_dict(cls, value: object) -> TrackGeometrySourceBinding: + document = _object(value, "track geometry source binding") + _exact_keys( + document, + { + "schema_version", + "source_pack_id", + "source_session_id", + "representation_profile_id", + "e31_qualification_id", + "calibration_sha256", + "coordinate_frame", + "time_basis", + "selected_offset_ms", + "authority", + }, + "track geometry source binding", + ) + if document.get("schema_version") != TRACK_GEOMETRY_BINDING_SCHEMA: + raise TrackGeometryContractError("track geometry source binding schema is incompatible") + _require_authority(document.get("authority")) + return cls( + source_pack_id=_string(document, "source_pack_id"), + source_session_id=_string(document, "source_session_id"), + representation_profile_id=_string( + document, + "representation_profile_id", + ), + e31_qualification_id=_string(document, "e31_qualification_id"), + calibration_sha256=_string(document, "calibration_sha256"), + coordinate_frame=_string(document, "coordinate_frame"), + time_basis=_string(document, "time_basis"), + selected_offset_ms=_integer(document, "selected_offset_ms"), + ) + + +@dataclass(frozen=True, slots=True) +class PointSlab: + """Contiguous accepted points with one frame-local owner per source row.""" + + frame_index: int + source_frame_index: int + source_point_count: int + coordinate_frame: str + owner_keys: tuple[str, ...] + source_indices: Int64Array + points_xyz_m: Float32Array + owner_indices: UInt32Array + + def __post_init__(self) -> None: + _nonnegative_integer(self.frame_index, "frame index") + _nonnegative_integer(self.source_frame_index, "source frame index") + _nonnegative_integer(self.source_point_count, "source point count") + _safe_identifier(self.coordinate_frame, "point coordinate frame") + if len(set(self.owner_keys)) != len(self.owner_keys): + raise TrackGeometryContractError("point owner keys must be unique") + for owner_key in self.owner_keys: + _safe_identifier(owner_key, "point owner key") + + source_indices = np.asarray(self.source_indices) + points = np.asarray(self.points_xyz_m) + owner_indices = np.asarray(self.owner_indices) + if source_indices.dtype != np.dtype("= self.source_point_count) + or np.unique(source_indices).size != source_indices.size + ): + raise TrackGeometryContractError( + "source point indices must be unique and frame-local" + ) + if np.any(owner_indices >= len(self.owner_keys)): + raise TrackGeometryContractError("point owner index is out of range") + if np.unique(owner_indices).size != len(self.owner_keys): + raise TrackGeometryContractError("point owner table cannot contain unused owners") + elif self.owner_keys: + raise TrackGeometryContractError("empty point slabs cannot declare unused owners") + + source_copy = np.array(source_indices, dtype=" int: + return int(self.source_indices.size) + + def owner_point_count(self, owner_key: str) -> int: + try: + owner_index = self.owner_keys.index(owner_key) + except ValueError: + return 0 + return int(np.count_nonzero(self.owner_indices == owner_index)) + + def owned_source_indices(self, owner_key: str) -> Int64Array: + try: + owner_index = self.owner_keys.index(owner_key) + except ValueError: + result = np.empty(0, dtype=" dict[str, object]: + return { + "schema_version": POINT_SLAB_SCHEMA, + "frame_index": self.frame_index, + "source_frame_index": self.source_frame_index, + "source_point_count": self.source_point_count, + "coordinate_frame": self.coordinate_frame, + "owner_keys": list(self.owner_keys), + "source_indices": self.source_indices.tolist(), + "points_xyz_m": self.points_xyz_m.tolist(), + "owner_indices": self.owner_indices.tolist(), + "semantics": { + "source_indices_are_frame_local": True, + "one_owner_per_row": True, + "absence_of_points_means_free": False, + }, + } + + @classmethod + def from_dict(cls, value: object) -> PointSlab: + document = _object(value, "point slab") + _exact_keys( + document, + { + "schema_version", + "frame_index", + "source_frame_index", + "source_point_count", + "coordinate_frame", + "owner_keys", + "source_indices", + "points_xyz_m", + "owner_indices", + "semantics", + }, + "point slab", + ) + if document.get("schema_version") != POINT_SLAB_SCHEMA: + raise TrackGeometryContractError("point slab schema is incompatible") + semantics = _object(document.get("semantics"), "point slab semantics") + _exact_keys( + semantics, + { + "source_indices_are_frame_local", + "one_owner_per_row", + "absence_of_points_means_free", + }, + "point slab semantics", + ) + if ( + not _boolean(semantics, "source_indices_are_frame_local") + or not _boolean(semantics, "one_owner_per_row") + or _boolean(semantics, "absence_of_points_means_free") + ): + raise TrackGeometryContractError("point slab semantics changed") + owner_keys = tuple( + _string_value(item, "point owner key") for item in _array(document, "owner_keys") + ) + source_indices = cast( + Int64Array, + _integer_array(document, "source_indices", " None: + _safe_identifier(self.owner_key, "geometry owner key") + if not self.reason_codes or len(set(self.reason_codes)) != len(self.reason_codes): + raise TrackGeometryContractError("geometry reason codes must be nonempty and unique") + for reason in self.reason_codes: + _safe_identifier(reason, "geometry reason code") + self._validate_owner_identity() + self._validate_currentness() + self._validate_metric() + self._validate_state() + + def _validate_owner_identity(self) -> None: + if self.owner_kind is TrackGeometryOwnerKind.CAMERA_TRACK: + if ( + not isinstance(self.semantic_track_id, int) + or isinstance(self.semantic_track_id, bool) + or self.semantic_track_id < 0 + or not isinstance(self.semantic_label, str) + or not self.semantic_label.strip() + or len(self.semantic_label) > 120 + or self.bbox_xyxy is None + ): + raise TrackGeometryContractError( + "camera track requires camera-owned semantic identity" + ) + bbox = np.asarray(self.bbox_xyxy, dtype=np.float64) + if ( + bbox.shape != (4,) + or not np.isfinite(bbox).all() + or bbox[2] <= bbox[0] + or bbox[3] <= bbox[1] + ): + raise TrackGeometryContractError("camera track bbox is invalid") + elif ( + self.semantic_track_id is not None + or self.semantic_label is not None + or self.bbox_xyxy is not None + ): + raise TrackGeometryContractError("geometry-only owners cannot invent camera semantics") + + def _validate_currentness(self) -> None: + if self.currentness is TrackGeometryCurrentness.CURRENT: + if self.held_from_frame_index is not None or self.persistent_model_id is not None: + raise TrackGeometryContractError( + "current geometry cannot carry held or persistent provenance" + ) + elif self.currentness is TrackGeometryCurrentness.HELD: + if ( + self.owner_kind is not TrackGeometryOwnerKind.CAMERA_TRACK + or self.held_from_frame_index is None + or self.held_from_frame_index < 0 + or self.persistent_model_id is not None + ): + raise TrackGeometryContractError( + "held geometry requires prior camera-track provenance" + ) + elif ( + self.owner_kind is not TrackGeometryOwnerKind.GEOMETRY_CLUSTER + or self.held_from_frame_index is not None + or self.persistent_model_id is None + ): + raise TrackGeometryContractError( + "persistent geometry requires a separate geometry model" + ) + if self.persistent_model_id is not None: + _safe_identifier(self.persistent_model_id, "persistent model id") + + def _validate_metric(self) -> None: + has_range = self.range_m is not None + if has_range and ( + not isinstance(self.range_m, (int, float)) + or isinstance(self.range_m, bool) + or not math.isfinite(float(self.range_m)) + or float(self.range_m) <= 0.0 + ): + raise TrackGeometryContractError("geometry range is invalid") + if self.metric_basis is TrackGeometryMetricBasis.UNAVAILABLE: + if has_range: + raise TrackGeometryContractError("unavailable metric geometry cannot publish range") + elif not has_range: + raise TrackGeometryContractError("qualified metric geometry requires a positive range") + if ( + self.metric_basis is TrackGeometryMetricBasis.CURRENT_POINTS + and self.currentness is not TrackGeometryCurrentness.CURRENT + ): + raise TrackGeometryContractError("current-point metrics require current geometry") + if ( + self.metric_basis is TrackGeometryMetricBasis.HELD_LAST_CURRENT + and self.currentness is not TrackGeometryCurrentness.HELD + ): + raise TrackGeometryContractError("held metrics require held geometry") + if ( + self.metric_basis is TrackGeometryMetricBasis.PERSISTENT_MODEL + and self.currentness is not TrackGeometryCurrentness.PERSISTENT + ): + raise TrackGeometryContractError("persistent metrics require persistent geometry") + + def _validate_state(self) -> None: + if self.owner_kind is TrackGeometryOwnerKind.GEOMETRY_CLUSTER: + if self.evidence_state is not TrackGeometryEvidenceState.GEOMETRY_ONLY: + raise TrackGeometryContractError("unknown-class geometry must remain geometry-only") + if self.currentness is TrackGeometryCurrentness.CURRENT: + expected = TrackGeometryMetricBasis.CURRENT_POINTS + else: + expected = TrackGeometryMetricBasis.PERSISTENT_MODEL + if self.metric_basis is not expected: + raise TrackGeometryContractError( + "geometry-only metric basis does not match currentness" + ) + return + if self.evidence_state is TrackGeometryEvidenceState.GEOMETRY_ONLY: + raise TrackGeometryContractError("camera tracks cannot publish geometry-only state") + if self.currentness is TrackGeometryCurrentness.HELD: + if ( + self.evidence_state is not TrackGeometryEvidenceState.UNKNOWN + or self.metric_basis + not in { + TrackGeometryMetricBasis.HELD_LAST_CURRENT, + TrackGeometryMetricBasis.UNAVAILABLE, + } + ): + raise TrackGeometryContractError( + "held camera tracks must remain explicit unknown evidence" + ) + return + expected_basis = { + TrackGeometryEvidenceState.AGREE: (TrackGeometryMetricBasis.CURRENT_POINTS), + TrackGeometryEvidenceState.CAMERA_ONLY: (TrackGeometryMetricBasis.UNAVAILABLE), + TrackGeometryEvidenceState.CONFLICT: (TrackGeometryMetricBasis.UNAVAILABLE), + TrackGeometryEvidenceState.UNKNOWN: (TrackGeometryMetricBasis.UNAVAILABLE), + }[self.evidence_state] + if self.metric_basis is not expected_basis: + raise TrackGeometryContractError( + "camera-track evidence state cannot invent metric geometry" + ) + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": TRACK_GEOMETRY_SCHEMA, + "owner_key": self.owner_key, + "owner_kind": self.owner_kind.value, + "evidence_state": self.evidence_state.value, + "currentness": self.currentness.value, + "metric_basis": self.metric_basis.value, + "reason_codes": list(self.reason_codes), + "semantic": ( + None + if self.owner_kind is TrackGeometryOwnerKind.GEOMETRY_CLUSTER + else { + "owner": "camera", + "track_id": self.semantic_track_id, + "label": self.semantic_label, + "bbox_xyxy": (None if self.bbox_xyxy is None else list(self.bbox_xyxy)), + } + ), + "range_m": self.range_m, + "held_from_frame_index": self.held_from_frame_index, + "persistent_model_id": self.persistent_model_id, + "policy": { + "camera_owns_semantics": True, + "geometry_cannot_invent_class": True, + "absence_of_points_means_free": False, + }, + "authority": _authority(), + } + + @classmethod + def from_dict(cls, value: object) -> TrackGeometry: + document = _object(value, "track geometry") + _exact_keys( + document, + { + "schema_version", + "owner_key", + "owner_kind", + "evidence_state", + "currentness", + "metric_basis", + "reason_codes", + "semantic", + "range_m", + "held_from_frame_index", + "persistent_model_id", + "policy", + "authority", + }, + "track geometry", + ) + if document.get("schema_version") != TRACK_GEOMETRY_SCHEMA: + raise TrackGeometryContractError("track geometry schema is incompatible") + policy = _object(document.get("policy"), "track geometry policy") + _exact_keys( + policy, + { + "camera_owns_semantics", + "geometry_cannot_invent_class", + "absence_of_points_means_free", + }, + "track geometry policy", + ) + if ( + not _boolean(policy, "camera_owns_semantics") + or not _boolean(policy, "geometry_cannot_invent_class") + or _boolean(policy, "absence_of_points_means_free") + ): + raise TrackGeometryContractError("track geometry policy changed") + _require_authority(document.get("authority")) + owner_kind = _enum( + TrackGeometryOwnerKind, + document.get("owner_kind"), + "track geometry owner kind", + ) + semantic = document.get("semantic") + if owner_kind is TrackGeometryOwnerKind.CAMERA_TRACK: + semantic_document = _object(semantic, "camera semantic identity") + _exact_keys( + semantic_document, + {"owner", "track_id", "label", "bbox_xyxy"}, + "camera semantic identity", + ) + if semantic_document.get("owner") != "camera": + raise TrackGeometryContractError("semantic identity must remain camera-owned") + semantic_track_id = _integer(semantic_document, "track_id") + semantic_label = _string(semantic_document, "label") + bbox_xyxy = _bbox(semantic_document.get("bbox_xyxy")) + elif semantic is None: + semantic_track_id = None + semantic_label = None + bbox_xyxy = None + else: + raise TrackGeometryContractError("geometry-only identity cannot contain semantics") + return cls( + owner_key=_string(document, "owner_key"), + owner_kind=owner_kind, + evidence_state=_enum( + TrackGeometryEvidenceState, + document.get("evidence_state"), + "track geometry evidence state", + ), + currentness=_enum( + TrackGeometryCurrentness, + document.get("currentness"), + "track geometry currentness", + ), + metric_basis=_enum( + TrackGeometryMetricBasis, + document.get("metric_basis"), + "track geometry metric basis", + ), + reason_codes=tuple( + _string_value(item, "geometry reason code") + for item in _array(document, "reason_codes") + ), + semantic_track_id=semantic_track_id, + semantic_label=semantic_label, + bbox_xyxy=bbox_xyxy, + range_m=_optional_number(document.get("range_m"), "range"), + held_from_frame_index=_optional_integer( + document.get("held_from_frame_index"), + "held source frame", + ), + persistent_model_id=_optional_string( + document.get("persistent_model_id"), + "persistent model id", + ), + ) + + +@dataclass(frozen=True, slots=True) +class TrackGeometryFrame: + """One source frame with explicit objects and exclusive accepted points.""" + + binding: TrackGeometrySourceBinding + frame_index: int + source_frame_index: int + session_seconds: float + source_available: bool + point_slab: PointSlab + geometries: tuple[TrackGeometry, ...] + + def __post_init__(self) -> None: + _nonnegative_integer(self.frame_index, "frame index") + _nonnegative_integer(self.source_frame_index, "source frame index") + if ( + not isinstance(self.session_seconds, (int, float)) + or isinstance(self.session_seconds, bool) + or not math.isfinite(float(self.session_seconds)) + or float(self.session_seconds) < 0.0 + ): + raise TrackGeometryContractError("session time is invalid") + if not isinstance(self.source_available, bool): + raise TrackGeometryContractError("source availability is invalid") + if ( + self.point_slab.frame_index != self.frame_index + or self.point_slab.source_frame_index != self.source_frame_index + or self.point_slab.coordinate_frame != self.binding.coordinate_frame + ): + raise TrackGeometryContractError("point slab frame binding is inconsistent") + by_owner = {geometry.owner_key: geometry for geometry in self.geometries} + if len(by_owner) != len(self.geometries): + raise TrackGeometryContractError("geometry owner keys must be unique") + unknown_owners = set(self.point_slab.owner_keys).difference(by_owner) + if unknown_owners: + raise TrackGeometryContractError("point slab references an unknown geometry owner") + for geometry in self.geometries: + owned_count = self.point_slab.owner_point_count(geometry.owner_key) + owns_current = geometry.metric_basis is TrackGeometryMetricBasis.CURRENT_POINTS + if owns_current is not (owned_count > 0): + raise TrackGeometryContractError( + "current point ownership and metric basis disagree" + ) + if geometry.currentness is TrackGeometryCurrentness.HELD and ( + geometry.held_from_frame_index is None + or geometry.held_from_frame_index >= self.frame_index + ): + raise TrackGeometryContractError("held geometry must reference an earlier frame") + if not self.source_available and ( + self.point_slab.row_count + or any( + geometry.metric_basis is TrackGeometryMetricBasis.CURRENT_POINTS + for geometry in self.geometries + ) + ): + raise TrackGeometryContractError( + "unavailable source cannot publish current metric points" + ) + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": TRACK_GEOMETRY_FRAME_SCHEMA, + "binding": self.binding.to_dict(), + "frame_index": self.frame_index, + "source_frame_index": self.source_frame_index, + "session_seconds": self.session_seconds, + "source_available": self.source_available, + "point_slab": self.point_slab.to_dict(), + "geometries": [geometry.to_dict() for geometry in self.geometries], + "policy": { + "camera_owns_semantics": True, + "one_owner_per_source_point": True, + "current_held_persistent_are_separate": True, + "absence_of_points_means_free": False, + "unknown_remains_unknown": True, + }, + "authority": _authority(), + } + + @classmethod + def from_dict(cls, value: object) -> TrackGeometryFrame: + document = _object(value, "track geometry frame") + _exact_keys( + document, + { + "schema_version", + "binding", + "frame_index", + "source_frame_index", + "session_seconds", + "source_available", + "point_slab", + "geometries", + "policy", + "authority", + }, + "track geometry frame", + ) + if document.get("schema_version") != TRACK_GEOMETRY_FRAME_SCHEMA: + raise TrackGeometryContractError("track geometry frame schema is incompatible") + policy = _object(document.get("policy"), "track geometry frame policy") + _exact_keys( + policy, + { + "camera_owns_semantics", + "one_owner_per_source_point", + "current_held_persistent_are_separate", + "absence_of_points_means_free", + "unknown_remains_unknown", + }, + "track geometry frame policy", + ) + if ( + not _boolean(policy, "camera_owns_semantics") + or not _boolean(policy, "one_owner_per_source_point") + or not _boolean( + policy, + "current_held_persistent_are_separate", + ) + or _boolean(policy, "absence_of_points_means_free") + or not _boolean(policy, "unknown_remains_unknown") + ): + raise TrackGeometryContractError("track geometry frame policy changed") + _require_authority(document.get("authority")) + return cls( + binding=TrackGeometrySourceBinding.from_dict(document.get("binding")), + frame_index=_integer(document, "frame_index"), + source_frame_index=_integer(document, "source_frame_index"), + session_seconds=_number(document, "session_seconds"), + source_available=_boolean(document, "source_available"), + point_slab=PointSlab.from_dict(document.get("point_slab")), + geometries=tuple( + TrackGeometry.from_dict(item) for item in _array(document, "geometries") + ), + ) + + +def _safe_identifier(value: str, label: str) -> str: + if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None: + raise TrackGeometryContractError(f"{label} is not a safe identifier") + return value + + +def _nonnegative_integer(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise TrackGeometryContractError(f"{label} is invalid") + return value + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + raise TrackGeometryContractError(f"{label} must be an object") + return value + + +def _exact_keys( + document: dict[str, object], + expected: set[str], + label: str, +) -> None: + if set(document) != expected: + raise TrackGeometryContractError(f"{label} fields are incompatible") + + +def _array(document: dict[str, object], key: str) -> list[object]: + value = document.get(key) + if not isinstance(value, list): + raise TrackGeometryContractError(f"{key} must be an array") + return value + + +def _string(document: dict[str, object], key: str) -> str: + return _string_value(document.get(key), key) + + +def _string_value(value: object, label: str) -> str: + if not isinstance(value, str) or not value: + raise TrackGeometryContractError(f"{label} must be a nonempty string") + return value + + +def _optional_string(value: object, label: str) -> str | None: + if value is None: + return None + return _string_value(value, label) + + +def _integer(document: dict[str, object], key: str) -> int: + value = document.get(key) + if not isinstance(value, int) or isinstance(value, bool): + raise TrackGeometryContractError(f"{key} must be an integer") + return value + + +def _optional_integer(value: object, label: str) -> int | None: + if value is None: + return None + if not isinstance(value, int) or isinstance(value, bool): + raise TrackGeometryContractError(f"{label} must be an integer") + return value + + +def _number(document: dict[str, object], key: str) -> float: + return _number_value(document.get(key), key) + + +def _optional_number(value: object, label: str) -> float | None: + if value is None: + return None + return _number_value(value, label) + + +def _number_value(value: object, label: str) -> float: + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(float(value)) + ): + raise TrackGeometryContractError(f"{label} must be a finite number") + return float(value) + + +def _boolean(document: dict[str, object], key: str) -> bool: + value = document.get(key) + if not isinstance(value, bool): + raise TrackGeometryContractError(f"{key} must be a boolean") + return value + + +def _integer_array( + document: dict[str, object], + key: str, + dtype: str, +) -> npt.NDArray[np.signedinteger | np.unsignedinteger]: + values = _array(document, key) + if any(not isinstance(value, int) or isinstance(value, bool) for value in values): + raise TrackGeometryContractError(f"{key} must contain integers") + try: + return np.asarray(values, dtype=dtype) + except (OverflowError, ValueError) as exc: + raise TrackGeometryContractError(f"{key} is out of range") from exc + + +def _points_array(value: object) -> Float32Array: + if not isinstance(value, list) or any( + not isinstance(row, list) + or len(row) != 3 + or any(not isinstance(item, (int, float)) or isinstance(item, bool) for item in row) + for row in value + ): + raise TrackGeometryContractError("point coordinates must be an Nx3 array") + if not value: + return np.empty((0, 3), dtype=" tuple[float, float, float, float]: + if not isinstance(value, list) or len(value) != 4: + raise TrackGeometryContractError("camera bbox must have four values") + numbers = tuple(_number_value(item, "camera bbox coordinate") for item in value) + return ( + float(numbers[0]), + float(numbers[1]), + float(numbers[2]), + float(numbers[3]), + ) + + +def _enum[EnumT: StrEnum]( + enum_type: type[EnumT], + value: object, + label: str, +) -> EnumT: + if not isinstance(value, str): + raise TrackGeometryContractError(f"{label} must be a string") + try: + return enum_type(value) + except ValueError as exc: + raise TrackGeometryContractError(f"{label} is unknown") from exc + + +def _authority() -> dict[str, bool]: + return { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + } + + +def _require_authority(value: object) -> None: + authority = _object(value, "track geometry authority") + _exact_keys( + authority, + {"commands_enabled", "navigation_or_safety_accepted"}, + "track geometry authority", + ) + if _boolean(authority, "commands_enabled") or _boolean( + authority, "navigation_or_safety_accepted" + ): + raise TrackGeometryContractError("TrackGeometry v1 cannot grant runtime authority") diff --git a/tests/test_track_geometry.py b/tests/test_track_geometry.py new file mode 100644 index 0000000..da06f61 --- /dev/null +++ b/tests/test_track_geometry.py @@ -0,0 +1,320 @@ +from __future__ import annotations + +from dataclasses import replace + +import numpy as np +import pytest + +from k1link.compute.sensor_representation import K1_LIO_PCL_CAPABILITIES +from k1link.compute.track_geometry import ( + PointSlab, + TrackGeometry, + TrackGeometryContractError, + TrackGeometryCurrentness, + TrackGeometryEvidenceState, + TrackGeometryFrame, + TrackGeometryMetricBasis, + TrackGeometryOwnerKind, + TrackGeometrySourceBinding, +) + + +def _binding() -> TrackGeometrySourceBinding: + return TrackGeometrySourceBinding( + source_pack_id="e10-lidar-pack-" + "a" * 64, + source_session_id="source-session", + representation_profile_id=K1_LIO_PCL_CAPABILITIES.profile_id, + e31_qualification_id="e31-source-qualification-" + "b" * 64, + calibration_sha256="c" * 64, + coordinate_frame="map", + time_basis="recorded-host-arrival-best-effort", + selected_offset_ms=0, + ) + + +def _agree(owner_key: str = "track:7") -> TrackGeometry: + return TrackGeometry( + owner_key=owner_key, + owner_kind=TrackGeometryOwnerKind.CAMERA_TRACK, + evidence_state=TrackGeometryEvidenceState.AGREE, + currentness=TrackGeometryCurrentness.CURRENT, + metric_basis=TrackGeometryMetricBasis.CURRENT_POINTS, + reason_codes=("connected-occupied-support",), + semantic_track_id=7, + semantic_label="car", + bbox_xyxy=(10.0, 20.0, 50.0, 60.0), + range_m=8.2, + ) + + +def _camera_only() -> TrackGeometry: + return TrackGeometry( + owner_key="track:8", + owner_kind=TrackGeometryOwnerKind.CAMERA_TRACK, + evidence_state=TrackGeometryEvidenceState.CAMERA_ONLY, + currentness=TrackGeometryCurrentness.CURRENT, + metric_basis=TrackGeometryMetricBasis.UNAVAILABLE, + reason_codes=("sparse-support",), + semantic_track_id=8, + semantic_label="person", + bbox_xyxy=(60.0, 20.0, 80.0, 90.0), + ) + + +def _geometry_only() -> TrackGeometry: + return TrackGeometry( + owner_key="geometry:2", + owner_kind=TrackGeometryOwnerKind.GEOMETRY_CLUSTER, + evidence_state=TrackGeometryEvidenceState.GEOMETRY_ONLY, + currentness=TrackGeometryCurrentness.CURRENT, + metric_basis=TrackGeometryMetricBasis.CURRENT_POINTS, + reason_codes=("unassociated-occupied-component",), + range_m=6.4, + ) + + +def _point_slab() -> PointSlab: + return PointSlab( + frame_index=10, + source_frame_index=1010, + source_point_count=100, + coordinate_frame="map", + owner_keys=("track:7", "geometry:2"), + source_indices=np.asarray([1, 2, 50], dtype=np.int64), + points_xyz_m=np.asarray( + [ + [1.0, 0.0, 2.0], + [1.1, 0.0, 2.0], + [5.0, 1.0, 0.5], + ], + dtype=np.float32, + ), + owner_indices=np.asarray([0, 0, 1], dtype=np.uint32), + ) + + +def test_track_geometry_frame_round_trips_exact_source_ownership() -> None: + frame = TrackGeometryFrame( + binding=_binding(), + frame_index=10, + source_frame_index=1010, + session_seconds=12.5, + source_available=True, + point_slab=_point_slab(), + geometries=(_agree(), _camera_only(), _geometry_only()), + ) + + document = frame.to_dict() + restored = TrackGeometryFrame.from_dict(document) + + assert restored.to_dict() == document + assert restored.point_slab.owned_source_indices("track:7").tolist() == [1, 2] + assert restored.point_slab.owned_source_indices("geometry:2").tolist() == [50] + assert restored.point_slab.source_indices.flags.writeable is False + assert document["policy"] == { + "camera_owns_semantics": True, + "one_owner_per_source_point": True, + "current_held_persistent_are_separate": True, + "absence_of_points_means_free": False, + "unknown_remains_unknown": True, + } + + +def test_point_slab_rejects_duplicate_or_out_of_frame_source_indices() -> None: + slab = _point_slab() + with pytest.raises( + TrackGeometryContractError, + match="unique and frame-local", + ): + replace( + slab, + source_indices=np.asarray([1, 1, 50], dtype=np.int64), + owner_indices=np.asarray([0, 1, 1], dtype=np.uint32), + ) + with pytest.raises( + TrackGeometryContractError, + match="unique and frame-local", + ): + replace( + slab, + source_indices=np.asarray([1, 2, 100], dtype=np.int64), + ) + + +def test_point_slab_rejects_unknown_or_multiple_owners() -> None: + slab = _point_slab() + with pytest.raises(TrackGeometryContractError, match="owner keys must be unique"): + replace(slab, owner_keys=("track:7", "track:7")) + with pytest.raises(TrackGeometryContractError, match="out of range"): + replace( + slab, + owner_indices=np.asarray([0, 0, 2], dtype=np.uint32), + ) + with pytest.raises(TrackGeometryContractError, match="unused owners"): + replace( + slab, + owner_indices=np.asarray([0, 0, 0], dtype=np.uint32), + ) + + +def test_frame_rejects_current_points_claimed_by_two_geometry_products() -> None: + slab = _point_slab() + with pytest.raises( + TrackGeometryContractError, + match="geometry owner keys must be unique", + ): + TrackGeometryFrame( + binding=_binding(), + frame_index=10, + source_frame_index=1010, + session_seconds=12.5, + source_available=True, + point_slab=slab, + geometries=(_agree(), _agree(), _geometry_only()), + ) + + +def test_camera_only_and_conflict_cannot_invent_metric_range() -> None: + with pytest.raises( + TrackGeometryContractError, + match="unavailable metric geometry cannot publish range", + ): + replace(_camera_only(), range_m=3.0) + with pytest.raises( + TrackGeometryContractError, + match="cannot invent metric geometry", + ): + replace( + _camera_only(), + evidence_state=TrackGeometryEvidenceState.CONFLICT, + metric_basis=TrackGeometryMetricBasis.CURRENT_POINTS, + range_m=3.0, + ) + + +def test_geometry_only_cannot_invent_camera_semantics() -> None: + with pytest.raises( + TrackGeometryContractError, + match="cannot invent camera semantics", + ): + replace( + _geometry_only(), + semantic_track_id=9, + semantic_label="car", + bbox_xyxy=(1.0, 1.0, 2.0, 2.0), + ) + + +def test_held_geometry_keeps_provenance_but_owns_no_current_points() -> None: + held = TrackGeometry( + owner_key="track:7", + owner_kind=TrackGeometryOwnerKind.CAMERA_TRACK, + evidence_state=TrackGeometryEvidenceState.UNKNOWN, + currentness=TrackGeometryCurrentness.HELD, + metric_basis=TrackGeometryMetricBasis.HELD_LAST_CURRENT, + reason_codes=("temporally-held-last-supported",), + semantic_track_id=7, + semantic_label="car", + bbox_xyxy=(10.0, 20.0, 50.0, 60.0), + range_m=8.2, + held_from_frame_index=9, + ) + empty_slab = PointSlab( + frame_index=10, + source_frame_index=1010, + source_point_count=100, + coordinate_frame="map", + owner_keys=(), + source_indices=np.empty(0, dtype=np.int64), + points_xyz_m=np.empty((0, 3), dtype=np.float32), + owner_indices=np.empty(0, dtype=np.uint32), + ) + frame = TrackGeometryFrame( + binding=_binding(), + frame_index=10, + source_frame_index=1010, + session_seconds=12.5, + source_available=True, + point_slab=empty_slab, + geometries=(held,), + ) + + assert TrackGeometryFrame.from_dict(frame.to_dict()).to_dict() == frame.to_dict() + with pytest.raises( + TrackGeometryContractError, + match="current point ownership and metric basis disagree", + ): + TrackGeometryFrame( + binding=_binding(), + frame_index=10, + source_frame_index=1010, + session_seconds=12.5, + source_available=True, + point_slab=replace( + _point_slab(), + owner_keys=("track:7", "geometry:2"), + ), + geometries=(held, _geometry_only()), + ) + + +def test_persistent_geometry_is_a_separate_model_not_current_source_points() -> None: + persistent = TrackGeometry( + owner_key="persistent:wall:1", + owner_kind=TrackGeometryOwnerKind.GEOMETRY_CLUSTER, + evidence_state=TrackGeometryEvidenceState.GEOMETRY_ONLY, + currentness=TrackGeometryCurrentness.PERSISTENT, + metric_basis=TrackGeometryMetricBasis.PERSISTENT_MODEL, + reason_codes=("persistent-occupied-model",), + range_m=5.0, + persistent_model_id="local-occupied-model/v1", + ) + + assert persistent.semantic_label is None + with pytest.raises( + TrackGeometryContractError, + match="current-point metrics require current geometry", + ): + replace( + persistent, + metric_basis=TrackGeometryMetricBasis.CURRENT_POINTS, + ) + + +def test_unavailable_source_cannot_publish_current_point_geometry() -> None: + with pytest.raises( + TrackGeometryContractError, + match="unavailable source cannot publish current metric points", + ): + TrackGeometryFrame( + binding=_binding(), + frame_index=10, + source_frame_index=1010, + session_seconds=12.5, + source_available=False, + point_slab=_point_slab(), + geometries=(_agree(), _geometry_only()), + ) + + +def test_contract_rejects_authority_and_unknown_fields() -> None: + frame = TrackGeometryFrame( + binding=_binding(), + frame_index=10, + source_frame_index=1010, + session_seconds=12.5, + source_available=True, + point_slab=_point_slab(), + geometries=(_agree(), _camera_only(), _geometry_only()), + ) + document = frame.to_dict() + authority = document["authority"] + assert isinstance(authority, dict) + authority["navigation_or_safety_accepted"] = True + with pytest.raises(TrackGeometryContractError, match="cannot grant"): + TrackGeometryFrame.from_dict(document) + + document = frame.to_dict() + document["free_space"] = True + with pytest.raises(TrackGeometryContractError, match="fields are incompatible"): + TrackGeometryFrame.from_dict(document)