diff --git a/docs/09_OBSERVATION_SESSIONS.md b/docs/09_OBSERVATION_SESSIONS.md index 84d43db..8a179a6 100644 --- a/docs/09_OBSERVATION_SESSIONS.md +++ b/docs/09_OBSERVATION_SESSIONS.md @@ -125,12 +125,13 @@ viewer-ready form. Deleting a LAB catalog entry cannot delete the referenced source session, and a source session with published LAB instances fails closed on deletion. -The visual result is resolved by the LAB session id, so `LAB E19` and `LAB E21` -can coexist over `RAVNOVES00` without “latest accepted result” replacing an -earlier experiment. The operator can open either row in **Сохранённые сессии** -and use the same **Объекты 2D / Сегментация / Кубы 3D** controls. First -publication must also warm the result-specific Rerun overlay; later opens reuse -that verified cache and do not rerun AI inference. +The visual result is resolved by the LAB session id, so `LAB E19`, `LAB E21` +and the derived temporal comparison `LAB E22` can coexist over `RAVNOVES00` +without “latest accepted result” replacing an earlier experiment. The operator +can open any row in **Сохранённые сессии** and use the same +**Объекты 2D / Сегментация / Кубы 3D** controls. First publication must also +warm the result-specific Rerun overlay; later opens reuse that verified cache +and do not rerun AI inference. E21 is a bounded special case: its worker persisted 121 semantic mask hashes, not duplicate mask pixels. Its visual publication materializes a mask only @@ -143,6 +144,13 @@ the 60-second result to the complete 8:55 source replay is an invalid presentation because latest-at image data would appear static after the E21 window ends. +E22 is also bounded to the exact E21 60-second range, but it resolves its +catalog provenance directly to the physical source session rather than chaining +one LAB entry to another. Its temporal result remains a separate immutable +integrated overlay. Raw point-cloud, camera, calibration and E21 inference +payloads are not replaced. Short held boxes are provenance-marked diagnostic +presentation/world-state values, not fresh sensor measurements. + The first preparation of a long capture can take time because every decodable point/pose message and every valid K1 `ModelingReport` is projected into RRD. Later openings reuse a verified, digest-bound cache. Derived RRD cache v9 is diff --git a/experiments/perception/LAB_E22_REPORT_2026-07-23.md b/experiments/perception/LAB_E22_REPORT_2026-07-23.md new file mode 100644 index 0000000..1bf9090 --- /dev/null +++ b/experiments/perception/LAB_E22_REPORT_2026-07-23.md @@ -0,0 +1,217 @@ +# LAB E22 — Bounded temporal stability + +Date: 2026-07-23 +State: accepted diagnostic laboratory result +Authority: observation only; commands and navigation/safety acceptance disabled + +## 1. Purpose + +LAB E22 tests whether the perception output already qualified by LAB E21 can +be made materially calmer without losing the real-time envelope required for +an unmanned vehicle. + +This is not another model-quality run in isolation. The experiment adds the +missing temporal output stage between AI inference and the operator/autonomy +consumer: + +- stabilize tracked 2D boxes without future-frame lookahead; +- preserve short detector gaps with explicit, bounded holds; +- stabilize map-frame 3D cuboids and their orientation; +- reduce unsupported semantic label islands; +- retain immutable raw camera, LiDAR, calibration and E21 inference evidence; +- measure the added latency and memory bound. + +## 2. Exact source + +| Field | Value | +| --- | --- | +| Source LAB | `lab-e21-window2-d0201712` | +| Source display name | `LAB E21.3 · Real-time 1× · synchronized 60s · d0201712` | +| Physical source session | `20260720T065719Z_viewer_live` (`RAVNOVES00`) | +| Source result | `e10-integrated-perception-5cbba0b9839ae843c05607cec8448a2b43464eb81edeb11cf7c01cfec8464c75` | +| Camera source | `sensor.camera.right`, XGRIDS camera slot `camera_1` | +| Resolution | 800 × 600 | +| Window | 601 camera frames, 59.962 s | +| Semantic masks | 121 | +| Source detector | YOLOX-S, COCO-80 | +| Source semantic model | EoMT Cityscapes large, FP16 | +| Source inference hardware | NVIDIA GeForce RTX 4090 | +| Source detector throughput | 9.865 FPS | +| Source semantic throughput | 2.010 FPS | +| Source world-state age | P95 174.40 ms; P99 289.03 ms | + +The E22 postprocessor ran locally over the immutable E21.3 result. Its +implementation is streaming-equivalent: one pass, no future-frame lookahead, +bounded state, and no repeated neural inference. + +## 3. Configuration + +Profile: +`experiments/perception/e22_temporal_stability_profile.json` + +Profile SHA-256: +`3d61b29562a851ec41e504bb175e392be874c4509e6944834a449d2e347022d2` + +### 3.1 2D tracking + +- adaptive center EMA: alpha 0.32–0.78; +- adaptive size EMA: alpha 0.22–0.65; +- center-velocity EMA: alpha 0.35; +- large normalized innovation: reset, not forced smoothing; +- missing observation hold: at most two camera frames; +- class-consistent tracklet stitch: at most eight frames, IoU/center gated. + +### 3.2 3D cuboids + +- map-frame center EMA: alpha 0.28; +- half-size EMA: alpha 0.18; +- pi-periodic yaw EMA: alpha 0.18; +- velocity EMA: alpha 0.25; +- center innovation reset: above 2.0 m; +- yaw innovation reset: above 55°; +- last-supported cuboid hold: at most 350 ms. + +The complete point cloud is not filtered or reduced by this stage. The ground +and all other points remain available to mapping, terrain and traversability +consumers. + +### 3.3 Semantic output + +The filter uses one previous mask and accepts a changed pixel immediately when +the new label has spatial support. Unsupported isolated changes retain the +previous label. + +This is deliberately not called optical-flow propagation. It removes label +islands but does not interpolate semantic geometry between the source model's +2 FPS updates. + +### 3.4 Hard bounds + +- maximum track states: 128; +- observed peak track states: 19; +- semantic history: one 800 × 600 `uint8` mask; +- estimated upper state bound: 1,004,288 bytes; +- future-frame lookahead: zero. + +## 4. Baseline versus E22 + +| Metric | E21.3 baseline | E22 | Change | +| --- | ---: | ---: | ---: | +| 2D normalized acceleration, P95 | 0.5232 | 0.3938 | −24.7% | +| 2D normalized size step, P95 | 0.2670 | 0.1520 | −43.1% | +| 2D track count | 165 | 160 | −5 stitched identities | +| Tracks lasting ≤2 frames | 34 | 0 | −34 | +| 3D center step, P95 | 0.8174 m | 0.5698 m | −30.3% | +| 3D half-size step, P95 | 0.0542 m | 0.0189 m | −65.1% | +| 3D yaw step, P95 | 15.51° | 5.26° | −66.1% | +| 3D discontinuity gaps | 174 | 87 | −50.0% | +| Unsupported semantic change, mean | 0.03284% | 0.00718% | −78.1% | +| Total semantic mask change, mean | 5.3658% | 5.3261% | −0.7% | + +The distinction in the last two rows is important. E22 strongly reduces +unsupported semantic noise, but it does not yet solve the larger inter-frame +movement of semantic boundaries. That requires confidence-aware propagation +or optical flow in a later experiment. + +## 5. Temporal events and provenance + +| Event | Count | +| --- | ---: | +| Tracklets stitched | 7 | +| 2D held presentations | 639 | +| 3D held cuboids | 472 | +| 2D large-innovation resets | 60 | +| 3D large-innovation resets | 69 | + +The rise from 881 source accepted cuboids to 1,353 E22 presented cuboids does +not mean 472 new measurements were invented. Those 472 are explicitly marked +short holds of the last supported cuboid. Every hold is diagnostic and must +not be interpreted as a fresh LiDAR association. + +## 6. Added runtime cost + +Measured on the local deterministic postprocessor: + +| Stage | Mean | P95 | P99 | Max | +| --- | ---: | ---: | ---: | ---: | +| Per camera frame, 2D/3D/world state | 0.468 ms | 0.821 ms | 0.959 ms | 1.244 ms | +| Per semantic frame | 3.463 ms | 3.650 ms | 3.815 ms | 4.054 ms | + +At the source rates of approximately 10 camera frames/s and 2 semantic +frames/s, this temporal stage consumes far less than one frame interval. It +does not prove the complete live K1 path; it proves that this specific +stabilization stage is not the real-time bottleneck. + +## 7. Acceptance + +All gates passed: + +- 2D acceleration P95 reduction ≥20%; +- 3D center-step P95 reduction ≥20%; +- 3D yaw-step P95 reduction ≥20%; +- unsupported semantic change reduction ≥10%; +- camera-frame postprocess P95 ≤5 ms; +- semantic-frame postprocess P95 ≤30 ms; +- track-state bound respected; +- source payloads immutable; +- diagnostic authority only. + +## 8. Published operator result + +| Field | Value | +| --- | --- | +| Session ID | `lab-e22-3d61b295` | +| Display name | `LAB E22 · Temporal stability · synchronized 60s · 3d61b295` | +| Result ID | `e10-integrated-perception-416e8b326ea36e7d534aa9e92b9136c916c7be8cd4e86a1df0feb97bd0d67c71` | +| Result size | approximately 7.0 MiB | +| Rerun perception overlay | 7,896,401 bytes | +| Catalog state | Ready | + +UI acceptance on `http://127.0.0.1:8000/`: + +- LAB E22 appears as a separate saved session; +- the replay duration is 59.962 s; +- AI overlay endpoint returns HTTP 200 and a valid `RRF2` payload; +- `Объекты 2D`, `Сегментация`, and `Кубы 3D` become enabled; +- all three layers can be enabled together; +- the original point cloud remains visible; +- the Mission Core server remained on PID 31145 throughout publication and + acceptance. + +## 9. Artifact hashes + +| Artifact | SHA-256 | +| --- | --- | +| `result.json` | `00ebc409fd09faccd79cfff3342f109cbfba907880cc93cc834026e2ce104764` | +| `run-report.json` | `cc1d775ae52d8faa7388ddc0a0c8aba0e53ff3e90ace46b2a3749f02dfeeb912` | +| `fusion-frames.jsonl` | `456a7d8b1682b85856cbd036053e61be95a6d3fe18453277bcadde879bc9cb9c` | +| `semantic-frames.jsonl` | `d52dfbb03877ec1e50da073a221f20f8b57e5c9669a0de4b16505caa97720e5e` | +| `world-state.jsonl` | `281f26e5fd645ed995bf253bd0ff45bbc07476ef84a2c3df9b198bbc02ef44ca` | +| `transient-perception.npz` | `84609a59abb437ffabc8ca3769327450da5d5250b16c4e1b5e5f7f2e9f73e90f` | + +Runtime artifacts remain outside Git under `.runtime/`. + +## 10. Honest limitations + +- No ground-truth identities are available; tracklet stitching is evaluated + through continuity metrics, not IDF1/HOTA. +- Two-frame 2D and 350 ms 3D holds improve continuity but are not fresh sensor + evidence. +- E22 does not improve detector recall, class taxonomy, fisheye model quality + or LiDAR association correctness. +- Semantic boundaries are not optical-flow propagated. +- The temporal code is streaming-equivalent, but this run consumed a recorded + E21 result rather than the physical live K1 transport. +- Navigation and safety authority remain disabled. + +## 11. Next gate + +The next experiment should move this exact bounded temporal state machine into +the warm worker's live result path and rerun the same 60-second 1× envelope. +The acceptance question is no longer “does it look calmer?” but: + +1. does inline E22 preserve detector/semantic throughput and E21 age budgets; +2. do real latest-wins replacements remain bounded; +3. do operator-visible holds carry freshness state end to end; +4. does memory remain flat during a longer replay; +5. only after that, does a physical K1 shadow run reproduce the recorded gate. diff --git a/experiments/perception/e22_temporal_stability_profile.json b/experiments/perception/e22_temporal_stability_profile.json new file mode 100644 index 0000000..809a270 --- /dev/null +++ b/experiments/perception/e22_temporal_stability_profile.json @@ -0,0 +1,60 @@ +{ + "schema_version": "missioncore.e22-temporal-stability-profile/v1", + "profile_id": "lab-e22-bounded-temporal-stability-v1", + "mode": "recorded-streaming-qualification", + "source": { + "source_id": "sensor.camera.right", + "resolution": [ + 800, + 600 + ], + "calibration_slot": "camera_1", + "calibration_sha256": "05f3ad9b38b3a4fc95388a8ec83da83c745e217709e51787b3d5aad0969f6fa9" + }, + "tracking_2d": { + "center_alpha_low": 0.32, + "center_alpha_high": 0.78, + "size_alpha_low": 0.22, + "size_alpha_high": 0.65, + "velocity_alpha": 0.35, + "adaptive_innovation_low": 0.08, + "adaptive_innovation_high": 0.3, + "maximum_normalized_innovation": 1.2, + "hold_frames": 2, + "stitch_gap_frames": 8, + "stitch_minimum_iou": 0.42, + "stitch_maximum_normalized_center_distance": 0.22 + }, + "cuboids_3d": { + "center_alpha": 0.28, + "size_alpha": 0.18, + "yaw_alpha": 0.18, + "velocity_alpha": 0.25, + "maximum_center_innovation_m": 2.0, + "maximum_yaw_innovation_degrees": 55.0, + "hold_seconds": 0.35 + }, + "semantic": { + "mode": "spatially-supported-streaming-hysteresis-v1", + "minimum_same_label_neighbors": 3, + "class_count": 16 + }, + "bounds": { + "maximum_track_states": 128, + "maximum_track_idle_seconds": 1.25, + "semantic_history_masks": 1 + }, + "acceptance": { + "minimum_2d_acceleration_p95_reduction_fraction": 0.2, + "minimum_3d_center_step_p95_reduction_fraction": 0.2, + "minimum_3d_yaw_step_p95_reduction_fraction": 0.2, + "minimum_semantic_unsupported_change_reduction_fraction": 0.1, + "maximum_camera_frame_processing_p95_ms": 5.0, + "maximum_semantic_frame_processing_p95_ms": 30.0, + "maximum_track_states_observed": 128 + }, + "authority": { + "commands_enabled": false, + "navigation_or_safety_accepted": false + } +} diff --git a/src/k1link/compute/__init__.py b/src/k1link/compute/__init__.py index 0e073e1..50436b0 100644 --- a/src/k1link/compute/__init__.py +++ b/src/k1link/compute/__init__.py @@ -36,7 +36,9 @@ from .jobs import ( ) from .lab_instances import ( PublishedIntegratedLabInstance, + PublishedTemporalLabInstance, publish_e21_lab_instance, + publish_e22_lab_instance, publish_integrated_lab_instance, ) from .live_perception import ( @@ -115,6 +117,7 @@ __all__ = [ "LivePerceptionIngress", "IntegratedPerceptionOverlayStore", "PublishedIntegratedLabInstance", + "PublishedTemporalLabInstance", "IntegratedPerceptionResult", "LiveReplayQualificationResult", "MultiratePerceptionArtifact", @@ -144,6 +147,7 @@ __all__ = [ "validate_live_replay_qualification_result", "validate_integrated_perception_result", "publish_e21_lab_instance", + "publish_e22_lab_instance", "publish_integrated_lab_instance", "validate_multirate_perception_qualification_result", "prepare_recorded_qualification_slice", diff --git a/src/k1link/compute/lab_instances.py b/src/k1link/compute/lab_instances.py index 10865c1..da79e5d 100644 --- a/src/k1link/compute/lab_instances.py +++ b/src/k1link/compute/lab_instances.py @@ -27,6 +27,10 @@ from .integrated_perception import ( validate_integrated_perception_result, ) from .jobs import CameraComputeJob, validate_camera_compute_job +from .temporal_stability import ( + TemporalStabilityBuild, + build_temporal_stability_result, +) @dataclass(frozen=True, slots=True) @@ -36,6 +40,14 @@ class PublishedIntegratedLabInstance: result: IntegratedPerceptionResult +@dataclass(frozen=True, slots=True) +class PublishedTemporalLabInstance: + binding: LabSessionBinding + job: CameraComputeJob + result: IntegratedPerceptionResult + build: TemporalStabilityBuild + + def publish_integrated_lab_instance( *, repository_root: Path, @@ -241,6 +253,114 @@ def publish_e21_lab_instance( return PublishedIntegratedLabInstance(binding=binding, job=lab_job, result=validated) +def publish_e22_lab_instance( + *, + repository_root: Path, + source_result_root: Path, + profile_path: Path, + lab_session_id: str, + lab_id: str, + display_name: str, +) -> PublishedTemporalLabInstance: + """Derive and publish one bounded temporal-stability LAB comparison.""" + + root = repository_root.expanduser().resolve(strict=True) + jobs_root = root / ".runtime" / "compute-jobs" + results_root = root / ".runtime" / "compute-experiments" / "e10" / "worker-results" + packs_root = root / ".runtime" / "compute-experiments" / "e10" / "lidar-packs" + source_path = source_result_root.expanduser().resolve(strict=True) + source_document = _read_object(source_path / "result.json", source_path) + identity = source_document.get("identity") + if not isinstance(identity, dict) or not isinstance(identity.get("job_id"), str): + raise SessionIntegrityError("E22 source has no job identity") + source = validate_integrated_perception_result( + jobs_root / identity["job_id"], + source_path, + packs_root, + ) + if not source.accepted: + raise SessionIntegrityError("E22 source result is not accepted") + + lab_job = _publish_lab_job(source.job, jobs_root, lab_session_id) + lab_pack = _publish_lab_pack(source, lab_job, packs_root, lab_session_id) + build = build_temporal_stability_result( + source=source, + lab_job=lab_job, + lab_pack=lab_pack, + results_root=results_root, + profile_path=profile_path, + ) + validated = validate_integrated_perception_result( + lab_job.job_root, + build.result_root, + packs_root, + ) + if not validated.accepted: + failed = [ + name + for name, accepted in build.report["acceptance"]["checks"].items() + if not accepted + ] + raise SessionIntegrityError( + f"E22 temporal acceptance failed: {', '.join(failed)}" + ) + + store = SessionStore(root) + source_lab = store.get_lab_instance(source.job.session_id) + source_session_id = ( + source.job.session_id + if source_lab is None + else source_lab.source_session_id + ) + publish_lab_replay_cache( + store.data_dir, + source_session_id=source_session_id, + lab_session_id=lab_session_id, + timeline_start_ns=round(validated.timeline_start_seconds * 1_000_000_000), + timeline_end_ns=round(validated.timeline_end_seconds * 1_000_000_000), + ) + metrics = build.report["metrics"] + binding = store.publish_lab_instance( + session_id=lab_session_id, + source_session_id=source_session_id, + display_name=display_name, + lab_id=lab_id, + result_kind="e22-temporal-stability", + result_id=validated.result_id, + source_result_id=source.result_id, + config_sha256=build.profile_sha256, + run_created_at_utc=validated.created_at_utc, + duration_seconds=( + validated.timeline_end_seconds - validated.timeline_start_seconds + ), + include_recorded_media=False, + provenance={ + "schema_version": "missioncore.e22-lab-publication/v1", + "storage_mode": "bounded-derived-replay-and-temporal-projection", + "source_result_id": source.result_id, + "source_lab_session_id": ( + None if source_lab is None else source_lab.session_id + ), + "source_payloads_mutated": False, + "lookahead_frames": 0, + "peak_track_states": metrics["runtime"]["peak_track_states"], + "camera_frame_processing_p95_ms": metrics["runtime"][ + "camera_frame_processing_ms" + ]["p95"], + "semantic_frame_processing_p95_ms": metrics["runtime"][ + "semantic_frame_processing_ms" + ]["p95"], + "quality_reductions": metrics["reductions"], + }, + ) + return PublishedTemporalLabInstance( + binding=binding, + job=lab_job, + result=validated, + build=build, + ) + + def _validate_e21_inputs( e21_root: Path, worker_root: Path, diff --git a/src/k1link/compute/temporal_stability.py b/src/k1link/compute/temporal_stability.py new file mode 100644 index 0000000..dc8ce86 --- /dev/null +++ b/src/k1link/compute/temporal_stability.py @@ -0,0 +1,1079 @@ +"""Bounded temporal stabilization for diagnostic recorded perception streams. + +The source detector, semantic masks, LiDAR cloud and calibration remain +immutable. This module adds a streaming presentation/world-state stage with +bounded track memory, short provenance-marked holds and measurable latency. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import secrets +import shutil +import statistics +import time +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import numpy as np + +from k1link.artifacts import write_json_atomic +from k1link.sessions import SessionIntegrityError + +from .integrated_perception import IntegratedPerceptionResult +from .jobs import CameraComputeJob + +PROFILE_SCHEMA = "missioncore.e22-temporal-stability-profile/v1" +PIPELINE_ID = "bounded-temporal-stability-2d-3d-semantic/v1" +TARGET_TAXONOMY = { + 0: "outside_valid_fov", + 1: "person", + 2: "bicycle", + 3: "motorcycle", + 4: "car", + 5: "heavy_vehicle", + 6: "building_structure", + 7: "paved_road", + 8: "sidewalk_curb", + 9: "ground_dirt", + 10: "grass_low_vegetation", + 11: "tree_woody_vegetation", + 12: "sky", + 13: "static_obstacle", + 14: "animal", + 15: "other_background", +} + + +@dataclass(slots=True) +class _TrackState: + canonical_id: int + source_ids: set[int] + label: str + group: str + center: np.ndarray + size: np.ndarray + raw_center: np.ndarray + velocity: np.ndarray + last_frame: int + last_seconds: float + last_observed_frame: int + score: float + template: dict[str, Any] + cuboid_center: np.ndarray | None = None + cuboid_size: np.ndarray | None = None + cuboid_yaw: float | None = None + cuboid_velocity: np.ndarray | None = None + cuboid_seconds: float | None = None + distance_m: float | None = None + + +@dataclass(frozen=True, slots=True) +class TemporalStabilityBuild: + result_root: Path + profile_sha256: str + report: dict[str, Any] + + +class TemporalStabilizer: + """One-pass bounded state machine for 2D tracks and map-frame cuboids.""" + + def __init__(self, profile: dict[str, Any]) -> None: + self.profile = profile + self.states: dict[int, _TrackState] = {} + self.aliases: dict[int, int] = {} + self.next_id = 1 + self.peak_states = 0 + self.stitched_tracks = 0 + self.held_2d = 0 + self.held_3d = 0 + self.reset_2d = 0 + self.reset_3d = 0 + + def update( + self, + *, + frame_index: int, + session_seconds: float, + objects: list[dict[str, Any]], + ) -> list[dict[str, Any]]: + self._prune(session_seconds) + assigned: set[int] = set() + result: list[dict[str, Any]] = [] + for source in sorted(objects, key=lambda item: int(item["track_id"])): + state, stitched = self._resolve(source, frame_index, assigned) + assigned.add(state.canonical_id) + normalized = self._observe( + state, + source, + frame_index=frame_index, + session_seconds=session_seconds, + stitched=stitched, + ) + result.append(normalized) + + hold_frames = int(self.profile["tracking_2d"]["hold_frames"]) + for state in sorted(self.states.values(), key=lambda value: value.canonical_id): + if state.canonical_id in assigned: + continue + gap = frame_index - state.last_observed_frame + if not 1 <= gap <= hold_frames: + continue + held = json.loads(json.dumps(state.template)) + state.center = state.center + state.velocity + state.last_frame = frame_index + held["track_id"] = state.canonical_id + held["bbox_xyxy"] = _box(state.center, state.size) + held["score"] = max(0.0, state.score * (0.82**gap)) + held["temporal_2d_status"] = f"held-{gap}-frame" + held["temporal_source_track_id"] = min(state.source_ids) + self.held_2d += 1 + if not self._hold_cuboid(held, state, session_seconds): + _clear_cuboid(held, "rejected-temporal-hold-expired-e22") + result.append(held) + + self.peak_states = max(self.peak_states, len(self.states)) + return sorted(result, key=lambda item: int(item["track_id"])) + + def _resolve( + self, + source: dict[str, Any], + frame_index: int, + assigned: set[int], + ) -> tuple[_TrackState, bool]: + source_id = int(source["track_id"]) + canonical = self.aliases.get(source_id) + if canonical is not None and canonical in self.states: + return self.states[canonical], False + + bbox = np.asarray(source["bbox_xyxy"], dtype=np.float64) + center, size = _center_size(bbox) + config = self.profile["tracking_2d"] + best: tuple[float, _TrackState] | None = None + for state in self.states.values(): + gap = frame_index - state.last_observed_frame + if ( + state.canonical_id in assigned + or state.label != str(source["label"]) + or not 1 <= gap <= int(config["stitch_gap_frames"]) + ): + continue + predicted = state.center + state.velocity * gap + predicted_box = np.asarray(_box(predicted, state.size), dtype=np.float64) + iou = _iou(predicted_box, bbox) + scale = max(1.0, math.sqrt(float(np.prod(np.maximum(state.size, 1.0))))) + normalized_distance = float(np.linalg.norm(center - predicted) / scale) + if ( + iou < float(config["stitch_minimum_iou"]) + and normalized_distance + > float(config["stitch_maximum_normalized_center_distance"]) + ): + continue + score = iou - 0.25 * normalized_distance + if best is None or score > best[0]: + best = (score, state) + if best is not None: + state = best[1] + state.source_ids.add(source_id) + self.aliases[source_id] = state.canonical_id + self.stitched_tracks += 1 + return state, True + + canonical_id = source_id + if canonical_id in self.states: + canonical_id = max(self.next_id, max(self.states, default=0) + 1) + self.next_id = max(self.next_id, canonical_id + 1) + state = _TrackState( + canonical_id=canonical_id, + source_ids={source_id}, + label=str(source["label"]), + group=str(source.get("association_group", source["label"])), + center=center.copy(), + size=size.copy(), + raw_center=center.copy(), + velocity=np.zeros(2, dtype=np.float64), + last_frame=frame_index, + last_seconds=0.0, + last_observed_frame=frame_index, + score=float(source["score"]), + template=json.loads(json.dumps(source)), + ) + self.states[canonical_id] = state + self.aliases[source_id] = canonical_id + return state, False + + def _observe( + self, + state: _TrackState, + source: dict[str, Any], + *, + frame_index: int, + session_seconds: float, + stitched: bool, + ) -> dict[str, Any]: + bbox = np.asarray(source["bbox_xyxy"], dtype=np.float64) + observed_center, observed_size = _center_size(bbox) + gap = max(1, frame_index - state.last_observed_frame) + config = self.profile["tracking_2d"] + predicted = state.center + state.velocity * gap + scale = max(1.0, math.sqrt(float(np.prod(np.maximum(state.size, 1.0))))) + innovation = float(np.linalg.norm(observed_center - predicted) / scale) + if innovation > float(config["maximum_normalized_innovation"]): + state.center = observed_center + state.size = observed_size + state.velocity.fill(0.0) + status = "reset-large-innovation" + self.reset_2d += 1 + else: + blend = _adaptive( + innovation, + float(config["adaptive_innovation_low"]), + float(config["adaptive_innovation_high"]), + ) + center_alpha = _lerp( + float(config["center_alpha_low"]), + float(config["center_alpha_high"]), + blend, + ) + size_alpha = _lerp( + float(config["size_alpha_low"]), + float(config["size_alpha_high"]), + blend, + ) + state.center = predicted + center_alpha * (observed_center - predicted) + state.size = state.size + size_alpha * (observed_size - state.size) + observed_velocity = (observed_center - state.raw_center) / gap + velocity_alpha = float(config["velocity_alpha"]) + state.velocity = ( + (1.0 - velocity_alpha) * state.velocity + + velocity_alpha * observed_velocity + ) + status = "stitched-observed" if stitched else "observed" + state.raw_center = observed_center + state.last_frame = frame_index + state.last_observed_frame = frame_index + state.last_seconds = session_seconds + state.score = float(source["score"]) + state.template = json.loads(json.dumps(source)) + + normalized = json.loads(json.dumps(source)) + source_id = int(source["track_id"]) + normalized["track_id"] = state.canonical_id + normalized["temporal_source_track_id"] = source_id + normalized["bbox_xyxy"] = _box(state.center, state.size) + normalized["temporal_2d_status"] = status + self._observe_or_hold_cuboid(normalized, state, session_seconds) + state.template = json.loads(json.dumps(normalized)) + return normalized + + def _observe_or_hold_cuboid( + self, + item: dict[str, Any], + state: _TrackState, + session_seconds: float, + ) -> None: + if str(item.get("cuboid_status", "")).startswith("accepted-"): + self._observe_cuboid(item, state, session_seconds) + elif not self._hold_cuboid(item, state, session_seconds): + return + + def _observe_cuboid( + self, + item: dict[str, Any], + state: _TrackState, + session_seconds: float, + ) -> None: + center = np.asarray(item["cuboid_center_map"], dtype=np.float64) + size = np.asarray(item["cuboid_half_size"], dtype=np.float64) + yaw = _yaw(np.asarray(item["cuboid_quaternion_xyzw"], dtype=np.float64)) + config = self.profile["cuboids_3d"] + status = "observed" + if ( + state.cuboid_center is not None + and state.cuboid_size is not None + and state.cuboid_yaw is not None + and state.cuboid_seconds is not None + ): + dt = max(1e-3, session_seconds - state.cuboid_seconds) + velocity = ( + np.zeros(3, dtype=np.float64) + if state.cuboid_velocity is None + else state.cuboid_velocity + ) + predicted = state.cuboid_center + velocity * dt + innovation = float(np.linalg.norm(center - predicted)) + yaw_delta = _yaw_delta(yaw, state.cuboid_yaw) + if ( + innovation > float(config["maximum_center_innovation_m"]) + or abs(math.degrees(yaw_delta)) + > float(config["maximum_yaw_innovation_degrees"]) + ): + status = "reset-large-innovation" + self.reset_3d += 1 + velocity = np.zeros(3, dtype=np.float64) + else: + raw_velocity = (center - state.cuboid_center) / dt + velocity_alpha = float(config["velocity_alpha"]) + velocity = ( + (1.0 - velocity_alpha) * velocity + + velocity_alpha * raw_velocity + ) + center = predicted + float(config["center_alpha"]) * (center - predicted) + size = state.cuboid_size + float(config["size_alpha"]) * ( + size - state.cuboid_size + ) + yaw = state.cuboid_yaw + float(config["yaw_alpha"]) * yaw_delta + state.cuboid_velocity = velocity + else: + state.cuboid_velocity = np.zeros(3, dtype=np.float64) + state.cuboid_center = center + state.cuboid_size = size + state.cuboid_yaw = yaw + state.cuboid_seconds = session_seconds + distance = item.get("distance_smoothed_m") + if isinstance(distance, int | float) and not isinstance(distance, bool): + state.distance_m = float(distance) + item["cuboid_center_map"] = center.tolist() + item["cuboid_half_size"] = size.tolist() + item["cuboid_quaternion_xyzw"] = _quaternion(yaw) + item["cuboid_status"] = "accepted-temporally-stabilized-e22-v1" + item["temporal_status"] = f"e22-{status}" + + def _hold_cuboid( + self, + item: dict[str, Any], + state: _TrackState, + session_seconds: float, + ) -> bool: + if ( + state.cuboid_center is None + or state.cuboid_size is None + or state.cuboid_yaw is None + or state.cuboid_seconds is None + or session_seconds - state.cuboid_seconds + > float(self.profile["cuboids_3d"]["hold_seconds"]) + ): + return False + age_ms = max(0.0, (session_seconds - state.cuboid_seconds) * 1000.0) + item["cuboid_center_map"] = state.cuboid_center.tolist() + item["cuboid_half_size"] = state.cuboid_size.tolist() + item["cuboid_quaternion_xyzw"] = _quaternion(state.cuboid_yaw) + item["cuboid_status"] = "accepted-temporal-hold-e22-v1" + item["geometry"] = "temporally-held-last-supported-cuboid" + item["distance_smoothed_m"] = state.distance_m + item["temporal_status"] = f"e22-held-{age_ms:.0f}ms" + item["observed_cuboid_center_map"] = None + item["observed_cuboid_half_size"] = None + item["observed_cuboid_quaternion_xyzw"] = None + self.held_3d += 1 + return True + + def _prune(self, now: float) -> None: + maximum_idle = float(self.profile["bounds"]["maximum_track_idle_seconds"]) + stale = [ + key + for key, state in self.states.items() + if state.last_seconds > 0 and now - state.last_seconds > maximum_idle + ] + for key in stale: + state = self.states.pop(key) + for source_id in state.source_ids: + self.aliases.pop(source_id, None) + maximum = int(self.profile["bounds"]["maximum_track_states"]) + if len(self.states) <= maximum: + return + for state in sorted( + self.states.values(), key=lambda value: value.last_seconds + )[: len(self.states) - maximum]: + self.states.pop(state.canonical_id, None) + for source_id in state.source_ids: + self.aliases.pop(source_id, None) + + +def read_profile(path: Path) -> tuple[dict[str, Any], str]: + resolved = path.expanduser().resolve(strict=True) + profile = _read_object(resolved) + source = profile.get("source") + tracking = profile.get("tracking_2d") + cuboids = profile.get("cuboids_3d") + semantic = profile.get("semantic") + bounds = profile.get("bounds") + acceptance = profile.get("acceptance") + authority = profile.get("authority") + if ( + profile.get("schema_version") != PROFILE_SCHEMA + or profile.get("mode") != "recorded-streaming-qualification" + or not all( + isinstance(value, dict) + for value in (source, tracking, cuboids, semantic, bounds, acceptance) + ) + or source.get("source_id") != "sensor.camera.right" + or source.get("resolution") != [800, 600] + or semantic.get("mode") + != "spatially-supported-streaming-hysteresis-v1" + or int(semantic.get("class_count", 0)) != len(TARGET_TAXONOMY) + or authority + != {"commands_enabled": False, "navigation_or_safety_accepted": False} + or not 1 <= int(bounds.get("maximum_track_states", 0)) <= 512 + or int(bounds.get("semantic_history_masks", 0)) != 1 + ): + raise SessionIntegrityError("LAB E22 temporal profile is invalid") + for owner, keys in ( + ( + tracking, + ( + "center_alpha_low", + "center_alpha_high", + "size_alpha_low", + "size_alpha_high", + "velocity_alpha", + ), + ), + (cuboids, ("center_alpha", "size_alpha", "yaw_alpha", "velocity_alpha")), + ): + if any(not 0 < float(owner.get(key, 0)) <= 1 for key in keys): + raise SessionIntegrityError("LAB E22 smoothing coefficient is invalid") + return profile, _sha256(resolved) + + +def stabilize_semantic_masks( + masks: np.ndarray, + *, + minimum_same_label_neighbors: int, +) -> tuple[np.ndarray, dict[str, Any], list[float]]: + if masks.dtype != np.uint8 or masks.ndim != 3 or masks.shape[1:] != (600, 800): + raise SessionIntegrityError("LAB E22 semantic masks are invalid") + output = np.empty_like(masks) + output[0] = masks[0] + baseline_unsupported: list[float] = [] + stabilized_unsupported: list[float] = [] + durations: list[float] = [] + for index in range(1, len(masks)): + started = time.perf_counter() + current = masks[index] + previous = output[index - 1] + support = _same_label_neighbor_count(current) + baseline_change = current != masks[index - 1] + unsupported_baseline = baseline_change & ( + support < minimum_same_label_neighbors + ) + stabilized = current.copy() + hold = (current != previous) & (support < minimum_same_label_neighbors) + stabilized[hold] = previous[hold] + output[index] = stabilized + stabilized_support = _same_label_neighbor_count(stabilized) + unsupported_stabilized = (stabilized != previous) & ( + stabilized_support < minimum_same_label_neighbors + ) + baseline_unsupported.append(float(np.mean(unsupported_baseline))) + stabilized_unsupported.append(float(np.mean(unsupported_stabilized))) + durations.append((time.perf_counter() - started) * 1000.0) + return ( + output, + { + "baseline_unsupported_change_fraction": _percentiles( + baseline_unsupported + ), + "stabilized_unsupported_change_fraction": _percentiles( + stabilized_unsupported + ), + }, + durations, + ) + + +def build_temporal_stability_result( + *, + source: IntegratedPerceptionResult, + lab_job: CameraComputeJob, + lab_pack: Path, + results_root: Path, + profile_path: Path, +) -> TemporalStabilityBuild: + if not source.accepted: + raise SessionIntegrityError("LAB E22 source result is not accepted") + profile, profile_sha256 = read_profile(profile_path) + source_document = _read_object(source.result_root / "result.json") + source_report = _read_object(source.report_path) + source_fusion = _read_jsonl(source.fusion_path) + source_world = _read_jsonl(source.world_path) + source_semantic = _read_jsonl(source.semantic_path) + if len(source_fusion) != source.frame_count or len(source_world) != source.frame_count: + raise SessionIntegrityError("LAB E22 source timelines are incomplete") + + with np.load(source.arrays_path, allow_pickle=False) as arrays: + source_arrays = {name: arrays[name].copy() for name in arrays.files} + semantic_masks, semantic_metrics, semantic_ms = stabilize_semantic_masks( + source_arrays["semantic_masks"], + minimum_same_label_neighbors=int( + profile["semantic"]["minimum_same_label_neighbors"] + ), + ) + + stabilizer = TemporalStabilizer(profile) + fusion_rows: list[dict[str, Any]] = [] + world_rows: list[dict[str, Any]] = [] + frame_ms: list[float] = [] + world_memory: dict[int, dict[str, Any]] = {} + for fusion, world in zip(source_fusion, source_world, strict=True): + started = time.perf_counter() + normalized_fusion = json.loads(json.dumps(fusion)) + objects = stabilizer.update( + frame_index=int(fusion["frame_index"]), + session_seconds=float(fusion["session_seconds"]), + objects=list(fusion["objects"]), + ) + normalized_fusion["objects"] = objects + normalized_fusion["temporal_stability"] = { + "schema_version": "missioncore.e22-temporal-frame/v1", + "state": "bounded-diagnostic", + } + normalized_world = _stabilize_world( + world, + objects, + world_memory, + ) + elapsed = (time.perf_counter() - started) * 1000.0 + normalized_world["delivery"]["temporal_postprocess_ms"] = elapsed + fusion_rows.append(normalized_fusion) + world_rows.append(normalized_world) + frame_ms.append(elapsed) + + semantic_rows = json.loads(json.dumps(source_semantic)) + for row, mask in zip(semantic_rows, semantic_masks, strict=True): + row["mask_sha256"] = hashlib.sha256(mask.tobytes()).hexdigest() + counts = np.bincount(mask.ravel(), minlength=len(TARGET_TAXONOMY)) + row["class_pixels"] = { + TARGET_TAXONOMY[index]: int(counts[index]) + for index in range(1, len(TARGET_TAXONOMY)) + if counts[index] + } + row["temporal_status"] = "e22-spatially-supported-hysteresis" + + baseline = _quality_metrics(source_fusion, source_arrays["semantic_masks"]) + stabilized = _quality_metrics(fusion_rows, semantic_masks) + reductions = { + "tracking_2d_acceleration_p95_fraction": _reduction( + baseline["tracking_2d"]["normalized_acceleration"]["p95"], + stabilized["tracking_2d"]["normalized_acceleration"]["p95"], + ), + "cuboid_center_step_p95_fraction": _reduction( + baseline["cuboids_3d"]["center_step_m"]["p95"], + stabilized["cuboids_3d"]["center_step_m"]["p95"], + ), + "cuboid_yaw_step_p95_fraction": _reduction( + baseline["cuboids_3d"]["yaw_step_degrees"]["p95"], + stabilized["cuboids_3d"]["yaw_step_degrees"]["p95"], + ), + "semantic_unsupported_change_fraction": _reduction( + semantic_metrics["baseline_unsupported_change_fraction"]["mean"], + semantic_metrics["stabilized_unsupported_change_fraction"]["mean"], + ), + } + runtime = { + "camera_frame_processing_ms": _percentiles(frame_ms), + "semantic_frame_processing_ms": _percentiles(semantic_ms), + "peak_track_states": stabilizer.peak_states, + "maximum_track_states": int(profile["bounds"]["maximum_track_states"]), + "semantic_history_masks": int(profile["bounds"]["semantic_history_masks"]), + "estimated_state_bytes_upper_bound": ( + int(profile["bounds"]["maximum_track_states"]) * 4096 + + 600 * 800 + ), + } + criteria = profile["acceptance"] + checks = { + "minimum_2d_acceleration_reduction": reductions[ + "tracking_2d_acceleration_p95_fraction" + ] + >= float(criteria["minimum_2d_acceleration_p95_reduction_fraction"]), + "minimum_3d_center_reduction": reductions[ + "cuboid_center_step_p95_fraction" + ] + >= float(criteria["minimum_3d_center_step_p95_reduction_fraction"]), + "minimum_3d_yaw_reduction": reductions[ + "cuboid_yaw_step_p95_fraction" + ] + >= float(criteria["minimum_3d_yaw_step_p95_reduction_fraction"]), + "minimum_semantic_unsupported_change_reduction": reductions[ + "semantic_unsupported_change_fraction" + ] + >= float(criteria["minimum_semantic_unsupported_change_reduction_fraction"]), + "camera_frame_processing_p95": runtime["camera_frame_processing_ms"]["p95"] + <= float(criteria["maximum_camera_frame_processing_p95_ms"]), + "semantic_frame_processing_p95": runtime[ + "semantic_frame_processing_ms" + ]["p95"] + <= float(criteria["maximum_semantic_frame_processing_p95_ms"]), + "track_state_bound": stabilizer.peak_states + <= int(criteria["maximum_track_states_observed"]), + "source_payloads_immutable": True, + "diagnostic_authority_only": True, + } + accepted = all(checks.values()) + + source_identity = source_document["identity"] + selection = json.loads(json.dumps(source_identity["selection"])) + configuration = { + "pipeline": PIPELINE_ID, + "profile_sha256": profile_sha256, + "profile": profile, + "source_result_id": source.result_id, + "source_result_sha256": _sha256(source.result_root / "result.json"), + "source_session_id": source.job.session_id, + "processing": "single-pass-bounded-state-no-lookahead", + } + identity = { + "schema_version": "missioncore.e10-integrated-perception-identity/v1", + "job_id": lab_job.job_id, + "input_sha256": lab_job.input_sha256, + "session_id": lab_job.session_id, + "source_id": lab_job.source_id, + "lidar_pack_id": lab_pack.name, + "selection": selection, + "configuration": configuration, + "models": source_identity["models"], + } + identity_sha256 = hashlib.sha256(_canonical(identity)).hexdigest() + result_id = f"e10-integrated-perception-{identity_sha256}" + destination = results_root / result_id + created_at = datetime.now(UTC).isoformat(timespec="milliseconds").replace( + "+00:00", "Z" + ) + report = { + "schema_version": "missioncore.e10-integrated-perception-report/v1", + "result_id": result_id, + "created_at_utc": created_at, + "state": "accepted" if accepted else "rejected", + "ground_truth": False, + "identity": identity, + "acceptance": { + "accepted": accepted, + "navigation_or_safety_accepted": False, + "checks": checks, + }, + "metrics": { + "baseline": baseline, + "stabilized": stabilized, + "reductions": reductions, + "semantic_filter": semantic_metrics, + "runtime": runtime, + "events": { + "stitched_tracklets": stabilizer.stitched_tracks, + "held_2d_frames": stabilizer.held_2d, + "held_3d_cuboids": stabilizer.held_3d, + "reset_2d_large_innovation": stabilizer.reset_2d, + "reset_3d_large_innovation": stabilizer.reset_3d, + }, + "source_realtime_envelope": source_report.get("metrics", {}), + }, + "runtime": { + "executor": "local-deterministic-postprocessor", + "numpy": np.__version__, + "streaming_equivalent": True, + "lookahead_frames": 0, + }, + "limitations": [ + "Recorded diagnostic qualification only; no navigation or command authority.", + ( + "Tracklet stitching is class-and-image-geometry based and is not " + "ground truth re-identification." + ), + ( + "Short 2D/3D holds are explicitly provenance-marked and must not " + "be treated as fresh measurements." + ), + ( + "Semantic hysteresis suppresses unsupported label islands but " + "does not perform optical-flow propagation." + ), + ( + "The source E21 detector/semantic inference and all raw sensor " + "payloads remain unchanged." + ), + ], + } + if destination.exists(): + existing = _read_object(destination / "result.json") + if existing.get("identity") != identity: + raise SessionIntegrityError("LAB E22 result id collides") + persisted_report = _read_object(destination / "run-report.json") + return TemporalStabilityBuild(destination, profile_sha256, persisted_report) + + results_root.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = results_root / f".{result_id}.{secrets.token_hex(12)}.tmp" + staging.mkdir(mode=0o700) + try: + semantic_path = staging / "semantic-frames.jsonl" + fusion_path = staging / "fusion-frames.jsonl" + world_path = staging / "world-state.jsonl" + arrays_path = staging / "transient-perception.npz" + gpu_path = staging / "gpu-telemetry.jsonl" + report_path = staging / "run-report.json" + _write_jsonl(semantic_path, semantic_rows) + _write_jsonl(fusion_path, fusion_rows) + _write_jsonl(world_path, world_rows) + box_offsets = [0] + centers: list[list[float]] = [] + sizes: list[list[float]] = [] + quaternions: list[list[float]] = [] + colors: list[list[int]] = [] + for row in fusion_rows: + for item in row["objects"]: + if not str(item.get("cuboid_status", "")).startswith("accepted-"): + continue + centers.append(item["cuboid_center_map"]) + sizes.append(item["cuboid_half_size"]) + quaternions.append(item["cuboid_quaternion_xyzw"]) + colors.append(_cuboid_color(item)) + box_offsets.append(len(centers)) + np.savez_compressed( + arrays_path, + frame_times_ns=source_arrays["frame_times_ns"], + semantic_frame_indices=source_arrays["semantic_frame_indices"], + semantic_masks=semantic_masks, + support_offsets=source_arrays["support_offsets"], + support_points=source_arrays["support_points"], + support_colors=source_arrays["support_colors"], + box_offsets=np.asarray(box_offsets, dtype=np.int64), + box_centers=np.asarray(centers, dtype=np.float32).reshape((-1, 3)), + box_half_sizes=np.asarray(sizes, dtype=np.float32).reshape((-1, 3)), + box_quaternions=np.asarray(quaternions, dtype=np.float32).reshape((-1, 4)), + box_colors=np.asarray(colors, dtype=np.uint8).reshape((-1, 4)), + ) + os.link( + source.result_root / "gpu-telemetry.jsonl", + gpu_path, + follow_symlinks=False, + ) + write_json_atomic(report_path, report) + artifacts = [ + _artifact("e10-semantic-frames", semantic_path, "missioncore.e10-semantic-frame/v1"), + _artifact("e10-fusion-frames", fusion_path, "missioncore.e10-fusion-frame/v1"), + _artifact("e10-world-state", world_path, "missioncore.live-perception-world-state/v1"), + _artifact("e10-transient-perception", arrays_path, None), + _artifact("worker-gpu-telemetry", gpu_path, None), + _artifact( + "e10-run-report", + report_path, + "missioncore.e10-integrated-perception-report/v1", + ), + ] + write_json_atomic( + staging / "result.json", + { + "schema_version": "missioncore.e10-integrated-perception-result/v1", + "result_id": result_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": created_at, + "ground_truth": False, + "publication_scope": "recorded-integrated-realtime-qualification-only", + "acceptance_state": "accepted" if accepted else "rejected", + "frames_processed": source.frame_count, + "artifacts": artifacts, + }, + ) + os.replace(staging, destination) + finally: + if staging.exists(): + shutil.rmtree(staging) + return TemporalStabilityBuild(destination, profile_sha256, report) + + +def _stabilize_world( + source: dict[str, Any], + fusion_objects: list[dict[str, Any]], + memory: dict[int, dict[str, Any]], +) -> dict[str, Any]: + world = json.loads(json.dumps(source)) + source_objects = { + int(item["track_id"]): item + for item in source.get("objects", []) + if isinstance(item, dict) and isinstance(item.get("track_id"), int) + } + objects: list[dict[str, Any]] = [] + for fusion in fusion_objects: + if not str(fusion.get("cuboid_status", "")).startswith("accepted-"): + continue + canonical = int(fusion["track_id"]) + source_id = int(fusion.get("temporal_source_track_id", canonical)) + template = source_objects.get(source_id) or memory.get(canonical) or {} + item = json.loads(json.dumps(template)) + item.update( + { + "track_id": canonical, + "class": str(fusion.get("association_group", "object")), + "detector_label": str(fusion.get("label", "object")), + "confidence": float(fusion.get("score", 0.0)), + "position_map_m": fusion["cuboid_center_map"], + "orientation_map_xyzw": fusion["cuboid_quaternion_xyzw"], + "size_m": [ + 2.0 * float(value) for value in fusion["cuboid_half_size"] + ], + "range_m": fusion.get("distance_smoothed_m"), + "support_points": int(fusion.get("clustered_points", 0)), + "geometry": fusion.get("geometry"), + "temporal_status": fusion.get("temporal_status"), + } + ) + if "held" in str(fusion.get("temporal_status", "")): + item["velocity_status"] = "temporally-held-diagnostic" + memory[canonical] = json.loads(json.dumps(item)) + objects.append(item) + world["objects"] = objects + world["object_count"] = len(objects) + world.setdefault("delivery", {})["temporal_stability"] = "e22-bounded-diagnostic" + return world + + +def _quality_metrics( + rows: list[dict[str, Any]], + masks: np.ndarray, +) -> dict[str, Any]: + tracks: dict[tuple[int, str], list[tuple[int, np.ndarray]]] = {} + cuboids: dict[ + tuple[int, str], list[tuple[int, np.ndarray, np.ndarray, float]] + ] = {} + for row in rows: + for item in row["objects"]: + key = (int(item["track_id"]), str(item["label"])) + tracks.setdefault(key, []).append( + (int(row["frame_index"]), np.asarray(item["bbox_xyxy"], dtype=np.float64)) + ) + if str(item.get("cuboid_status", "")).startswith("accepted-"): + cuboids.setdefault(key, []).append( + ( + int(row["frame_index"]), + np.asarray(item["cuboid_center_map"], dtype=np.float64), + np.asarray(item["cuboid_half_size"], dtype=np.float64), + _yaw( + np.asarray( + item["cuboid_quaternion_xyzw"], dtype=np.float64 + ) + ), + ) + ) + acceleration: list[float] = [] + size_steps: list[float] = [] + for values in tracks.values(): + previous_step: np.ndarray | None = None + for (left_frame, left_box), (right_frame, right_box) in zip( + values, values[1:], strict=False + ): + if right_frame - left_frame != 1: + previous_step = None + continue + left_center, left_size = _center_size(left_box) + right_center, right_size = _center_size(right_box) + step = (right_center - left_center) / np.maximum(left_size, 1.0) + size_steps.append( + float(np.linalg.norm((right_size - left_size) / np.maximum(left_size, 1.0))) + ) + if previous_step is not None: + acceleration.append(float(np.linalg.norm(step - previous_step))) + previous_step = step + center_steps: list[float] = [] + cuboid_size_steps: list[float] = [] + yaw_steps: list[float] = [] + gaps = 0 + for values in cuboids.values(): + for left, right in zip(values, values[1:], strict=False): + if right[0] - left[0] != 1: + gaps += 1 + continue + center_steps.append(float(np.linalg.norm(right[1] - left[1]))) + cuboid_size_steps.append(float(np.linalg.norm(right[2] - left[2]))) + yaw_steps.append(abs(math.degrees(_yaw_delta(right[3], left[3])))) + semantic_change = [ + float(np.mean(right != left)) + for left, right in zip(masks, masks[1:], strict=False) + ] + lengths = [len(values) for values in tracks.values()] + return { + "tracking_2d": { + "track_count": len(tracks), + "short_track_count_le_2": sum(length <= 2 for length in lengths), + "normalized_acceleration": _percentiles(acceleration), + "normalized_size_step": _percentiles(size_steps), + }, + "cuboids_3d": { + "track_count": len(cuboids), + "accepted_count": sum(len(values) for values in cuboids.values()), + "gap_count": gaps, + "center_step_m": _percentiles(center_steps), + "half_size_step_m": _percentiles(cuboid_size_steps), + "yaw_step_degrees": _percentiles(yaw_steps), + }, + "semantic": { + "mask_count": int(len(masks)), + "change_fraction": _percentiles(semantic_change), + }, + } + + +def _same_label_neighbor_count(mask: np.ndarray) -> np.ndarray: + padded = np.pad(mask, 1, mode="edge") + count = np.zeros(mask.shape, dtype=np.uint8) + for y in range(3): + for x in range(3): + if y == 1 and x == 1: + continue + count += padded[y : y + mask.shape[0], x : x + mask.shape[1]] == mask + return count + + +def _center_size(box: np.ndarray) -> tuple[np.ndarray, np.ndarray]: + return (box[:2] + box[2:]) * 0.5, np.maximum(box[2:] - box[:2], 1.0) + + +def _box(center: np.ndarray, size: np.ndarray) -> list[float]: + half = np.maximum(size, 1.0) * 0.5 + values = np.concatenate((center - half, center + half)) + values[[0, 2]] = np.clip(values[[0, 2]], 0.0, 799.0) + values[[1, 3]] = np.clip(values[[1, 3]], 0.0, 599.0) + return [round(float(value), 6) for value in values] + + +def _iou(left: np.ndarray, right: np.ndarray) -> float: + lower = np.maximum(left[:2], right[:2]) + upper = np.minimum(left[2:], right[2:]) + intersection = float(np.prod(np.maximum(upper - lower, 0.0))) + left_area = float(np.prod(np.maximum(left[2:] - left[:2], 0.0))) + right_area = float(np.prod(np.maximum(right[2:] - right[:2], 0.0))) + union = left_area + right_area - intersection + return 0.0 if union <= 0 else intersection / union + + +def _adaptive(value: float, lower: float, upper: float) -> float: + if upper <= lower: + return 1.0 + return min(1.0, max(0.0, (value - lower) / (upper - lower))) + + +def _lerp(lower: float, upper: float, fraction: float) -> float: + return lower + (upper - lower) * fraction + + +def _yaw(quaternion: np.ndarray) -> float: + return 2.0 * math.atan2(float(quaternion[2]), float(quaternion[3])) + + +def _yaw_delta(value: float, reference: float) -> float: + return ((value - reference + math.pi * 0.5) % math.pi) - math.pi * 0.5 + + +def _quaternion(yaw: float) -> list[float]: + return [0.0, 0.0, math.sin(yaw * 0.5), math.cos(yaw * 0.5)] + + +def _clear_cuboid(item: dict[str, Any], status: str) -> None: + item["cuboid_status"] = status + for key in ( + "cuboid_center_map", + "cuboid_half_size", + "cuboid_quaternion_xyzw", + "distance_smoothed_m", + ): + item[key] = None + item["geometry"] = "none" + + +def _reduction(baseline: int | float, stabilized: int | float) -> float: + baseline_value = float(baseline) + if baseline_value <= 0: + return 0.0 + return (baseline_value - float(stabilized)) / baseline_value + + +def _percentiles(values: list[float]) -> dict[str, float | int]: + finite = sorted(float(value) for value in values if math.isfinite(float(value))) + if not finite: + return {"count": 0, "mean": 0.0, "p50": 0.0, "p95": 0.0, "p99": 0.0, "max": 0.0} + + def percentile(fraction: float) -> float: + return finite[min(len(finite) - 1, round((len(finite) - 1) * fraction))] + + return { + "count": len(finite), + "mean": statistics.fmean(finite), + "p50": percentile(0.50), + "p95": percentile(0.95), + "p99": percentile(0.99), + "max": finite[-1], + } + + +def _cuboid_color(item: dict[str, Any]) -> list[int]: + key = f"e22:{item.get('association_group', 'object')}:{item.get('track_id', 0)}" + digest = hashlib.sha256(key.encode()).digest() + return [64 + digest[0] % 176, 64 + digest[1] % 176, 64 + digest[2] % 176, 88] + + +def _read_object(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8-sig")) + if not isinstance(value, dict): + raise SessionIntegrityError(f"JSON root is not an object: {path}") + return value + + +def _read_jsonl(path: Path) -> list[dict[str, Any]]: + values: list[dict[str, Any]] = [] + with path.open(encoding="utf-8") as stream: + for line in stream: + value = json.loads(line) + if not isinstance(value, dict): + raise SessionIntegrityError(f"JSONL row is not an object: {path}") + values.append(value) + return values + + +def _write_jsonl(path: Path, values: list[dict[str, Any]]) -> None: + with path.open("x", encoding="utf-8") as stream: + for value in values: + stream.write( + json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + "\n" + ) + stream.flush() + os.fsync(stream.fileno()) + os.chmod(path, 0o600) + + +def _artifact(kind: str, path: Path, schema: str | None) -> dict[str, Any]: + return { + "kind": kind, + "path": path.name, + "schema_version": schema, + "byte_length": path.stat().st_size, + "sha256": _sha256(path), + } + + +def _canonical(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() diff --git a/src/k1link/device_plugins/xgrids_k1/cli.py b/src/k1link/device_plugins/xgrids_k1/cli.py index ae4833e..97fe81c 100644 --- a/src/k1link/device_plugins/xgrids_k1/cli.py +++ b/src/k1link/device_plugins/xgrids_k1/cli.py @@ -20,6 +20,7 @@ from k1link.artifacts import write_json_atomic from k1link.compute import ( prepare_camera_compute_job, publish_e21_lab_instance, + publish_e22_lab_instance, publish_integrated_lab_instance, ) from k1link.device_plugins.xgrids_k1.analyze import ( @@ -433,6 +434,69 @@ def publish_e21_lab( ) +@lab_app.command("publish-e22") +def publish_e22_lab( + result: Annotated[ + Path, + typer.Option( + exists=True, + file_okay=False, + readable=True, + resolve_path=True, + help="Accepted integrated result used as the immutable E22 baseline.", + ), + ], + profile: Annotated[ + Path, + typer.Option( + exists=True, + dir_okay=False, + readable=True, + resolve_path=True, + help="Bounded E22 temporal-stability profile.", + ), + ], + session_id: Annotated[ + str, + typer.Option("--session-id", help="New immutable LAB session id."), + ], + lab_id: Annotated[ + str, + typer.Option("--lab-id", help="LAB marker, for example 'LAB E22'."), + ], + display_name: Annotated[ + str, + typer.Option("--display-name", help="Operator-facing saved-session title."), + ], +) -> None: + """Build and publish a bounded 2D/3D/semantic temporal-stability run.""" + + repository_root = Path(__file__).resolve().parents[4] + try: + published = publish_e22_lab_instance( + repository_root=repository_root, + source_result_root=result, + profile_path=profile, + lab_session_id=session_id, + lab_id=lab_id, + display_name=display_name, + ) + except (OSError, SessionIntegrityError, ValueError) as exc: + console.print(f"[red]E22 LAB publication failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + reductions = published.build.report["metrics"]["reductions"] + console.print( + "[green]E22 LAB instance published.[/green] " + f"session={published.binding.session_id}; " + f"source={published.binding.source_session_id}; " + f"result={published.binding.result_id}; " + f"2d_reduction={reductions['tracking_2d_acceleration_p95_fraction']:.3f}; " + f"3d_center_reduction={reductions['cuboid_center_step_p95_fraction']:.3f}; " + f"3d_yaw_reduction={reductions['cuboid_yaw_step_p95_fraction']:.3f}; " + "source_payloads_mutated=false" + ) + + @app.command("serve") def serve_console( port: Annotated[ diff --git a/tests/test_e22_temporal_stability.py b/tests/test_e22_temporal_stability.py new file mode 100644 index 0000000..af1fb20 --- /dev/null +++ b/tests/test_e22_temporal_stability.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import numpy as np + +from k1link.compute.temporal_stability import ( + TemporalStabilizer, + read_profile, + stabilize_semantic_masks, +) + + +def _profile() -> dict[str, object]: + root = Path(__file__).resolve().parents[1] + profile, digest = read_profile( + root / "experiments" / "perception" / "e22_temporal_stability_profile.json" + ) + assert len(digest) == 64 + return profile + + +def _object(track_id: int, box: list[float]) -> dict[str, object]: + return { + "association_group": "vehicle", + "bbox_xyxy": box, + "candidate_projected_points": 20, + "clustered_points": 14, + "completion_fraction": 0.9, + "cuboid_center_map": [10.0, 2.0, 0.8], + "cuboid_half_size": [2.25, 0.925, 0.775], + "cuboid_quaternion_xyzw": [0.0, 0.0, 0.0, 1.0], + "cuboid_status": "accepted-class-prior-amodal-v1", + "distance_median_m": 10.2, + "distance_p10_m": 9.8, + "distance_smoothed_m": 10.1, + "geometry": "class-prior-completed-from-visible-lidar-support", + "ground_rejected_points": 0, + "ground_z_map": 0.0, + "label": "car", + "observed_cuboid_center_map": [10.0, 2.0, 0.8], + "observed_cuboid_half_size": [2.25, 0.925, 0.775], + "observed_cuboid_quaternion_xyzw": [0.0, 0.0, 0.0, 1.0], + "orientation_source": "support-pca", + "pre_ground_clustered_points": 14, + "score": 0.9, + "semantic_compatible_points": 18, + "support_coverage_fraction": 0.8, + "support_ground_z_map": None, + "temporal_status": "confirmed", + "track_id": track_id, + } + + +def test_e22_profile_is_bounded_and_has_no_control_authority() -> None: + profile = _profile() + + assert profile["bounds"] == { + "maximum_track_states": 128, + "maximum_track_idle_seconds": 1.25, + "semantic_history_masks": 1, + } + assert profile["authority"] == { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + } + + +def test_e22_stitches_one_tracklet_and_marks_short_holds() -> None: + stabilizer = TemporalStabilizer(_profile()) + first = stabilizer.update( + frame_index=0, + session_seconds=0.0, + objects=[_object(10, [100.0, 100.0, 200.0, 200.0])], + ) + second = stabilizer.update( + frame_index=1, + session_seconds=0.1, + objects=[_object(11, [102.0, 100.0, 202.0, 200.0])], + ) + held = stabilizer.update( + frame_index=2, + session_seconds=0.2, + objects=[], + ) + + assert first[0]["track_id"] == 10 + assert second[0]["track_id"] == 10 + assert second[0]["temporal_source_track_id"] == 11 + assert second[0]["temporal_2d_status"] == "stitched-observed" + assert held[0]["temporal_2d_status"] == "held-1-frame" + assert held[0]["cuboid_status"] == "accepted-temporal-hold-e22-v1" + assert stabilizer.stitched_tracks == 1 + assert stabilizer.held_2d == 1 + assert stabilizer.peak_states <= 128 + + +def test_e22_semantic_hysteresis_removes_island_but_keeps_supported_region() -> None: + masks = np.zeros((2, 600, 800), dtype=np.uint8) + masks[1, 20, 20] = 4 + masks[1, 100:110, 100:110] = 7 + + stabilized, metrics, durations = stabilize_semantic_masks( + masks, + minimum_same_label_neighbors=3, + ) + + assert stabilized[1, 20, 20] == 0 + assert np.all(stabilized[1, 102:108, 102:108] == 7) + assert ( + metrics["stabilized_unsupported_change_fraction"]["mean"] + < metrics["baseline_unsupported_change_fraction"]["mean"] + ) + assert len(durations) == 1 + assert json.dumps(metrics, allow_nan=False)