From ae1e41f7fe1698b18461cf75c1a4909d7b25020f Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 5 Aug 2026 13:32:34 +0300 Subject: [PATCH] feat(perception): establish object centric contracts --- ...E_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md | 33 + src/k1link/perception/__init__.py | 125 ++ src/k1link/perception/adapters.py | 274 ++++ src/k1link/perception/baseline.py | 287 +++++ src/k1link/perception/contracts.py | 1109 +++++++++++++++++ src/k1link/perception/providers.py | 334 +++++ tests/test_perception_architecture.py | 84 ++ tests/test_perception_contracts.py | 419 +++++++ 8 files changed, 2665 insertions(+) create mode 100644 src/k1link/perception/__init__.py create mode 100644 src/k1link/perception/adapters.py create mode 100644 src/k1link/perception/baseline.py create mode 100644 src/k1link/perception/contracts.py create mode 100644 src/k1link/perception/providers.py create mode 100644 tests/test_perception_architecture.py create mode 100644 tests/test_perception_contracts.py diff --git a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md index f96bd6e..9a07c07 100644 --- a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md +++ b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md @@ -623,6 +623,39 @@ Milestone 4 is complete only when all of the following are true: - [ ] The Ops card contains exact files, commits, result IDs and validation. - [ ] Physical live, mounted geometry, navigation and safety remain visibly deferred. +## Implementation record + +### 2026-08-05 — M4.0 and M4.1 + +M4.0 is closed by the executable baseline +`config/perception/m4-recorded-realtime-baseline-v1.json`. It binds only +RAVNOVES00 and the exact source-fusion, E32, E33, E34, E35 and E46J result +documents, KB4 calibration, valid-FOV, YOLOX model/config/profile identities and +the observed E15 Worker 006 rollback process. `k1link.perception.baseline` +validates the profile and resolves all six immutable evidence documents by file +SHA-256, schema, result and identity. The reusable/historical split is frozen in +`config/perception/m4-reuse-inventory-v1.json` and enforced by an import-boundary +test. + +M4.1 is closed by `k1link.perception.contracts`, `providers` and `adapters`: + +- six exact-key versioned serializers implement SourceEnvelope, + ObjectProposal2D, ObstacleObservation, TemporalObstacle, LocalObstacleMap and + ThreatAssessment; +- provider protocols and one strict graph-configuration contract pin all six + roles, bounded queues, deadlines and replay-only authority; +- one-way TrackGeometry and E34 adapters preserve exact point ownership, + camera-only uncertainty, held/expired state and ephemeral component identity; +- adversarial tests reject class-dependent occupancy identity, duplicate point + ownership, range without current qualified points, held-as-current, + missing-LiDAR-as-free and physical/command authority in replay. + +Validation at this increment: 17 focused M4 tests, 60 related geometry/temporal/ +telemetry tests and the complete Python suite (`1205 passed, 1 skipped`). Scoped +Ruff and strict mypy pass for `src/k1link/perception`. Repository-wide Ruff and +mypy remain red only in pre-existing legacy experiment/web modules; those errors +were not hidden or expanded into this product boundary. + ## Implementation order The implementation sequence is intentionally strict: diff --git a/src/k1link/perception/__init__.py b/src/k1link/perception/__init__.py new file mode 100644 index 0000000..a1f6a29 --- /dev/null +++ b/src/k1link/perception/__init__.py @@ -0,0 +1,125 @@ +"""Object-centric, source-neutral Mission Core perception product boundary.""" + +from .adapters import ( + PerceptionAdapterError, + observations_from_track_geometry, + temporal_obstacles_from_e34_projection, +) +from .baseline import ( + BASELINE_PROFILE_ID, + BASELINE_SCHEMA, + BaselineContractError, + BaselineProfile, + BaselineVerification, + load_m4_baseline, + validate_reuse_inventory, + verify_m4_baseline, +) +from .contracts import ( + LOCAL_OBSTACLE_MAP_SCHEMA, + OBJECT_PROPOSAL_SCHEMA, + OBSTACLE_OBSERVATION_SCHEMA, + SOURCE_ENVELOPE_SCHEMA, + TEMPORAL_OBSTACLE_SCHEMA, + THREAT_ASSESSMENT_SCHEMA, + BoundingRegion2D, + ClockBasis, + CorridorIntersection, + EvidenceBasis, + EvidenceCurrentness, + FalseAuthority, + GridCell, + HistorySample, + LocalObstacleMap, + MetricGeometry, + ModalityOutcome, + ModalityStatus, + MotionState, + ObjectProposal2D, + ObstacleObservation, + PerceptionContractError, + QualificationState, + SourceAccounting, + SourceEnvelope, + TemporalObstacle, + TemporalState, + ThreatAssessment, + ThreatAuthority, + ThreatDecision, + TimestampBundle, + validate_exclusive_point_ownership, +) +from .providers import ( + REFERENCE_GRAPH_CONFIG_SCHEMA, + DetectorProvider, + GeometryAssociationProvider, + GraphAuthority, + MotionProvider, + ProviderContractError, + ProviderPin, + ProviderRole, + QueuePolicy, + ReferencePerceptionGraphConfig, + SourceProvider, + TemporalStateProvider, + ThreatProvider, +) + +__all__ = [ + "BASELINE_PROFILE_ID", + "BASELINE_SCHEMA", + "BaselineContractError", + "BaselineProfile", + "BaselineVerification", + "load_m4_baseline", + "validate_reuse_inventory", + "verify_m4_baseline", + "PerceptionAdapterError", + "observations_from_track_geometry", + "temporal_obstacles_from_e34_projection", + "LOCAL_OBSTACLE_MAP_SCHEMA", + "OBJECT_PROPOSAL_SCHEMA", + "OBSTACLE_OBSERVATION_SCHEMA", + "SOURCE_ENVELOPE_SCHEMA", + "TEMPORAL_OBSTACLE_SCHEMA", + "THREAT_ASSESSMENT_SCHEMA", + "BoundingRegion2D", + "ClockBasis", + "CorridorIntersection", + "EvidenceBasis", + "EvidenceCurrentness", + "FalseAuthority", + "GridCell", + "HistorySample", + "LocalObstacleMap", + "MetricGeometry", + "ModalityOutcome", + "ModalityStatus", + "MotionState", + "ObjectProposal2D", + "ObstacleObservation", + "PerceptionContractError", + "QualificationState", + "SourceAccounting", + "SourceEnvelope", + "TemporalObstacle", + "TemporalState", + "ThreatAssessment", + "ThreatAuthority", + "ThreatDecision", + "TimestampBundle", + "validate_exclusive_point_ownership", + "REFERENCE_GRAPH_CONFIG_SCHEMA", + "DetectorProvider", + "GeometryAssociationProvider", + "GraphAuthority", + "MotionProvider", + "ProviderContractError", + "ProviderPin", + "ProviderRole", + "QueuePolicy", + "ReferencePerceptionGraphConfig", + "SourceProvider", + "TemporalStateProvider", + "ThreatProvider", +] diff --git a/src/k1link/perception/adapters.py b/src/k1link/perception/adapters.py new file mode 100644 index 0000000..e1aa3b5 --- /dev/null +++ b/src/k1link/perception/adapters.py @@ -0,0 +1,274 @@ +"""One-way adapters from admitted historical primitives to product contracts. + +The adapters do not mutate TrackGeometry/E34 documents and never upgrade held, +persistent or missing evidence into current metric occupancy. +""" + +from __future__ import annotations + +import math +from collections.abc import Mapping + +import numpy as np + +from k1link.compute.temporal_occupied_layer import TemporalFrameProjection +from k1link.compute.track_geometry import ( + TrackGeometryCurrentness, + TrackGeometryEvidenceState, + TrackGeometryFrame, + TrackGeometryMetricBasis, + TrackGeometryOwnerKind, +) + +from .contracts import ( + EvidenceBasis, + EvidenceCurrentness, + GridCell, + HistorySample, + MetricGeometry, + MotionState, + ObstacleObservation, + TemporalObstacle, + TemporalState, + validate_exclusive_point_ownership, +) + + +class PerceptionAdapterError(ValueError): + """An admitted historical value cannot be represented without inventing evidence.""" + + +def observations_from_track_geometry( + frame: TrackGeometryFrame, + *, + source_id: str, + frame_id: str, + evidence_time_ns: int, + proposal_ids_by_owner: Mapping[str, tuple[str, ...]] | None = None, +) -> tuple[ObstacleObservation, ...]: + """Adapt TrackGeometry v1 while preserving exact point ownership and uncertainty.""" + + proposals = proposal_ids_by_owner or {} + observations: list[ObstacleObservation] = [] + for geometry in frame.geometries: + source_indices = frame.point_slab.owned_source_indices(geometry.owner_key) + source_point_ids = tuple(int(value) for value in source_indices.tolist()) + metric_geometry: MetricGeometry | None = None + currentness = _currentness(geometry.currentness) + occupied_support = False + if geometry.metric_basis is TrackGeometryMetricBasis.CURRENT_POINTS: + if not source_point_ids or geometry.range_m is None: + raise PerceptionAdapterError( + "current TrackGeometry lost its qualified point support" + ) + owner_index = frame.point_slab.owner_keys.index(geometry.owner_key) + points = frame.point_slab.points_xyz_m[ + frame.point_slab.owner_indices == owner_index + ].astype(np.float64, copy=False) + centroid = points.mean(axis=0) + covariance = points.var(axis=0) + metric_geometry = MetricGeometry( + coordinate_frame=frame.binding.coordinate_frame, + centroid_xyz_m=(float(centroid[0]), float(centroid[1]), float(centroid[2])), + range_m=float(geometry.range_m), + covariance_diagonal_m2=( + float(covariance[0]), + float(covariance[1]), + float(covariance[2]), + ), + ) + occupied_support = True + else: + source_point_ids = () + + basis = _basis(geometry.evidence_state, geometry.owner_kind) + if basis in {EvidenceBasis.CAMERA, EvidenceBasis.CONFLICT}: + metric_geometry = None + source_point_ids = () + occupied_support = False + if currentness is not EvidenceCurrentness.CURRENT: + metric_geometry = None + source_point_ids = () + occupied_support = False + + reason_codes = list(geometry.reason_codes) + if geometry.currentness is TrackGeometryCurrentness.PERSISTENT: + reason_codes.append("persistent-model-not-product-authority") + observation = ObstacleObservation( + observation_id=f"{frame_id}:{geometry.owner_key}", + occupancy_key=f"{frame_id}:{geometry.owner_key}", + source_id=source_id, + frame_id=frame_id, + evidence_time_ns=evidence_time_ns, + basis=basis, + currentness=currentness, + occupied_support=occupied_support, + source_point_ids=source_point_ids, + metric_geometry=metric_geometry, + proposal_ids=proposals.get(geometry.owner_key, ()), + semantic_hint=geometry.semantic_label, + reason_codes=tuple(dict.fromkeys(reason_codes)), + ) + if basis is EvidenceBasis.CAMERA and not observation.proposal_ids: + raise PerceptionAdapterError( + "camera-only geometry requires its source proposal binding" + ) + observations.append(observation) + result = tuple(observations) + validate_exclusive_point_ownership(result) + return result + + +def temporal_obstacles_from_e34_projection( + projection: TemporalFrameProjection, + *, + coordinate_frame: str, + ttl_ns: int, +) -> tuple[TemporalObstacle, ...]: + """Adapt current/held/expired E34 components without persistent identity claims.""" + + document = projection.document + results: list[TemporalObstacle] = [] + for collection_name, state in ( + ("current", TemporalState.CURRENT), + ("held", TemporalState.HELD), + ("expired", TemporalState.EXPIRED), + ): + collection = document.get(collection_name) + if not isinstance(collection, list): + raise PerceptionAdapterError(f"E34 projection {collection_name} must be an array") + for value in collection: + component = _object(value, f"E34 {collection_name} component") + temporal_id = _integer(component.get("temporal_id"), "E34 temporal id") + age_seconds = _number( + component.get("last_observed_age_seconds"), + "E34 component age", + ) + age_ns = max(0, int(round(age_seconds * 1_000_000_000))) + history = _history(component.get("history_tail")) + cells = _cells(projection, component) if state is not TemporalState.EXPIRED else () + centroid = ( + None + if not cells + else _vector3(component.get("centroid_map_xyz_m"), "E34 centroid") + ) + labels = _semantic_labels(component.get("semantic_provenance")) + results.append( + TemporalObstacle( + component_id=f"e34-component-{temporal_id}", + identity_scope="ephemeral", + state=state, + ttl_ns=ttl_ns, + last_hit_ns=max(0, history[-1].evidence_time_ns), + age_ns=age_ns, + association_basis=_safe_reason(component.get("association_reason")), + history=history, + cells=cells, + coordinate_frame=coordinate_frame if cells else None, + last_centroid_xyz_m=centroid, + motion=MotionState.UNKNOWN, + motion_confidence=0.0, + motion_reason="motion-not-estimated", + semantic_hint=labels[-1] if labels else None, + ) + ) + component_ids = [item.component_id for item in results] + if len(set(component_ids)) != len(component_ids): + raise PerceptionAdapterError("E34 projection contains duplicate temporal ids") + return tuple(results) + + +def _basis( + evidence_state: TrackGeometryEvidenceState, + owner_kind: TrackGeometryOwnerKind, +) -> EvidenceBasis: + if evidence_state is TrackGeometryEvidenceState.AGREE: + return EvidenceBasis.FUSED + if evidence_state is TrackGeometryEvidenceState.CONFLICT: + return EvidenceBasis.CONFLICT + if owner_kind is TrackGeometryOwnerKind.GEOMETRY_CLUSTER: + return EvidenceBasis.LIDAR + return EvidenceBasis.CAMERA + + +def _currentness(value: TrackGeometryCurrentness) -> EvidenceCurrentness: + if value is TrackGeometryCurrentness.CURRENT: + return EvidenceCurrentness.CURRENT + if value is TrackGeometryCurrentness.HELD: + return EvidenceCurrentness.HELD + return EvidenceCurrentness.STALE + + +def _history(value: object) -> tuple[HistorySample, ...]: + if not isinstance(value, list) or not value: + raise PerceptionAdapterError("E34 component history must be nonempty") + history: list[HistorySample] = [] + for item in value[-32:]: + document = _object(item, "E34 history sample") + frame_index = _integer(document.get("frame_index"), "E34 history frame") + session_seconds = _number(document.get("session_seconds"), "E34 history time") + history.append( + HistorySample( + frame_id=f"e34-frame-{frame_index}", + evidence_time_ns=max(0, int(round(session_seconds * 1_000_000_000))), + centroid_xyz_m=_vector3( + document.get("centroid_map_xyz_m"), + "E34 history centroid", + ), + ) + ) + return tuple(history) + + +def _cells( + projection: TemporalFrameProjection, + component: dict[str, object], +) -> tuple[GridCell, ...]: + start = _integer(component.get("cell_row_start"), "E34 cell start") + count = _integer(component.get("cell_row_count"), "E34 cell count") + if start < 0 or count < 0 or start + count > int(projection.cell_rows.shape[0]): + raise PerceptionAdapterError("E34 component cell range is invalid") + rows = projection.cell_rows[start : start + count] + return tuple(GridCell(int(row[0]), int(row[1]), int(row[2])) for row in rows) + + +def _semantic_labels(value: object) -> tuple[str, ...]: + document = _object(value, "E34 semantic provenance") + labels = document.get("labels") + if not isinstance(labels, list): + raise PerceptionAdapterError("E34 semantic labels must be an array") + return tuple(_safe_reason(label) for label in labels) + + +def _safe_reason(value: object) -> str: + if not isinstance(value, str) or not value: + raise PerceptionAdapterError("E34 reason must be a nonempty string") + 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 PerceptionAdapterError(f"{label} must be an object") + return value + + +def _integer(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool): + raise PerceptionAdapterError(f"{label} must be an integer") + return value + + +def _number(value: object, label: str) -> float: + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(float(value)) + ): + raise PerceptionAdapterError(f"{label} must be finite") + return float(value) + + +def _vector3(value: object, label: str) -> tuple[float, float, float]: + if not isinstance(value, list) or len(value) != 3: + raise PerceptionAdapterError(f"{label} must contain three values") + return (_number(value[0], label), _number(value[1], label), _number(value[2], label)) diff --git a/src/k1link/perception/baseline.py b/src/k1link/perception/baseline.py new file mode 100644 index 0000000..2adb702 --- /dev/null +++ b/src/k1link/perception/baseline.py @@ -0,0 +1,287 @@ +"""Executable Milestone 4 baseline binding. + +The profile points at immutable local evidence without copying large or sensitive +artifacts into Git. Verification is explicit: missing evidence is a failure, not +an invitation to silently select a different source or experiment result. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +BASELINE_SCHEMA: Final = "missioncore.perception-m4-baseline/v1" +REUSE_INVENTORY_SCHEMA: Final = "missioncore.perception-reuse-inventory/v1" +BASELINE_PROFILE_ID: Final = "m4-ravnoves00-recorded-realtime/v1" +BASELINE_SOURCE_ID: Final = "RAVNOVES00" +BASELINE_SESSION_ID: Final = "20260720T065719Z_viewer_live" + +_SHA256 = re.compile(r"^[a-f0-9]{64}$") +_EXPECTED_EVIDENCE_ROLES: Final = { + "source-fusion", + "track-geometry", + "source-paced-worker", + "temporal-occupied", + "degradation-recovery", + "raw-fisheye-detector-capacity", +} +_BASELINE_KEYS: Final = { + "schema_version", + "profile_id", + "source", + "calibration", + "detector", + "evidence", + "authority", + "non_goals", + "rollback", +} +_EVIDENCE_KEYS: Final = { + "role", + "relative_path", + "schema_version", + "result_id", + "identity_sha256", + "file_sha256", +} + + +class BaselineContractError(ValueError): + """The recorded-realtime baseline is ambiguous, mutated or incomplete.""" + + +@dataclass(frozen=True, slots=True) +class BaselineEvidence: + role: str + relative_path: str + schema_version: str + result_id: str + identity_sha256: str + file_sha256: str + + +@dataclass(frozen=True, slots=True) +class BaselineProfile: + path: Path + document: dict[str, object] + evidence: tuple[BaselineEvidence, ...] + + +@dataclass(frozen=True, slots=True) +class BaselineVerification: + profile_id: str + source_id: str + session_id: str + verified_paths: tuple[str, ...] + + +def load_m4_baseline(path: Path) -> BaselineProfile: + """Load and fail-closed validate the one admitted M4 baseline profile.""" + + document = _read_object(path) + _exact_keys(document, _BASELINE_KEYS, "baseline") + if document.get("schema_version") != BASELINE_SCHEMA: + raise BaselineContractError("baseline schema is incompatible") + if document.get("profile_id") != BASELINE_PROFILE_ID: + raise BaselineContractError("baseline profile identity changed") + + source = _object(document.get("source"), "source") + if source.get("source_id") != BASELINE_SOURCE_ID: + raise BaselineContractError("M4 source must remain RAVNOVES00") + if source.get("session_id") != BASELINE_SESSION_ID: + raise BaselineContractError("M4 source session identity changed") + modalities = _string_array(source.get("modalities"), "source modalities") + if set(modalities) != {"image", "registered-point-increment", "pose"}: + raise BaselineContractError("baseline source must bind image, points and pose") + if _integer(source.get("frame_count"), "source frame count") != 4489: + raise BaselineContractError("baseline source frame count changed") + + authority = _object(document.get("authority"), "authority") + if authority.get("mode") != "replay-simulated": + raise BaselineContractError("M4 authority must remain replay-simulated") + for key in ( + "ground_truth", + "physical_live", + "physical_collision_accepted", + "commands_enabled", + "actuation_allowed", + "navigation_or_safety_accepted", + ): + if authority.get(key) is not False: + raise BaselineContractError(f"baseline authority {key} must remain false") + + evidence_items = document.get("evidence") + if not isinstance(evidence_items, list): + raise BaselineContractError("baseline evidence must be an array") + evidence = tuple(_evidence(item) for item in evidence_items) + roles = [item.role for item in evidence] + if len(set(roles)) != len(roles) or set(roles) != _EXPECTED_EVIDENCE_ROLES: + raise BaselineContractError("baseline evidence roles are incomplete or duplicated") + paths = [item.relative_path for item in evidence] + if len(set(paths)) != len(paths): + raise BaselineContractError("baseline evidence paths must be unique") + + rollback = _object(document.get("rollback"), "rollback") + if rollback.get("worker_id") != "worker-006": + raise BaselineContractError("rollback worker identity changed") + if rollback.get("worker_node") != "DESKTOP-OPJ8J04": + raise BaselineContractError("rollback worker node changed") + entrypoint = rollback.get("entrypoint") + if not isinstance(entrypoint, str) or "run_e15_shadow_inference.py serve" not in entrypoint: + raise BaselineContractError("rollback E15 process identity is missing") + _digest(rollback.get("runner_sha256"), "rollback runner digest") + _digest(rollback.get("orchestrator_sha256"), "rollback orchestrator digest") + + return BaselineProfile(path=path, document=document, evidence=evidence) + + +def verify_m4_baseline(repository_root: Path, profile: BaselineProfile) -> BaselineVerification: + """Resolve every immutable evidence document and verify its exact digest.""" + + root = repository_root.resolve() + verified: list[str] = [] + for item in profile.evidence: + evidence_path = (root / item.relative_path).resolve() + if root not in evidence_path.parents: + raise BaselineContractError("baseline evidence escapes the repository root") + if not evidence_path.is_file(): + raise BaselineContractError(f"baseline evidence is missing: {item.relative_path}") + if _file_sha256(evidence_path) != item.file_sha256: + raise BaselineContractError(f"baseline evidence digest changed: {item.role}") + evidence_document = _read_object(evidence_path) + if evidence_document.get("schema_version") != item.schema_version: + raise BaselineContractError(f"baseline evidence schema changed: {item.role}") + if evidence_document.get("identity_sha256") != item.identity_sha256: + raise BaselineContractError(f"baseline evidence identity changed: {item.role}") + result_id = evidence_document.get("result_id") + if result_id is None and item.role == "source-fusion": + result_id = evidence_path.parent.name + if result_id != item.result_id: + raise BaselineContractError(f"baseline evidence result changed: {item.role}") + verified.append(item.relative_path) + + source = _object(profile.document.get("source"), "source") + return BaselineVerification( + profile_id=BASELINE_PROFILE_ID, + source_id=_string(source.get("source_id"), "source id"), + session_id=_string(source.get("session_id"), "session id"), + verified_paths=tuple(verified), + ) + + +def validate_reuse_inventory(path: Path) -> dict[str, object]: + """Validate the M4 primitive/wrapper split used by architecture tests.""" + + document = _read_object(path) + _exact_keys( + document, + { + "schema_version", + "profile_id", + "reusable_primitives", + "historical_wrappers", + "rules", + }, + "reuse inventory", + ) + if document.get("schema_version") != REUSE_INVENTORY_SCHEMA: + raise BaselineContractError("reuse inventory schema is incompatible") + reusable = document.get("reusable_primitives") + wrappers = document.get("historical_wrappers") + if not isinstance(reusable, list) or not reusable: + raise BaselineContractError("reuse inventory has no admitted primitives") + if not isinstance(wrappers, list) or not wrappers: + raise BaselineContractError("reuse inventory has no historical wrappers") + reusable_modules = {_module(item, "reusable primitive") for item in reusable} + wrapper_modules = {_module(item, "historical wrapper") for item in wrappers} + if reusable_modules & wrapper_modules: + raise BaselineContractError("a module cannot be reusable and historical") + rules = _object(document.get("rules"), "reuse rules") + expected_rules = { + "historical_wrappers_are_product_dependencies": False, + "contracts_may_import_compute": False, + "graph_may_import_experiment_modules": False, + "providers_may_import_admitted_primitives": True, + "bulk_legacy_migration_required": False, + } + if rules != expected_rules: + raise BaselineContractError("reuse dependency rules changed") + return document + + +def _evidence(value: object) -> BaselineEvidence: + document = _object(value, "evidence item") + _exact_keys(document, _EVIDENCE_KEYS, "evidence item") + relative_path = _string(document.get("relative_path"), "evidence path") + path = Path(relative_path) + if path.is_absolute() or ".." in path.parts or path.suffix != ".json": + raise BaselineContractError("evidence path must be a relative JSON path") + return BaselineEvidence( + role=_string(document.get("role"), "evidence role"), + relative_path=relative_path, + schema_version=_string(document.get("schema_version"), "evidence schema"), + result_id=_string(document.get("result_id"), "evidence result id"), + identity_sha256=_digest(document.get("identity_sha256"), "evidence identity"), + file_sha256=_digest(document.get("file_sha256"), "evidence file digest"), + ) + + +def _module(value: object, label: str) -> str: + document = _object(value, label) + return _string(document.get("module"), f"{label} module") + + +def _read_object(path: Path) -> dict[str, object]: + try: + value = json.loads(path.read_text("utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise BaselineContractError(f"cannot read baseline document: {path}") from exc + return _object(value, str(path)) + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +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 BaselineContractError(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 BaselineContractError(f"{label} fields are incompatible") + + +def _string(value: object, label: str) -> str: + if not isinstance(value, str) or not value: + raise BaselineContractError(f"{label} must be a nonempty string") + return value + + +def _integer(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool): + raise BaselineContractError(f"{label} must be an integer") + return value + + +def _string_array(value: object, label: str) -> tuple[str, ...]: + if not isinstance(value, list): + raise BaselineContractError(f"{label} must be an array") + return tuple(_string(item, label) for item in value) + + +def _digest(value: object, label: str) -> str: + digest = _string(value, label) + if _SHA256.fullmatch(digest) is None: + raise BaselineContractError(f"{label} must be a SHA-256 digest") + return digest diff --git a/src/k1link/perception/contracts.py b/src/k1link/perception/contracts.py new file mode 100644 index 0000000..6cb6276 --- /dev/null +++ b/src/k1link/perception/contracts.py @@ -0,0 +1,1109 @@ +"""Strict object-centric perception contracts for Mission Core Milestone 4. + +These documents are the product boundary. They intentionally contain no LAB, +experiment, provider or web identity and fail closed on unknown JSON fields. +Semantic labels and provider tracklets are optional diagnostics; neither is an +occupancy identity or a downstream persistence contract. +""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass +from enum import StrEnum +from typing import Final + +SOURCE_ENVELOPE_SCHEMA: Final = "missioncore.source-envelope/v1" +OBJECT_PROPOSAL_SCHEMA: Final = "missioncore.object-proposal-2d/v1" +OBSTACLE_OBSERVATION_SCHEMA: Final = "missioncore.obstacle-observation/v1" +TEMPORAL_OBSTACLE_SCHEMA: Final = "missioncore.temporal-obstacle/v1" +LOCAL_OBSTACLE_MAP_SCHEMA: Final = "missioncore.local-obstacle-map/v1" +THREAT_ASSESSMENT_SCHEMA: Final = "missioncore.threat-assessment/v1" + +_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$") +class PerceptionContractError(ValueError): + """A value violates the source-neutral object perception contract.""" + + +class ClockBasis(StrEnum): + RECORDED_HOST = "recorded-host" + LIVE_HOST = "live-host" + DEVICE = "device" + + +class ModalityOutcome(StrEnum): + AVAILABLE = "available" + UNAVAILABLE = "unavailable" + DROPPED = "dropped" + STALE = "stale" + NOT_EXPECTED = "not-expected" + + +class EvidenceBasis(StrEnum): + CAMERA = "camera" + LIDAR = "lidar" + FUSED = "fused" + CONFLICT = "conflict" + + +class EvidenceCurrentness(StrEnum): + CURRENT = "current" + HELD = "held" + STALE = "stale" + UNAVAILABLE = "unavailable" + + +class TemporalState(StrEnum): + CURRENT = "current" + HELD = "held" + EXPIRED = "expired" + + +class MotionState(StrEnum): + MOVING = "moving" + STATIONARY = "stationary" + UNKNOWN = "unknown" + + +class QualificationState(StrEnum): + QUALIFIED = "qualified" + UNQUALIFIED = "unqualified" + + +class CorridorIntersection(StrEnum): + INTERSECTS = "intersects" + CLEAR = "clear" + UNKNOWN = "unknown" + + +class ThreatDecision(StrEnum): + THREAT = "threat" + NOT_THREAT = "not-threat" + UNKNOWN = "unknown" + + +class ThreatAuthority(StrEnum): + REPLAY_SIMULATED = "replay-simulated" + + +@dataclass(frozen=True, slots=True) +class TimestampBundle: + utc_ns: int + monotonic_ns: int + source_ns: int + clock_basis: ClockBasis + + def __post_init__(self) -> None: + _nonnegative_integer(self.utc_ns, "UTC timestamp") + _nonnegative_integer(self.monotonic_ns, "monotonic timestamp") + _nonnegative_integer(self.source_ns, "source timestamp") + + def to_dict(self) -> dict[str, object]: + return { + "utc_ns": self.utc_ns, + "monotonic_ns": self.monotonic_ns, + "source_ns": self.source_ns, + "clock_basis": self.clock_basis.value, + } + + @classmethod + def from_dict(cls, value: object) -> TimestampBundle: + document = _object(value, "timestamps") + _exact_keys( + document, + {"utc_ns", "monotonic_ns", "source_ns", "clock_basis"}, + "timestamps", + ) + return cls( + utc_ns=_integer(document, "utc_ns"), + monotonic_ns=_integer(document, "monotonic_ns"), + source_ns=_integer(document, "source_ns"), + clock_basis=_enum(ClockBasis, document.get("clock_basis"), "clock basis"), + ) + + +@dataclass(frozen=True, slots=True) +class ModalityStatus: + available: bool + outcome: ModalityOutcome + reason: str + + def __post_init__(self) -> None: + _identifier(self.reason, "modality reason") + if self.available is not (self.outcome is ModalityOutcome.AVAILABLE): + raise PerceptionContractError("modality availability and outcome disagree") + + def to_dict(self) -> dict[str, object]: + return { + "available": self.available, + "outcome": self.outcome.value, + "reason": self.reason, + } + + @classmethod + def from_dict(cls, value: object) -> ModalityStatus: + document = _object(value, "modality status") + _exact_keys(document, {"available", "outcome", "reason"}, "modality status") + return cls( + available=_boolean(document, "available"), + outcome=_enum(ModalityOutcome, document.get("outcome"), "modality outcome"), + reason=_string(document, "reason"), + ) + + +@dataclass(frozen=True, slots=True) +class BoundingRegion2D: + x_min: float + y_min: float + x_max: float + y_max: float + + def __post_init__(self) -> None: + values = tuple(_finite_number(value, "bounding region") for value in self.as_tuple()) + if values[0] < 0.0 or values[1] < 0.0 or values[2] <= values[0] or values[3] <= values[1]: + raise PerceptionContractError("bounding region is invalid") + + def as_tuple(self) -> tuple[float, float, float, float]: + return (self.x_min, self.y_min, self.x_max, self.y_max) + + def to_dict(self) -> dict[str, object]: + return { + "x_min": self.x_min, + "y_min": self.y_min, + "x_max": self.x_max, + "y_max": self.y_max, + } + + @classmethod + def from_dict(cls, value: object) -> BoundingRegion2D: + document = _object(value, "bounding region") + _exact_keys(document, {"x_min", "y_min", "x_max", "y_max"}, "bounding region") + return cls( + x_min=_number(document, "x_min"), + y_min=_number(document, "y_min"), + x_max=_number(document, "x_max"), + y_max=_number(document, "y_max"), + ) + + +@dataclass(frozen=True, slots=True) +class MetricGeometry: + coordinate_frame: str + centroid_xyz_m: tuple[float, float, float] + range_m: float + covariance_diagonal_m2: tuple[float, float, float] + + def __post_init__(self) -> None: + _identifier(self.coordinate_frame, "coordinate frame") + _vector3(self.centroid_xyz_m, "metric centroid") + if _finite_number(self.range_m, "metric range") <= 0.0: + raise PerceptionContractError("metric range must be positive") + covariance = _vector3(self.covariance_diagonal_m2, "metric covariance") + if any(value < 0.0 for value in covariance): + raise PerceptionContractError("metric covariance must be nonnegative") + + def to_dict(self) -> dict[str, object]: + return { + "coordinate_frame": self.coordinate_frame, + "centroid_xyz_m": list(self.centroid_xyz_m), + "range_m": self.range_m, + "covariance_diagonal_m2": list(self.covariance_diagonal_m2), + } + + @classmethod + def from_dict(cls, value: object) -> MetricGeometry: + document = _object(value, "metric geometry") + _exact_keys( + document, + {"coordinate_frame", "centroid_xyz_m", "range_m", "covariance_diagonal_m2"}, + "metric geometry", + ) + return cls( + coordinate_frame=_string(document, "coordinate_frame"), + centroid_xyz_m=_number_vector3(document.get("centroid_xyz_m"), "metric centroid"), + range_m=_number(document, "range_m"), + covariance_diagonal_m2=_number_vector3( + document.get("covariance_diagonal_m2"), + "metric covariance", + ), + ) + + +@dataclass(frozen=True, slots=True) +class FalseAuthority: + ground_truth: bool = False + physical_live: bool = False + physical_collision_accepted: bool = False + commands_enabled: bool = False + actuation_allowed: bool = False + navigation_or_safety_accepted: bool = False + + def __post_init__(self) -> None: + if any( + ( + self.ground_truth, + self.physical_live, + self.physical_collision_accepted, + self.commands_enabled, + self.actuation_allowed, + self.navigation_or_safety_accepted, + ) + ): + raise PerceptionContractError( + "Milestone 4 cannot publish physical or command authority" + ) + + def to_dict(self) -> dict[str, object]: + return { + "ground_truth": self.ground_truth, + "physical_live": self.physical_live, + "physical_collision_accepted": self.physical_collision_accepted, + "commands_enabled": self.commands_enabled, + "actuation_allowed": self.actuation_allowed, + "navigation_or_safety_accepted": self.navigation_or_safety_accepted, + } + + @classmethod + def from_dict(cls, value: object) -> FalseAuthority: + document = _object(value, "authority") + keys = { + "ground_truth", + "physical_live", + "physical_collision_accepted", + "commands_enabled", + "actuation_allowed", + "navigation_or_safety_accepted", + } + _exact_keys(document, keys, "authority") + return cls(**{key: _boolean(document, key) for key in keys}) + + +@dataclass(frozen=True, slots=True) +class SourceEnvelope: + source_id: str + session_id: str + frame_id: str + sequence: int + timestamps: TimestampBundle + source_age_ns: int + binding_reason: str + calibration_id: str + representation_id: str + image: ModalityStatus + registered_point_increment: ModalityStatus + pose: ModalityStatus + + def __post_init__(self) -> None: + for value, label in ( + (self.source_id, "source id"), + (self.session_id, "session id"), + (self.frame_id, "frame id"), + (self.binding_reason, "binding reason"), + (self.calibration_id, "calibration id"), + (self.representation_id, "representation id"), + ): + _identifier(value, label) + _nonnegative_integer(self.sequence, "source sequence") + _nonnegative_integer(self.source_age_ns, "source age") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": SOURCE_ENVELOPE_SCHEMA, + "source_id": self.source_id, + "session_id": self.session_id, + "frame_id": self.frame_id, + "sequence": self.sequence, + "timestamps": self.timestamps.to_dict(), + "source_age_ns": self.source_age_ns, + "binding_reason": self.binding_reason, + "calibration_id": self.calibration_id, + "representation_id": self.representation_id, + "image": self.image.to_dict(), + "registered_point_increment": self.registered_point_increment.to_dict(), + "pose": self.pose.to_dict(), + } + + @classmethod + def from_dict(cls, value: object) -> SourceEnvelope: + document = _contract(value, SOURCE_ENVELOPE_SCHEMA, { + "source_id", "session_id", "frame_id", "sequence", "timestamps", + "source_age_ns", "binding_reason", "calibration_id", "representation_id", + "image", "registered_point_increment", "pose", + }, "source envelope") + return cls( + source_id=_string(document, "source_id"), + session_id=_string(document, "session_id"), + frame_id=_string(document, "frame_id"), + sequence=_integer(document, "sequence"), + timestamps=TimestampBundle.from_dict(document.get("timestamps")), + source_age_ns=_integer(document, "source_age_ns"), + binding_reason=_string(document, "binding_reason"), + calibration_id=_string(document, "calibration_id"), + representation_id=_string(document, "representation_id"), + image=ModalityStatus.from_dict(document.get("image")), + registered_point_increment=ModalityStatus.from_dict( + document.get("registered_point_increment") + ), + pose=ModalityStatus.from_dict(document.get("pose")), + ) + + +@dataclass(frozen=True, slots=True) +class ObjectProposal2D: + proposal_id: str + source_id: str + frame_id: str + region: BoundingRegion2D + objectness: float + provider_id: str + model_id: str + preprocess_id: str + semantic_hint: str | None = None + provider_tracklet: str | None = None + + def __post_init__(self) -> None: + for value, label in ( + (self.proposal_id, "proposal id"), + (self.source_id, "proposal source id"), + (self.frame_id, "proposal frame id"), + (self.provider_id, "detector provider id"), + (self.model_id, "detector model id"), + (self.preprocess_id, "preprocess id"), + ): + _identifier(value, label) + confidence = _finite_number(self.objectness, "objectness") + if not 0.0 <= confidence <= 1.0: + raise PerceptionContractError("objectness must be within [0, 1]") + _optional_identifier(self.semantic_hint, "semantic hint") + _optional_identifier(self.provider_tracklet, "provider tracklet") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": OBJECT_PROPOSAL_SCHEMA, + "proposal_id": self.proposal_id, + "source_id": self.source_id, + "frame_id": self.frame_id, + "region": self.region.to_dict(), + "objectness": self.objectness, + "provider_id": self.provider_id, + "model_id": self.model_id, + "preprocess_id": self.preprocess_id, + "semantic_hint": self.semantic_hint, + "provider_tracklet": self.provider_tracklet, + } + + @classmethod + def from_dict(cls, value: object) -> ObjectProposal2D: + document = _contract(value, OBJECT_PROPOSAL_SCHEMA, { + "proposal_id", "source_id", "frame_id", "region", "objectness", + "provider_id", "model_id", "preprocess_id", "semantic_hint", + "provider_tracklet", + }, "object proposal") + return cls( + proposal_id=_string(document, "proposal_id"), + source_id=_string(document, "source_id"), + frame_id=_string(document, "frame_id"), + region=BoundingRegion2D.from_dict(document.get("region")), + objectness=_number(document, "objectness"), + provider_id=_string(document, "provider_id"), + model_id=_string(document, "model_id"), + preprocess_id=_string(document, "preprocess_id"), + semantic_hint=_optional_string(document.get("semantic_hint"), "semantic hint"), + provider_tracklet=_optional_string( + document.get("provider_tracklet"), "provider tracklet" + ), + ) + + +@dataclass(frozen=True, slots=True) +class ObstacleObservation: + observation_id: str + occupancy_key: str + source_id: str + frame_id: str + evidence_time_ns: int + basis: EvidenceBasis + currentness: EvidenceCurrentness + occupied_support: bool + source_point_ids: tuple[int, ...] + metric_geometry: MetricGeometry | None + proposal_ids: tuple[str, ...] + semantic_hint: str | None + reason_codes: tuple[str, ...] + authority: FalseAuthority = FalseAuthority() + + def __post_init__(self) -> None: + for value, label in ( + (self.observation_id, "observation id"), + (self.occupancy_key, "occupancy key"), + (self.source_id, "observation source id"), + (self.frame_id, "observation frame id"), + ): + _identifier(value, label) + _nonnegative_integer(self.evidence_time_ns, "evidence time") + _unique_nonnegative_integers(self.source_point_ids, "source point ids") + _unique_identifiers(self.proposal_ids, "proposal ids", allow_empty=True) + _unique_identifiers(self.reason_codes, "observation reason codes") + _optional_identifier(self.semantic_hint, "semantic hint") + if not isinstance(self.occupied_support, bool): + raise PerceptionContractError("occupied support must be boolean") + if self.currentness is not EvidenceCurrentness.CURRENT and ( + self.occupied_support or self.source_point_ids or self.metric_geometry is not None + ): + raise PerceptionContractError( + "non-current evidence cannot publish current geometry" + ) + if self.basis is EvidenceBasis.CAMERA: + if self.occupied_support or self.source_point_ids or self.metric_geometry is not None: + raise PerceptionContractError("camera-only evidence is non-metric uncertainty") + if not self.proposal_ids: + raise PerceptionContractError("camera evidence requires an object proposal") + elif self.basis is EvidenceBasis.CONFLICT: + if self.occupied_support or self.source_point_ids or self.metric_geometry is not None: + raise PerceptionContractError( + "conflicting evidence cannot publish metric occupancy" + ) + elif self.currentness is EvidenceCurrentness.CURRENT and ( + not self.occupied_support + or not self.source_point_ids + or self.metric_geometry is None + ): + raise PerceptionContractError( + "current LiDAR/fused occupancy requires qualified points" + ) + + @property + def occupancy_identity(self) -> str: + """Identity deliberately excludes semantic labels and provider tracklets.""" + + return self.occupancy_key + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": OBSTACLE_OBSERVATION_SCHEMA, + "observation_id": self.observation_id, + "occupancy_key": self.occupancy_key, + "source_id": self.source_id, + "frame_id": self.frame_id, + "evidence_time_ns": self.evidence_time_ns, + "basis": self.basis.value, + "currentness": self.currentness.value, + "occupied_support": self.occupied_support, + "source_point_ids": list(self.source_point_ids), + "metric_geometry": ( + None if self.metric_geometry is None else self.metric_geometry.to_dict() + ), + "proposal_ids": list(self.proposal_ids), + "semantic_hint": self.semantic_hint, + "reason_codes": list(self.reason_codes), + "authority": self.authority.to_dict(), + } + + @classmethod + def from_dict(cls, value: object) -> ObstacleObservation: + document = _contract(value, OBSTACLE_OBSERVATION_SCHEMA, { + "observation_id", "occupancy_key", "source_id", "frame_id", + "evidence_time_ns", "basis", "currentness", "occupied_support", + "source_point_ids", "metric_geometry", "proposal_ids", "semantic_hint", + "reason_codes", "authority", + }, "obstacle observation") + metric = document.get("metric_geometry") + return cls( + observation_id=_string(document, "observation_id"), + occupancy_key=_string(document, "occupancy_key"), + source_id=_string(document, "source_id"), + frame_id=_string(document, "frame_id"), + evidence_time_ns=_integer(document, "evidence_time_ns"), + basis=_enum(EvidenceBasis, document.get("basis"), "evidence basis"), + currentness=_enum( + EvidenceCurrentness, document.get("currentness"), "evidence currentness" + ), + occupied_support=_boolean(document, "occupied_support"), + source_point_ids=_integer_tuple(document.get("source_point_ids"), "source point ids"), + metric_geometry=None if metric is None else MetricGeometry.from_dict(metric), + proposal_ids=_string_tuple(document.get("proposal_ids"), "proposal ids"), + semantic_hint=_optional_string(document.get("semantic_hint"), "semantic hint"), + reason_codes=_string_tuple(document.get("reason_codes"), "reason codes"), + authority=FalseAuthority.from_dict(document.get("authority")), + ) + + +@dataclass(frozen=True, slots=True) +class HistorySample: + frame_id: str + evidence_time_ns: int + centroid_xyz_m: tuple[float, float, float] + + def __post_init__(self) -> None: + _identifier(self.frame_id, "history frame id") + _nonnegative_integer(self.evidence_time_ns, "history evidence time") + _vector3(self.centroid_xyz_m, "history centroid") + + def to_dict(self) -> dict[str, object]: + return { + "frame_id": self.frame_id, + "evidence_time_ns": self.evidence_time_ns, + "centroid_xyz_m": list(self.centroid_xyz_m), + } + + @classmethod + def from_dict(cls, value: object) -> HistorySample: + document = _object(value, "history sample") + _exact_keys(document, {"frame_id", "evidence_time_ns", "centroid_xyz_m"}, "history sample") + return cls( + frame_id=_string(document, "frame_id"), + evidence_time_ns=_integer(document, "evidence_time_ns"), + centroid_xyz_m=_number_vector3(document.get("centroid_xyz_m"), "history centroid"), + ) + + +@dataclass(frozen=True, slots=True) +class GridCell: + x: int + y: int + z: int + + def __post_init__(self) -> None: + for value in (self.x, self.y, self.z): + if not isinstance(value, int) or isinstance(value, bool): + raise PerceptionContractError("grid cell coordinates must be integers") + + def to_dict(self) -> dict[str, object]: + return {"x": self.x, "y": self.y, "z": self.z} + + @classmethod + def from_dict(cls, value: object) -> GridCell: + document = _object(value, "grid cell") + _exact_keys(document, {"x", "y", "z"}, "grid cell") + return cls(x=_integer(document, "x"), y=_integer(document, "y"), z=_integer(document, "z")) + + +@dataclass(frozen=True, slots=True) +class TemporalObstacle: + component_id: str + identity_scope: str + state: TemporalState + ttl_ns: int + last_hit_ns: int + age_ns: int + association_basis: str + history: tuple[HistorySample, ...] + cells: tuple[GridCell, ...] + coordinate_frame: str | None + last_centroid_xyz_m: tuple[float, float, float] | None + motion: MotionState + motion_confidence: float + motion_reason: str + semantic_hint: str | None = None + + def __post_init__(self) -> None: + _identifier(self.component_id, "temporal component id") + if self.identity_scope != "ephemeral": + raise PerceptionContractError("temporal component identity must remain ephemeral") + _positive_integer(self.ttl_ns, "temporal TTL") + _nonnegative_integer(self.last_hit_ns, "last hit time") + _nonnegative_integer(self.age_ns, "temporal age") + _identifier(self.association_basis, "association basis") + _identifier(self.motion_reason, "motion reason") + _optional_identifier(self.semantic_hint, "semantic hint") + confidence = _finite_number(self.motion_confidence, "motion confidence") + if not 0.0 <= confidence <= 1.0: + raise PerceptionContractError("motion confidence must be within [0, 1]") + if not self.history or len(self.history) > 32: + raise PerceptionContractError("temporal history must contain 1..32 samples") + if len(set(self.cells)) != len(self.cells): + raise PerceptionContractError("temporal cells must be unique") + if self.state is TemporalState.CURRENT: + if self.age_ns > self.ttl_ns or not self.cells: + raise PerceptionContractError("current temporal occupancy requires bounded cells") + elif self.state is TemporalState.HELD: + if not 0 < self.age_ns <= self.ttl_ns or not self.cells: + raise PerceptionContractError("held temporal state must remain within TTL") + elif self.cells: + raise PerceptionContractError( + "expired temporal components cannot publish occupied cells" + ) + if self.cells: + if self.coordinate_frame is None or self.last_centroid_xyz_m is None: + raise PerceptionContractError("temporal cells require their last metric geometry") + _identifier(self.coordinate_frame, "temporal coordinate frame") + _vector3(self.last_centroid_xyz_m, "temporal centroid") + elif self.coordinate_frame is not None or self.last_centroid_xyz_m is not None: + raise PerceptionContractError("temporal metric geometry requires occupied cells") + if self.motion is MotionState.UNKNOWN and confidence != 0.0: + raise PerceptionContractError("unknown motion must have zero confidence") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": TEMPORAL_OBSTACLE_SCHEMA, + "component_id": self.component_id, + "identity_scope": self.identity_scope, + "state": self.state.value, + "ttl_ns": self.ttl_ns, + "last_hit_ns": self.last_hit_ns, + "age_ns": self.age_ns, + "association_basis": self.association_basis, + "history": [item.to_dict() for item in self.history], + "cells": [item.to_dict() for item in self.cells], + "coordinate_frame": self.coordinate_frame, + "last_centroid_xyz_m": ( + None if self.last_centroid_xyz_m is None else list(self.last_centroid_xyz_m) + ), + "motion": self.motion.value, + "motion_confidence": self.motion_confidence, + "motion_reason": self.motion_reason, + "semantic_hint": self.semantic_hint, + } + + @classmethod + def from_dict(cls, value: object) -> TemporalObstacle: + document = _contract(value, TEMPORAL_OBSTACLE_SCHEMA, { + "component_id", "identity_scope", "state", "ttl_ns", "last_hit_ns", + "age_ns", "association_basis", "history", "cells", "coordinate_frame", + "last_centroid_xyz_m", "motion", "motion_confidence", "motion_reason", + "semantic_hint", + }, "temporal obstacle") + centroid = document.get("last_centroid_xyz_m") + return cls( + component_id=_string(document, "component_id"), + identity_scope=_string(document, "identity_scope"), + state=_enum(TemporalState, document.get("state"), "temporal state"), + ttl_ns=_integer(document, "ttl_ns"), + last_hit_ns=_integer(document, "last_hit_ns"), + age_ns=_integer(document, "age_ns"), + association_basis=_string(document, "association_basis"), + history=tuple(HistorySample.from_dict(item) for item in _array(document, "history")), + cells=tuple(GridCell.from_dict(item) for item in _array(document, "cells")), + coordinate_frame=_optional_string(document.get("coordinate_frame"), "coordinate frame"), + last_centroid_xyz_m=( + None if centroid is None else _number_vector3(centroid, "temporal centroid") + ), + motion=_enum(MotionState, document.get("motion"), "motion state"), + motion_confidence=_number(document, "motion_confidence"), + motion_reason=_string(document, "motion_reason"), + semantic_hint=_optional_string(document.get("semantic_hint"), "semantic hint"), + ) + + +@dataclass(frozen=True, slots=True) +class SourceAccounting: + source_envelopes: int + terminal_outcomes: int + dropped: int + failed: int + + def __post_init__(self) -> None: + for value, label in ( + (self.source_envelopes, "source envelope count"), + (self.terminal_outcomes, "terminal outcome count"), + (self.dropped, "drop count"), + (self.failed, "failure count"), + ): + _nonnegative_integer(value, label) + if self.terminal_outcomes + self.dropped + self.failed != self.source_envelopes: + raise PerceptionContractError("source accounting is not closed") + + def to_dict(self) -> dict[str, object]: + return { + "source_envelopes": self.source_envelopes, + "terminal_outcomes": self.terminal_outcomes, + "dropped": self.dropped, + "failed": self.failed, + } + + @classmethod + def from_dict(cls, value: object) -> SourceAccounting: + document = _object(value, "source accounting") + _exact_keys( + document, + {"source_envelopes", "terminal_outcomes", "dropped", "failed"}, + "source accounting", + ) + return cls( + source_envelopes=_integer(document, "source_envelopes"), + terminal_outcomes=_integer(document, "terminal_outcomes"), + dropped=_integer(document, "dropped"), + failed=_integer(document, "failed"), + ) + + +@dataclass(frozen=True, slots=True) +class LocalObstacleMap: + source_id: str + session_id: str + frame_id: str + graph_id: str + generated_monotonic_ns: int + output_age_ns: int + occupied: tuple[TemporalObstacle, ...] + unknown: tuple[TemporalObstacle, ...] + camera_uncertainty: tuple[ObjectProposal2D, ...] + accounting: SourceAccounting + free_space_claimed: bool = False + authority: FalseAuthority = FalseAuthority() + + def __post_init__(self) -> None: + for value, label in ( + (self.source_id, "map source id"), + (self.session_id, "map session id"), + (self.frame_id, "map frame id"), + (self.graph_id, "graph id"), + ): + _identifier(value, label) + _nonnegative_integer(self.generated_monotonic_ns, "map generation time") + _nonnegative_integer(self.output_age_ns, "map output age") + if self.free_space_claimed: + raise PerceptionContractError("Milestone 4 cannot publish implicit free space") + if any(item.state is not TemporalState.CURRENT for item in self.occupied): + raise PerceptionContractError("occupied map entries must be current") + if any(item.state is TemporalState.CURRENT for item in self.unknown): + raise PerceptionContractError("held or expired map entries must remain unknown") + component_ids = [item.component_id for item in (*self.occupied, *self.unknown)] + if len(set(component_ids)) != len(component_ids): + raise PerceptionContractError("map component identities must be unique") + proposal_ids = [item.proposal_id for item in self.camera_uncertainty] + if len(set(proposal_ids)) != len(proposal_ids): + raise PerceptionContractError("camera uncertainty proposals must be unique") + if any( + item.source_id != self.source_id or item.frame_id != self.frame_id + for item in self.camera_uncertainty + ): + raise PerceptionContractError("camera uncertainty is bound to another source frame") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": LOCAL_OBSTACLE_MAP_SCHEMA, + "source_id": self.source_id, + "session_id": self.session_id, + "frame_id": self.frame_id, + "graph_id": self.graph_id, + "generated_monotonic_ns": self.generated_monotonic_ns, + "output_age_ns": self.output_age_ns, + "occupied": [item.to_dict() for item in self.occupied], + "unknown": [item.to_dict() for item in self.unknown], + "camera_uncertainty": [item.to_dict() for item in self.camera_uncertainty], + "accounting": self.accounting.to_dict(), + "free_space_claimed": self.free_space_claimed, + "authority": self.authority.to_dict(), + } + + @classmethod + def from_dict(cls, value: object) -> LocalObstacleMap: + document = _contract(value, LOCAL_OBSTACLE_MAP_SCHEMA, { + "source_id", "session_id", "frame_id", "graph_id", + "generated_monotonic_ns", "output_age_ns", "occupied", "unknown", + "camera_uncertainty", "accounting", "free_space_claimed", "authority", + }, "local obstacle map") + return cls( + source_id=_string(document, "source_id"), + session_id=_string(document, "session_id"), + frame_id=_string(document, "frame_id"), + graph_id=_string(document, "graph_id"), + generated_monotonic_ns=_integer(document, "generated_monotonic_ns"), + output_age_ns=_integer(document, "output_age_ns"), + occupied=tuple( + TemporalObstacle.from_dict(item) for item in _array(document, "occupied") + ), + unknown=tuple(TemporalObstacle.from_dict(item) for item in _array(document, "unknown")), + camera_uncertainty=tuple( + ObjectProposal2D.from_dict(item) for item in _array(document, "camera_uncertainty") + ), + accounting=SourceAccounting.from_dict(document.get("accounting")), + free_space_claimed=_boolean(document, "free_space_claimed"), + authority=FalseAuthority.from_dict(document.get("authority")), + ) + + +@dataclass(frozen=True, slots=True) +class ThreatAssessment: + assessment_id: str + component_id: str + rig_profile_id: str + corridor_profile_id: str + qualification: QualificationState + relative_speed_mps: float | None + closest_approach_m: float | None + ttc_seconds: float | None + corridor_intersection: CorridorIntersection + decision: ThreatDecision + reason_codes: tuple[str, ...] + authority: ThreatAuthority = ThreatAuthority.REPLAY_SIMULATED + physical_collision_accepted: bool = False + actuation_allowed: bool = False + + def __post_init__(self) -> None: + for value, label in ( + (self.assessment_id, "assessment id"), + (self.component_id, "assessed component id"), + (self.rig_profile_id, "rig profile id"), + (self.corridor_profile_id, "corridor profile id"), + ): + _identifier(value, label) + _unique_identifiers(self.reason_codes, "threat reason codes") + relative_speed = _optional_finite(self.relative_speed_mps, "relative speed") + closest = _optional_finite(self.closest_approach_m, "closest approach") + ttc = _optional_finite(self.ttc_seconds, "TTC") + if closest is not None and closest < 0.0: + raise PerceptionContractError("closest approach must be nonnegative") + if ttc is not None and ttc < 0.0: + raise PerceptionContractError("TTC must be nonnegative") + if self.physical_collision_accepted or self.actuation_allowed: + raise PerceptionContractError("replay threat cannot authorize collision or actuation") + if self.authority is not ThreatAuthority.REPLAY_SIMULATED: + raise PerceptionContractError("Milestone 4 threat authority must be replay-simulated") + if self.qualification is QualificationState.UNQUALIFIED: + if self.decision is not ThreatDecision.UNKNOWN: + raise PerceptionContractError("unqualified threat evidence must remain unknown") + if any(value is not None for value in (relative_speed, closest, ttc)): + raise PerceptionContractError("unqualified threat cannot publish derived metrics") + if self.decision is ThreatDecision.THREAT and ( + self.qualification is not QualificationState.QUALIFIED + or self.corridor_intersection is not CorridorIntersection.INTERSECTS + ): + raise PerceptionContractError("threat requires qualified corridor intersection") + if self.decision is ThreatDecision.NOT_THREAT and ( + self.qualification is not QualificationState.QUALIFIED + or self.corridor_intersection is not CorridorIntersection.CLEAR + ): + raise PerceptionContractError("not-threat requires qualified corridor clearance") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": THREAT_ASSESSMENT_SCHEMA, + "assessment_id": self.assessment_id, + "component_id": self.component_id, + "rig_profile_id": self.rig_profile_id, + "corridor_profile_id": self.corridor_profile_id, + "qualification": self.qualification.value, + "relative_speed_mps": self.relative_speed_mps, + "closest_approach_m": self.closest_approach_m, + "ttc_seconds": self.ttc_seconds, + "corridor_intersection": self.corridor_intersection.value, + "decision": self.decision.value, + "reason_codes": list(self.reason_codes), + "authority": self.authority.value, + "physical_collision_accepted": self.physical_collision_accepted, + "actuation_allowed": self.actuation_allowed, + } + + @classmethod + def from_dict(cls, value: object) -> ThreatAssessment: + document = _contract(value, THREAT_ASSESSMENT_SCHEMA, { + "assessment_id", "component_id", "rig_profile_id", "corridor_profile_id", + "qualification", "relative_speed_mps", "closest_approach_m", "ttc_seconds", + "corridor_intersection", "decision", "reason_codes", "authority", + "physical_collision_accepted", "actuation_allowed", + }, "threat assessment") + return cls( + assessment_id=_string(document, "assessment_id"), + component_id=_string(document, "component_id"), + rig_profile_id=_string(document, "rig_profile_id"), + corridor_profile_id=_string(document, "corridor_profile_id"), + qualification=_enum( + QualificationState, document.get("qualification"), "qualification state" + ), + relative_speed_mps=_optional_number( + document.get("relative_speed_mps"), "relative speed" + ), + closest_approach_m=_optional_number( + document.get("closest_approach_m"), "closest approach" + ), + ttc_seconds=_optional_number(document.get("ttc_seconds"), "TTC"), + corridor_intersection=_enum( + CorridorIntersection, + document.get("corridor_intersection"), + "corridor intersection", + ), + decision=_enum(ThreatDecision, document.get("decision"), "threat decision"), + reason_codes=_string_tuple(document.get("reason_codes"), "reason codes"), + authority=_enum(ThreatAuthority, document.get("authority"), "threat authority"), + physical_collision_accepted=_boolean(document, "physical_collision_accepted"), + actuation_allowed=_boolean(document, "actuation_allowed"), + ) + + +def validate_exclusive_point_ownership( + observations: tuple[ObstacleObservation, ...], +) -> None: + """Reject one source point being claimed by more than one observation.""" + + owners: dict[int, str] = {} + for observation in observations: + for point_id in observation.source_point_ids: + previous = owners.setdefault(point_id, observation.observation_id) + if previous != observation.observation_id: + raise PerceptionContractError( + f"source point {point_id} has duplicate observation ownership" + ) + + +def _contract( + value: object, + schema: str, + fields: set[str], + label: str, +) -> dict[str, object]: + document = _object(value, label) + _exact_keys(document, {"schema_version", *fields}, label) + if document.get("schema_version") != schema: + raise PerceptionContractError(f"{label} schema is incompatible") + return document + + +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 PerceptionContractError(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 PerceptionContractError(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 PerceptionContractError(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 PerceptionContractError(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 _boolean(document: dict[str, object], key: str) -> bool: + value = document.get(key) + if not isinstance(value, bool): + raise PerceptionContractError(f"{key} must be boolean") + return value + + +def _integer(document: dict[str, object], key: str) -> int: + value = document.get(key) + if not isinstance(value, int) or isinstance(value, bool): + raise PerceptionContractError(f"{key} must be an integer") + return value + + +def _number(document: dict[str, object], key: str) -> float: + return _finite_number(document.get(key), key) + + +def _optional_number(value: object, label: str) -> float | None: + if value is None: + return None + return _finite_number(value, label) + + +def _finite_number(value: object, label: str) -> float: + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(float(value)) + ): + raise PerceptionContractError(f"{label} must be finite") + return float(value) + + +def _optional_finite(value: float | None, label: str) -> float | None: + return None if value is None else _finite_number(value, label) + + +def _nonnegative_integer(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise PerceptionContractError(f"{label} must be a nonnegative integer") + return value + + +def _positive_integer(value: object, label: str) -> int: + result = _nonnegative_integer(value, label) + if result == 0: + raise PerceptionContractError(f"{label} must be positive") + return result + + +def _identifier(value: str, label: str) -> str: + if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None: + raise PerceptionContractError(f"{label} is not a safe identifier") + return value + + +def _optional_identifier(value: str | None, label: str) -> None: + if value is not None: + _identifier(value, label) + + +def _unique_identifiers( + values: tuple[str, ...], + label: str, + *, + allow_empty: bool = False, +) -> None: + if (not values and not allow_empty) or len(set(values)) != len(values): + raise PerceptionContractError(f"{label} must be nonempty and unique") + for value in values: + _identifier(value, label) + + +def _unique_nonnegative_integers(values: tuple[int, ...], label: str) -> None: + if len(set(values)) != len(values): + raise PerceptionContractError(f"{label} must be unique") + for value in values: + _nonnegative_integer(value, label) + + +def _string_tuple(value: object, label: str) -> tuple[str, ...]: + if not isinstance(value, list): + raise PerceptionContractError(f"{label} must be an array") + return tuple(_string_value(item, label) for item in value) + + +def _integer_tuple(value: object, label: str) -> tuple[int, ...]: + if not isinstance(value, list): + raise PerceptionContractError(f"{label} must be an array") + result: list[int] = [] + for item in value: + if not isinstance(item, int) or isinstance(item, bool): + raise PerceptionContractError(f"{label} must contain integers") + result.append(item) + return tuple(result) + + +def _vector3(values: tuple[float, float, float], label: str) -> tuple[float, float, float]: + if len(values) != 3: + raise PerceptionContractError(f"{label} must contain three values") + return tuple(_finite_number(value, label) for value in values) # type: ignore[return-value] + + +def _number_vector3(value: object, label: str) -> tuple[float, float, float]: + if not isinstance(value, list) or len(value) != 3: + raise PerceptionContractError(f"{label} must contain three values") + values = tuple(_finite_number(item, label) for item in value) + return (values[0], values[1], values[2]) + + +def _enum[ENUM: StrEnum]( + enum_type: type[ENUM], + value: object, + label: str, +) -> ENUM: + if not isinstance(value, str): + raise PerceptionContractError(f"{label} must be a string") + try: + return enum_type(value) + except ValueError as exc: + raise PerceptionContractError(f"{label} is incompatible") from exc diff --git a/src/k1link/perception/providers.py b/src/k1link/perception/providers.py new file mode 100644 index 0000000..8a3702f --- /dev/null +++ b/src/k1link/perception/providers.py @@ -0,0 +1,334 @@ +"""Provider protocols and one versioned graph configuration contract.""" + +from __future__ import annotations + +import re +from collections.abc import Iterator +from dataclasses import dataclass +from enum import StrEnum +from typing import Final, Protocol + +from .contracts import ( + LocalObstacleMap, + ObjectProposal2D, + ObstacleObservation, + SourceEnvelope, + TemporalObstacle, + ThreatAssessment, +) + +REFERENCE_GRAPH_CONFIG_SCHEMA: Final = "missioncore.reference-perception-graph-config/v1" +_SHA256 = re.compile(r"^[a-f0-9]{64}$") +_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$") + + +class ProviderContractError(ValueError): + """A provider pin or graph execution policy is ambiguous or unsafe.""" + + +class ProviderRole(StrEnum): + SOURCE = "source" + DETECTOR = "detector" + GEOMETRY = "geometry" + TEMPORAL = "temporal" + MOTION = "motion" + THREAT = "threat" + + +@dataclass(frozen=True, slots=True) +class ProviderPin: + role: ProviderRole + provider_id: str + version: str + revision: str + sha256: str + + def __post_init__(self) -> None: + for value, label in ( + (self.provider_id, "provider id"), + (self.version, "provider version"), + (self.revision, "provider revision"), + ): + _identifier(value, label) + _digest(self.sha256, "provider digest") + + def to_dict(self) -> dict[str, object]: + return { + "role": self.role.value, + "provider_id": self.provider_id, + "version": self.version, + "revision": self.revision, + "sha256": self.sha256, + } + + @classmethod + def from_dict(cls, value: object) -> ProviderPin: + document = _object(value, "provider pin") + _exact_keys( + document, + {"role", "provider_id", "version", "revision", "sha256"}, + "provider pin", + ) + return cls( + role=_enum(ProviderRole, document.get("role"), "provider role"), + provider_id=_string(document, "provider_id"), + version=_string(document, "version"), + revision=_string(document, "revision"), + sha256=_string(document, "sha256"), + ) + + +@dataclass(frozen=True, slots=True) +class QueuePolicy: + stage_id: str + capacity: int + deadline_ns: int + terminal_timeout_ns: int + + def __post_init__(self) -> None: + _identifier(self.stage_id, "queue stage id") + _positive_integer(self.capacity, "queue capacity") + _positive_integer(self.deadline_ns, "queue deadline") + _positive_integer(self.terminal_timeout_ns, "terminal timeout") + if self.terminal_timeout_ns < self.deadline_ns: + raise ProviderContractError("terminal timeout cannot precede stage deadline") + + def to_dict(self) -> dict[str, object]: + return { + "stage_id": self.stage_id, + "capacity": self.capacity, + "deadline_ns": self.deadline_ns, + "terminal_timeout_ns": self.terminal_timeout_ns, + } + + @classmethod + def from_dict(cls, value: object) -> QueuePolicy: + document = _object(value, "queue policy") + _exact_keys( + document, + {"stage_id", "capacity", "deadline_ns", "terminal_timeout_ns"}, + "queue policy", + ) + return cls( + stage_id=_string(document, "stage_id"), + capacity=_integer(document, "capacity"), + deadline_ns=_integer(document, "deadline_ns"), + terminal_timeout_ns=_integer(document, "terminal_timeout_ns"), + ) + + +@dataclass(frozen=True, slots=True) +class GraphAuthority: + mode: str = "replay-simulated" + physical_live: bool = False + commands_enabled: bool = False + actuation_allowed: bool = False + navigation_or_safety_accepted: bool = False + + def __post_init__(self) -> None: + if self.mode != "replay-simulated": + raise ProviderContractError("M4 graph authority must be replay-simulated") + if any( + ( + self.physical_live, + self.commands_enabled, + self.actuation_allowed, + self.navigation_or_safety_accepted, + ) + ): + raise ProviderContractError("M4 graph cannot publish physical or command authority") + + def to_dict(self) -> dict[str, object]: + return { + "mode": self.mode, + "physical_live": self.physical_live, + "commands_enabled": self.commands_enabled, + "actuation_allowed": self.actuation_allowed, + "navigation_or_safety_accepted": self.navigation_or_safety_accepted, + } + + @classmethod + def from_dict(cls, value: object) -> GraphAuthority: + document = _object(value, "graph authority") + _exact_keys( + document, + { + "mode", + "physical_live", + "commands_enabled", + "actuation_allowed", + "navigation_or_safety_accepted", + }, + "graph authority", + ) + return cls( + mode=_string(document, "mode"), + physical_live=_boolean(document, "physical_live"), + commands_enabled=_boolean(document, "commands_enabled"), + actuation_allowed=_boolean(document, "actuation_allowed"), + navigation_or_safety_accepted=_boolean( + document, + "navigation_or_safety_accepted", + ), + ) + + +@dataclass(frozen=True, slots=True) +class ReferencePerceptionGraphConfig: + graph_id: str + source_profile_id: str + providers: tuple[ProviderPin, ...] + queues: tuple[QueuePolicy, ...] + authority: GraphAuthority = GraphAuthority() + + def __post_init__(self) -> None: + _identifier(self.graph_id, "graph id") + _identifier(self.source_profile_id, "source profile id") + roles = [provider.role for provider in self.providers] + if len(set(roles)) != len(roles) or set(roles) != set(ProviderRole): + raise ProviderContractError("graph must pin each provider role exactly once") + stage_ids = [queue.stage_id for queue in self.queues] + if not stage_ids or len(set(stage_ids)) != len(stage_ids): + raise ProviderContractError("graph queue policies must be nonempty and unique") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": REFERENCE_GRAPH_CONFIG_SCHEMA, + "graph_id": self.graph_id, + "source_profile_id": self.source_profile_id, + "providers": [provider.to_dict() for provider in self.providers], + "queues": [queue.to_dict() for queue in self.queues], + "authority": self.authority.to_dict(), + } + + @classmethod + def from_dict(cls, value: object) -> ReferencePerceptionGraphConfig: + document = _object(value, "reference graph config") + _exact_keys( + document, + {"schema_version", "graph_id", "source_profile_id", "providers", "queues", "authority"}, + "reference graph config", + ) + if document.get("schema_version") != REFERENCE_GRAPH_CONFIG_SCHEMA: + raise ProviderContractError("reference graph config schema is incompatible") + return cls( + graph_id=_string(document, "graph_id"), + source_profile_id=_string(document, "source_profile_id"), + providers=tuple(ProviderPin.from_dict(item) for item in _array(document, "providers")), + queues=tuple(QueuePolicy.from_dict(item) for item in _array(document, "queues")), + authority=GraphAuthority.from_dict(document.get("authority")), + ) + + +class SourceProvider(Protocol): + provider_id: str + + def envelopes(self) -> Iterator[SourceEnvelope]: ... + + +class DetectorProvider(Protocol): + provider_id: str + + def detect(self, envelope: SourceEnvelope) -> tuple[ObjectProposal2D, ...]: ... + + +class GeometryAssociationProvider(Protocol): + provider_id: str + + def associate( + self, + envelope: SourceEnvelope, + proposals: tuple[ObjectProposal2D, ...], + ) -> tuple[ObstacleObservation, ...]: ... + + +class TemporalStateProvider(Protocol): + provider_id: str + + def update( + self, + envelope: SourceEnvelope, + observations: tuple[ObstacleObservation, ...], + ) -> tuple[TemporalObstacle, ...]: ... + + +class MotionProvider(Protocol): + provider_id: str + + def estimate( + self, + envelope: SourceEnvelope, + obstacles: tuple[TemporalObstacle, ...], + ) -> tuple[TemporalObstacle, ...]: ... + + +class ThreatProvider(Protocol): + provider_id: str + + def assess(self, obstacle_map: LocalObstacleMap) -> tuple[ThreatAssessment, ...]: ... + + +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 ProviderContractError(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 ProviderContractError(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 ProviderContractError(f"{key} must be an array") + return value + + +def _string(document: dict[str, object], key: str) -> str: + value = document.get(key) + if not isinstance(value, str) or not value: + raise ProviderContractError(f"{key} must be a nonempty string") + return value + + +def _boolean(document: dict[str, object], key: str) -> bool: + value = document.get(key) + if not isinstance(value, bool): + raise ProviderContractError(f"{key} must be boolean") + return value + + +def _integer(document: dict[str, object], key: str) -> int: + value = document.get(key) + if not isinstance(value, int) or isinstance(value, bool): + raise ProviderContractError(f"{key} must be an integer") + return value + + +def _positive_integer(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value < 1: + raise ProviderContractError(f"{label} must be positive") + return value + + +def _identifier(value: str, label: str) -> str: + if _IDENTIFIER.fullmatch(value) is None: + raise ProviderContractError(f"{label} is not a safe identifier") + return value + + +def _digest(value: str, label: str) -> str: + if _SHA256.fullmatch(value) is None: + raise ProviderContractError(f"{label} must be a SHA-256 digest") + return value + + +def _enum(enum_type: type[ProviderRole], value: object, label: str) -> ProviderRole: + if not isinstance(value, str): + raise ProviderContractError(f"{label} must be a string") + try: + return enum_type(value) + except ValueError as exc: + raise ProviderContractError(f"{label} is incompatible") from exc diff --git a/tests/test_perception_architecture.py b/tests/test_perception_architecture.py new file mode 100644 index 0000000..1c81dee --- /dev/null +++ b/tests/test_perception_architecture.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import ast +import copy +import json +from pathlib import Path + +import pytest + +from k1link.perception.baseline import ( + BaselineContractError, + load_m4_baseline, + validate_reuse_inventory, + verify_m4_baseline, +) + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +PERCEPTION_ROOT = REPOSITORY_ROOT / "src" / "k1link" / "perception" +BASELINE_PATH = REPOSITORY_ROOT / "config" / "perception" / "m4-recorded-realtime-baseline-v1.json" +REUSE_PATH = REPOSITORY_ROOT / "config" / "perception" / "m4-reuse-inventory-v1.json" + + +def _imports(path: Path) -> set[str]: + tree = ast.parse(path.read_text("utf-8"), filename=str(path)) + modules: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + modules.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + modules.add(node.module) + return modules + + +def test_m4_baseline_is_exact_and_every_local_evidence_digest_resolves() -> None: + profile = load_m4_baseline(BASELINE_PATH) + verification = verify_m4_baseline(REPOSITORY_ROOT, profile) + assert verification.source_id == "RAVNOVES00" + assert verification.session_id == "20260720T065719Z_viewer_live" + assert len(verification.verified_paths) == 6 + + +def test_m4_baseline_cannot_silently_select_another_source(tmp_path: Path) -> None: + document = json.loads(BASELINE_PATH.read_text("utf-8")) + incompatible = copy.deepcopy(document) + incompatible["source"]["source_id"] = "RAVNOVES01" + path = tmp_path / "baseline.json" + path.write_text(json.dumps(incompatible), "utf-8") + with pytest.raises(BaselineContractError, match="RAVNOVES00"): + load_m4_baseline(path) + + +def test_reuse_inventory_separates_primitives_from_historical_wrappers() -> None: + document = validate_reuse_inventory(REUSE_PATH) + assert document["rules"]["bulk_legacy_migration_required"] is False + + +def test_perception_contracts_import_no_compute_lab_graph_or_web_module() -> None: + imports = _imports(PERCEPTION_ROOT / "contracts.py") + forbidden = { + module + for module in imports + if module.startswith( + ( + "k1link.compute", + "k1link.laboratory", + "k1link.web", + "k1link.perception.providers", + "k1link.perception.graph", + ) + ) + } + assert forbidden == set() + + +def test_new_perception_boundary_has_no_experiment_specific_imports() -> None: + violations: dict[str, str] = {} + for path in PERCEPTION_ROOT.glob("*.py"): + for module in _imports(path): + leaf = module.rsplit(".", 1)[-1] + if module.startswith("k1link.compute") and ( + leaf.startswith("e") or leaf.startswith("l") + ): + violations[path.name] = module + assert violations == {} diff --git a/tests/test_perception_contracts.py b/tests/test_perception_contracts.py new file mode 100644 index 0000000..d2d907c --- /dev/null +++ b/tests/test_perception_contracts.py @@ -0,0 +1,419 @@ +from __future__ import annotations + +import copy +import json +from dataclasses import replace + +import numpy as np +import pytest + +from k1link.compute.temporal_occupied_layer import TemporalFrameProjection +from k1link.compute.track_geometry import ( + PointSlab, + TrackGeometry, + TrackGeometryCurrentness, + TrackGeometryEvidenceState, + TrackGeometryFrame, + TrackGeometryMetricBasis, + TrackGeometryOwnerKind, + TrackGeometrySourceBinding, +) +from k1link.perception.adapters import ( + observations_from_track_geometry, + temporal_obstacles_from_e34_projection, +) +from k1link.perception.contracts import ( + BoundingRegion2D, + ClockBasis, + CorridorIntersection, + EvidenceBasis, + EvidenceCurrentness, + GridCell, + HistorySample, + LocalObstacleMap, + MetricGeometry, + ModalityOutcome, + ModalityStatus, + MotionState, + ObjectProposal2D, + ObstacleObservation, + PerceptionContractError, + QualificationState, + SourceAccounting, + SourceEnvelope, + TemporalObstacle, + TemporalState, + ThreatAssessment, + ThreatDecision, + TimestampBundle, + validate_exclusive_point_ownership, +) +from k1link.perception.providers import ( + GraphAuthority, + ProviderContractError, + ProviderPin, + ProviderRole, + QueuePolicy, + ReferencePerceptionGraphConfig, +) + + +def _status(outcome: ModalityOutcome = ModalityOutcome.AVAILABLE) -> ModalityStatus: + return ModalityStatus( + available=outcome is ModalityOutcome.AVAILABLE, + outcome=outcome, + reason=outcome.value, + ) + + +def _source(*, lidar: ModalityOutcome = ModalityOutcome.AVAILABLE) -> SourceEnvelope: + return SourceEnvelope( + source_id="RAVNOVES00", + session_id="20260720T065719Z_viewer_live", + frame_id="frame-000001", + sequence=1, + timestamps=TimestampBundle( + utc_ns=1_786_000_000_000_000_000, + monotonic_ns=1_000_000, + source_ns=35_421_857_292, + clock_basis=ClockBasis.RECORDED_HOST, + ), + source_age_ns=0, + binding_reason="exact-recorded-source", + calibration_id="camera-1-kb4-05f3ad9b", + representation_id="registered-map-increment-v1", + image=_status(), + registered_point_increment=_status(lidar), + pose=_status(), + ) + + +def _proposal(*, semantic_hint: str | None = None) -> ObjectProposal2D: + return ObjectProposal2D( + proposal_id="proposal-1", + source_id="RAVNOVES00", + frame_id="frame-000001", + region=BoundingRegion2D(10.0, 20.0, 80.0, 100.0), + objectness=0.91, + provider_id="triton-yolox-s-raw-kb4/v1", + model_id="yolox_s:1", + preprocess_id="raw-kb4-valid-fov-letterbox/v1", + semantic_hint=semantic_hint, + provider_tracklet="detector-local-7", + ) + + +def _geometry() -> MetricGeometry: + return MetricGeometry( + coordinate_frame="map", + centroid_xyz_m=(4.0, 1.0, 0.5), + range_m=4.15, + covariance_diagonal_m2=(0.04, 0.04, 0.09), + ) + + +def _observation(*, semantic_hint: str | None = None, point_id: int = 5) -> ObstacleObservation: + return ObstacleObservation( + observation_id=f"observation-{point_id}", + occupancy_key="frame-local-occupied-1", + source_id="RAVNOVES00", + frame_id="frame-000001", + evidence_time_ns=35_421_857_292, + basis=EvidenceBasis.FUSED, + currentness=EvidenceCurrentness.CURRENT, + occupied_support=True, + source_point_ids=(point_id,), + metric_geometry=_geometry(), + proposal_ids=("proposal-1",), + semantic_hint=semantic_hint, + reason_codes=("current-qualified-points",), + ) + + +def _temporal(*, state: TemporalState = TemporalState.CURRENT) -> TemporalObstacle: + age_ns = 0 if state is TemporalState.CURRENT else 50_000_000 + cells = () if state is TemporalState.EXPIRED else (GridCell(8, 2, 1),) + return TemporalObstacle( + component_id=f"component-{state.value}", + identity_scope="ephemeral", + state=state, + ttl_ns=750_000_000, + last_hit_ns=35_421_857_292, + age_ns=age_ns, + association_basis="current-spatial-support", + history=( + HistorySample( + frame_id="frame-000001", + evidence_time_ns=35_421_857_292, + centroid_xyz_m=(4.0, 1.0, 0.5), + ), + ), + cells=cells, + coordinate_frame=None if state is TemporalState.EXPIRED else "map", + last_centroid_xyz_m=None if state is TemporalState.EXPIRED else (4.0, 1.0, 0.5), + motion=MotionState.UNKNOWN, + motion_confidence=0.0, + motion_reason="insufficient-history", + ) + + +def test_six_contracts_round_trip_with_exact_json_shapes() -> None: + source = _source() + proposal = _proposal() + observation = _observation() + temporal = _temporal() + obstacle_map = LocalObstacleMap( + source_id=source.source_id, + session_id=source.session_id, + frame_id=source.frame_id, + graph_id="reference-perception-graph/v1", + generated_monotonic_ns=1_010_000, + output_age_ns=10_000, + occupied=(temporal,), + unknown=(_temporal(state=TemporalState.HELD),), + camera_uncertainty=(proposal,), + accounting=SourceAccounting(1, 1, 0, 0), + ) + assessment = ThreatAssessment( + assessment_id="assessment-1", + component_id=temporal.component_id, + rig_profile_id="virtual-rig-ravnoves00/v1", + corridor_profile_id="virtual-corridor-ravnoves00/v1", + qualification=QualificationState.QUALIFIED, + relative_speed_mps=-0.2, + closest_approach_m=3.0, + ttc_seconds=None, + corridor_intersection=CorridorIntersection.CLEAR, + decision=ThreatDecision.NOT_THREAT, + reason_codes=("qualified-corridor-clear",), + ) + + values = ( + (SourceEnvelope, source), + (ObjectProposal2D, proposal), + (ObstacleObservation, observation), + (TemporalObstacle, temporal), + (LocalObstacleMap, obstacle_map), + (ThreatAssessment, assessment), + ) + for contract_type, contract in values: + document = json.loads(json.dumps(contract.to_dict())) + assert contract_type.from_dict(document) == contract + incompatible = copy.deepcopy(document) + incompatible["unexpected"] = True + with pytest.raises(PerceptionContractError, match="fields are incompatible"): + contract_type.from_dict(incompatible) + + +def test_object_proposal_is_valid_without_a_semantic_class() -> None: + proposal = _proposal(semantic_hint=None) + assert ObjectProposal2D.from_dict(proposal.to_dict()) == proposal + assert proposal.semantic_hint is None + assert not hasattr(proposal, "range_m") + + +def test_semantic_change_does_not_change_occupancy_identity() -> None: + before = _observation(semantic_hint="car") + after = replace(before, semantic_hint="person") + assert before.occupancy_identity == after.occupancy_identity + assert before.source_point_ids == after.source_point_ids + + +def test_geometry_only_obstacle_is_valid_without_class_or_proposal() -> None: + observation = replace( + _observation(), + basis=EvidenceBasis.LIDAR, + proposal_ids=(), + semantic_hint=None, + ) + assert ObstacleObservation.from_dict(observation.to_dict()) == observation + + +def test_camera_only_observation_remains_non_metric_uncertainty() -> None: + observation = ObstacleObservation( + observation_id="camera-observation-1", + occupancy_key="camera-uncertainty-1", + source_id="RAVNOVES00", + frame_id="frame-000001", + evidence_time_ns=35_421_857_292, + basis=EvidenceBasis.CAMERA, + currentness=EvidenceCurrentness.CURRENT, + occupied_support=False, + source_point_ids=(), + metric_geometry=None, + proposal_ids=("proposal-1",), + semantic_hint=None, + reason_codes=("camera-only-no-metric-support",), + ) + assert observation.metric_geometry is None + + +def test_range_without_current_qualified_points_is_rejected() -> None: + with pytest.raises(PerceptionContractError, match="qualified points"): + replace(_observation(), source_point_ids=()) + with pytest.raises(PerceptionContractError, match="non-current"): + replace(_observation(), currentness=EvidenceCurrentness.HELD) + + +def test_duplicate_source_point_ownership_is_rejected_across_observations() -> None: + first = _observation(point_id=5) + second = replace(first, observation_id="observation-duplicate") + with pytest.raises(PerceptionContractError, match="duplicate observation ownership"): + validate_exclusive_point_ownership((first, second)) + + +def test_missing_lidar_cannot_be_published_as_free_space() -> None: + source = _source(lidar=ModalityOutcome.UNAVAILABLE) + assert source.registered_point_increment.available is False + with pytest.raises(PerceptionContractError, match="implicit free space"): + LocalObstacleMap( + source_id=source.source_id, + session_id=source.session_id, + frame_id=source.frame_id, + graph_id="reference-perception-graph/v1", + generated_monotonic_ns=1, + output_age_ns=0, + occupied=(), + unknown=(), + camera_uncertainty=(_proposal(),), + accounting=SourceAccounting(1, 1, 0, 0), + free_space_claimed=True, + ) + + +def test_threat_requires_profiles_and_never_grants_physical_authority() -> None: + with pytest.raises(PerceptionContractError, match="rig profile id"): + ThreatAssessment( + assessment_id="assessment-1", + component_id="component-current", + rig_profile_id="", + corridor_profile_id="corridor/v1", + qualification=QualificationState.UNQUALIFIED, + relative_speed_mps=None, + closest_approach_m=None, + ttc_seconds=None, + corridor_intersection=CorridorIntersection.UNKNOWN, + decision=ThreatDecision.UNKNOWN, + reason_codes=("missing-rig",), + ) + with pytest.raises(PerceptionContractError, match="collision or actuation"): + ThreatAssessment( + assessment_id="assessment-1", + component_id="component-current", + rig_profile_id="rig/v1", + corridor_profile_id="corridor/v1", + qualification=QualificationState.QUALIFIED, + relative_speed_mps=1.0, + closest_approach_m=0.5, + ttc_seconds=1.0, + corridor_intersection=CorridorIntersection.INTERSECTS, + decision=ThreatDecision.THREAT, + reason_codes=("intersects",), + actuation_allowed=True, + ) + + +def test_reference_graph_config_pins_all_roles_and_queue_bounds() -> None: + config = ReferencePerceptionGraphConfig( + graph_id="reference-perception-graph/v1", + source_profile_id="m4-ravnoves00-recorded-realtime/v1", + providers=tuple( + ProviderPin(role, f"{role.value}-provider", "v1", "78a3dc2", "a" * 64) + for role in ProviderRole + ), + queues=(QueuePolicy("detector", 2, 80_000_000, 200_000_000),), + authority=GraphAuthority(), + ) + assert ReferencePerceptionGraphConfig.from_dict(config.to_dict()) == config + with pytest.raises(ProviderContractError, match="each provider role"): + replace(config, providers=config.providers[:-1]) + with pytest.raises(ProviderContractError, match="physical or command authority"): + GraphAuthority(commands_enabled=True) + + +def test_track_geometry_adapter_preserves_exact_point_ownership_without_class() -> None: + binding = TrackGeometrySourceBinding( + source_pack_id=( + "e10-lidar-pack-576c994a6c814e2592dd6240ace3902a5db94843312c759a73ba0c9166157d2b" + ), + source_session_id="20260720T065719Z_viewer_live", + representation_profile_id="registered-map-increment-v1", + e31_qualification_id=( + "e31-source-qualification-b2460a5eb143688c7eea6821b2277e13aea79868abe81d83f7e78548c119159a" + ), + calibration_sha256="0" * 64, + coordinate_frame="map", + time_basis="recorded-host", + selected_offset_ms=0, + ) + slab = PointSlab( + frame_index=1, + source_frame_index=1, + source_point_count=10, + coordinate_frame="map", + owner_keys=("geometry-1",), + source_indices=np.asarray([7, 8], dtype=" None: + component = { + "temporal_id": 3, + "last_observed_age_seconds": 0.1, + "association_reason": "ttl-hold-last-hit", + "centroid_map_xyz_m": [4.0, 1.0, 0.5], + "cell_row_start": 0, + "cell_row_count": 1, + "history_tail": [ + { + "frame_index": 10, + "session_seconds": 36.0, + "centroid_map_xyz_m": [4.0, 1.0, 0.5], + } + ], + "semantic_provenance": {"labels": ["car"]}, + } + projection = TemporalFrameProjection( + document={"current": [], "held": [component], "expired": []}, + cell_rows=np.asarray([[8, 2, 1]], dtype="