diff --git a/apps/control-station/src/styles/spatial.css b/apps/control-station/src/styles/spatial.css index a0ff329..cd68519 100644 --- a/apps/control-station/src/styles/spatial.css +++ b/apps/control-station/src/styles/spatial.css @@ -13,11 +13,10 @@ display: flex; min-width: 0; align-items: center; - justify-content: space-between; + justify-content: flex-start; gap: 1rem; } -.spatial-toolbar__mode, .spatial-toolbar__actions { display: flex; min-width: 0; @@ -39,39 +38,6 @@ border-radius: 999px; } -.spatial-toolbar__perception-status, -.spatial-toolbar__perception-retry { - display: inline-flex; - min-width: 0; - align-items: center; - gap: 0.4rem; - border: 1px solid rgb(255 255 255 / 0.07); - border-radius: 999px; - background: rgb(255 255 255 / 0.035); - padding: 0.42rem 0.62rem; - color: var(--nodedc-text-muted); - font: inherit; - font-size: 0.58rem; - white-space: nowrap; -} - -.spatial-toolbar__perception-status .busy-indicator { - width: 0.72rem; - height: 0.72rem; - flex-basis: 0.72rem; -} - -.spatial-toolbar__perception-retry { - color: var(--nodedc-text-primary); - cursor: pointer; -} - -.spatial-toolbar__perception-retry:hover, -.spatial-toolbar__perception-retry:focus-visible { - background: rgb(255 255 255 / 0.09); - outline: none; -} - .spatial-viewport-shell { position: relative; min-width: 0; @@ -369,6 +335,58 @@ .scene-selection small { color: var(--nodedc-text-muted); font-size: 0.55rem; } .scene-selection strong { overflow: hidden; font-size: 0.66rem; text-overflow: ellipsis; white-space: nowrap; } +.scene-operation-status-stack { + position: absolute; + z-index: 11; + right: 0.85rem; + bottom: 7.15rem; + display: grid; + justify-items: end; + gap: 0.3rem; +} + +.scene-operation-status { + display: inline-flex; + max-width: min(24rem, calc(100% - 1.7rem)); + align-items: center; + gap: 0.42rem; + border: 1px solid rgb(255 255 255 / 0.07); + border-radius: 999px; + background: rgb(9 10 13 / 0.7); + color: var(--nodedc-text-muted); + padding: 0.42rem 0.65rem; + font: inherit; + font-size: 0.52rem; + font-weight: 680; + line-height: 1; + white-space: nowrap; + backdrop-filter: blur(14px); +} + +.scene-operation-status .busy-indicator { + width: 0.66rem; + height: 0.66rem; + flex-basis: 0.66rem; + border-width: 1px; +} + +.scene-operation-status--action { + color: var(--nodedc-text-secondary); + cursor: pointer; +} + +.scene-operation-status--action:hover, +.scene-operation-status--action:focus-visible { + border-color: rgb(255 255 255 / 0.14); + background: rgb(20 21 25 / 0.82); + color: var(--nodedc-text-primary); + outline: none; +} + +.scene-operation-status--error { + color: rgb(var(--nodedc-danger-rgb)); +} + .scene-timeline { right: 0.75rem; bottom: 0.75rem; diff --git a/apps/control-station/src/workspaces/Workspaces.tsx b/apps/control-station/src/workspaces/Workspaces.tsx index ad01795..1e19e26 100644 --- a/apps/control-station/src/workspaces/Workspaces.tsx +++ b/apps/control-station/src/workspaces/Workspaces.tsx @@ -536,9 +536,6 @@ function SpatialWorkspace({ data-focused={pointCloudFocused || floatingSourceMaximized ? "true" : undefined} >
-
- СЦЕНА 3D · RERUN -
{recordedPerceptionSupported || livePerceptionAvailable ? (
@@ -598,33 +595,6 @@ function SpatialWorkspace({
) : null} - {recordedSource && perceptionLoad.phase === "loading" ? ( -
-
- ) : null} - {recordedSource && perceptionLoad.phase === "error" ? ( - - ) : null} - {recordedSource && pointColorLoad.phase === "loading" ? ( -
-
- ) : null} - {recordedSource && pointColorLoad.phase === "error" ? ( -
- {pointColorLoad.message} -
- ) : null} {recordedSource && presentedViewerStatus === "ready" ? ( + ) : null} + {pointColorLoad.phase === "loading" ? ( +
+
+ ) : null} + {pointColorLoad.phase === "error" ? ( +
+ {pointColorLoad.message} +
+ ) : null} +
+ ) : null} + {presentedViewerStatus === "ready" && !floatingSourceMaximized ? (
Колесо · зум к курсору diff --git a/experiments/perception/LAB_E23_REPORT_2026-07-24.md b/experiments/perception/LAB_E23_REPORT_2026-07-24.md new file mode 100644 index 0000000..b2e80c3 --- /dev/null +++ b/experiments/perception/LAB_E23_REPORT_2026-07-24.md @@ -0,0 +1,276 @@ +# LAB E23 — inline temporal stability at 1× + +Date: 2026-07-24 + +Final accepted run: **LAB E23.2** + +Status: **accepted for bounded shadow real-time qualification** + +Control authority: **disabled** + +## Executive result + +LAB E23 answers one narrow but important question: can the temporal stabilization used to reduce 2D, semantic, and 3D jitter execute inside the warm perception worker while a recorded sensor stream is paced at real time? + +The answer is **yes for this 60-second urban RAVNOVES00 slice on the RTX 4090 worker**. + +- The source replay ran at `1.0×`, without look-ahead: 59.994 s of source time in 60.241 s of wall time. +- The stabilizer ran inline after fusion and before world/result publication. +- Its P95 overhead was 1.721 ms for 2D/3D and 12.307 ms for a semantic frame. +- Detector throughput was 9.894 FPS for an approximately 10 FPS camera stream. +- Semantic throughput was 2.009 FPS by design: one semantic inference for every five camera frames. +- World-state age P95 was 185.375 ms. +- All runtime, bounded-state, memory, authority, and final quality checks passed. +- The result was published as a new immutable saved session, not written over an earlier LAB. + +This is not yet permission to drive a vehicle. It is a shadow diagnostic qualification on a recorded stream. The physical K1 live ingress, hardware-clock synchronization, forest-domain models, and navigation/safety acceptance remain separate gates. + +## Runs and disposition + +### LAB E23.1 + +- Worker result: `e15-shadow-inference-37381767fb06928951aa173c3e2b43a81b8e492916e4ac8622d7d78d7a475f8e` +- Profile SHA-256: `a22de018626e11a2c4d18d87d6b682846921f72c6cd9f290f20db9a71218e1e3` +- Runtime envelope: passed. +- Final raw-versus-inline quality audit: failed. +- Failure: 3D cuboid center-step P95 reduction was 17.10%, below the required 20%. +- Publication: rejected; no saved LAB instance was created. + +This distinction is intentional. Passing timing and memory gates does not allow a run to be called successful when the requested quality effect was not demonstrated. + +### LAB E23.2 + +- Worker result: `e15-shadow-inference-32a95297664241f7102b31e064ab0c168bbd97d4f5cc152892fd378d58617ffb` +- Profile SHA-256: `86238c369aa2ec0f5de50a2c676faf9b11884e1355f8cccf416125b5f73fd38e` +- Published result: `e10-integrated-perception-f3cefd9464c8b0ed19bb0c8949c1d890b55bf57353850ba155c4dc8d7d90c371` +- Saved session: `lab-e23-2-86238c36` +- Display label: `LAB E23.2 · Inline temporal 1× · synchronized 60s · 86238c36` +- Runtime envelope: passed. +- Final raw-versus-inline quality audit: passed. +- Browser playback and all visual layers: verified. + +The only configuration adjustment after E23.1 was bounded and measured. A deterministic sweep over the raw E23.1 fusion artifact evaluated several 3D center coefficients; `0.32` was selected to clear the quality gate while avoiding an unnecessarily slow response to real rig motion. E23.2 was then rerun from the source at 1×. The sweep result itself was not treated as acceptance evidence. + +## Experiment boundary + +### What this run proves + +1. The full temporal stage can execute inside the persistent warm worker. +2. The stage uses only current and bounded past state; there is no future-frame look-ahead. +3. The camera, detector, semantic, LiDAR fusion, 2D tracking, 3D cuboids, and world-state publication can operate together inside the measured 1× envelope. +4. State and telemetry collections stay explicitly bounded. +5. Inline output materially reduces the selected jitter metrics. +6. The exact accepted result can be replayed in Mission Core as a live 60-second session with synchronized camera, 2D, segmentation, point cloud, and 3D cuboids. + +### What this run does not prove + +1. It does not prove a physical K1 live connection is stable for an extended mission. +2. It does not prove cross-device hardware timestamp accuracy. The replay is paced by recorded host-arrival time. +3. It does not validate COCO/Cityscapes models for forest obstacles or safety decisions. +4. It does not grant command, navigation, collision-avoidance, or safety authority. +5. It does not establish performance on the final onboard computer unless that computer is the measured RTX 4090 worker. +6. It does not establish ten-hour memory stability; this run is 60 seconds. + +## Input data + +| Item | Value | +|---|---| +| Source session | `20260720T065719Z_viewer_live` | +| Source mode | recorded source paced near-live | +| Speed | `1.0×` | +| Look-ahead | disabled | +| Selected source span | 59.994249 s | +| Wall time | 60.240711 s | +| Camera | `sensor.camera.right` | +| Calibration slot | `camera_1` | +| Resolution | 800 × 600 | +| Camera frames | 601 | +| LiDAR events | 585 | +| Pose events | 600 | +| Camera index SHA-256 | `e029815a60ad9fbfedb6169142c7449df2b119a51d1ce001f08806e04eb0be14` | +| Raw MQTT SHA-256 | `70da0edad5cbf0e89b6e2355c0aac294f8c36a67249cd33b8918353fd88c83af` | +| Calibration SHA-256 | `05f3ad9b38b3a4fc95388a8ec83da83c745e217709e51787b3d5aad0969f6fa9` | +| Valid-FOV pixels | 270,606 / 480,000 | +| Valid-FOV mask SHA-256 | `a40cee06b7c6f69b6a09a11563dcfd237f3de833b1ccd31459e66692e528ba63` | + +Payload loading was streaming and bounded to one event per modality at the source adapter. No complete video, LiDAR stream, or MQTT capture was accumulated in RAM. + +## Software and model identity + +Pipeline: `warm-worker-inline-bounded-temporal-2d-3d-semantic/v1` + +| Component | Identity | +|---|---| +| Worker package | `e15-worker-package-dbf55ccb75664b778a2e0f5d34284af10ec0971945bbbee67b27bc264b765b51` | +| Worker package manifest SHA-256 | `d3089e241286b871ff4e9db569df10222bf2feb07500e10eb641ecce88ace868` | +| Projection pack | `e15-live-projection-713fa7344805b7a592d12e949a6db38116e5370f768bde1cf9144fab5f93f0ae` | +| Detector | YOLOX-S, COCO-80, ONNX | +| Detector model SHA-256 | `c5c2d13e59ae883e6af3b45daea64af4833a4951c92d116ec270d9ddbe998063` | +| Semantic model | `tue-mps/cityscapes_semantic_eomt_large_1024` | +| Semantic revision | `8d6b6d1a3f7b50d441afd7d247c2ed10db186e8f` | +| Semantic weights SHA-256 | `c265da9a74f58f5c3f4826d23ca4ca78beac0b106cca5842beca61580de5b782` | +| Semantic precision | FP16 autocast | +| Python | 3.12.3 | +| PyTorch | 2.13.0+cu130 | +| Transformers | 4.57.6 | +| NumPy | 1.26.4 | +| SciPy | 1.16.3 | +| Triton image | `nvcr.io/nvidia/tritonserver:26.06-py3@sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794` | + +Profile identities: + +- detector: `819bdfb0da521d187ec9ce3d3039a0ff75daa0ab473fb3d8e5fa92db9c1ddca1` +- live: `8b7f267ee947932bdd52d1ef4b241adc8c9b4c4d5a8a3cd6c1e7f6e066983699` +- 3D/fusion E14: `a010e32070459b123423f3d4f57aa1d839f73ed05f3df8580d12693406a1b218` +- semantic: `ea583966bc3409f5cf563cbf4fad05e366907e67187082eb692aff53d9f5d875` +- inline temporal E23.2: `86238c369aa2ec0f5de50a2c676faf9b11884e1355f8cccf416125b5f73fd38e` + +## Processing topology + +The warm worker processed independent bounded branches: + +1. Camera decode and valid-FOV application. +2. Detector branch at the camera rate, with a latest-wins queue of capacity 2. +3. Semantic branch every fifth camera frame, with a latest-wins queue of capacity 1. +4. Bounded LiDAR and pose synchronizer, capacity 32 per modality and 3-second retention. +5. Camera/LiDAR projection and object association. +6. Raw 2D/3D fusion artifact emission for qualification only. +7. Inline temporal stabilization of 2D tracks and 3D cuboids. +8. Inline streaming semantic hysteresis with one previous output mask. +9. Stabilized fusion/world-state publication to the diagnostic result queue. + +Detector and semantic inference therefore ran concurrently. They were not executed as one long sequential offline job. The temporal stage sat directly in the publication path and its measured latency is included in the result. + +## Bounded-state contract + +| Bound | Configured | Observed | +|---|---:|---:| +| 2D/3D track states | 128 | 19 peak | +| Track idle lifetime | 1.25 s | bounded by eviction | +| Semantic history | 1 mask | 1 mask | +| Runtime diagnostic samples | 4096 | 61 used | +| Detector queue | 2 | depth 2 peak | +| Semantic queue | 1 | depth 1 peak | +| Synchronizer per modality | 32 | depth 32 peak | + +This is the key architectural difference from accumulating an entire mission in Rerun or in the AI worker. Runtime memory is bounded by live state, queue capacity, model state, and fixed diagnostic rings—not by mission duration. + +## Throughput and freshness + +| Metric | Result | +|---|---:| +| Detector effective rate | 9.8939 FPS | +| Detector camera frames published / consumed | 601 / 596 | +| Detector latest-wins replacements | 5 (0.83%) | +| Semantic effective rate | 2.0087 FPS | +| Semantic frames published / consumed | 121 / 121 | +| Semantic queue drops | 0 | +| Semantic fresh coverage | 98.8255% | +| Fused frames | 544 | +| Fused fraction | 91.2752% | +| Raw accepted cuboids | 867 | +| Stabilized/held cuboids | 1323 | + +The five detector replacements are represented as explicit empty frames in the visual projection: 12, 240, 433, 496, and 499. They are not silently copied from an older inference. + +## Latency + +| Stage | Mean | P95 | Max | +|---|---:|---:|---:| +| Detector | 36.330 ms | 55.757 ms | 171.674 ms | +| Semantic full processing | 205.385 ms | 247.682 ms | 481.552 ms | +| Semantic completion age | 216.966 ms | 265.524 ms | 535.985 ms | +| Projection | 0.756 ms | 1.187 ms | 17.499 ms | +| Association | 4.383 ms | 12.736 ms | 25.382 ms | +| Inline temporal 2D/3D | 0.911 ms | 1.721 ms | 11.348 ms | +| Inline temporal semantic | 3.616 ms | 12.307 ms | 31.396 ms | +| World-state construction | 1.063 ms | 2.022 ms | 12.583 ms | +| World-state age | 72.670 ms | 185.375 ms | 366.839 ms | + +Semantic inference is deliberately slower than the camera interval and is reused between sampled frames under a freshness TTL. It is not claimed to run at 10 FPS. + +## Resource use + +Measured worker: NVIDIA GeForce RTX 4090. + +| Metric | Result | +|---|---:| +| Mean process CPU | 141.75% (~1.42 logical cores) | +| Mean process RSS | 1547.11 MiB | +| RSS growth over run | 12.14 MiB | +| Maximum worker threads | 134 | +| CUDA peak allocated | 2107.93 MiB | +| CUDA peak reserved | 2840.00 MiB | +| Mean device GPU memory used | 13,544.25 MiB | +| Mean GPU utilization | 60.18% | +| GPU utilization P95 | 82% | +| Mean GPU power | 192.93 W | +| Maximum GPU temperature | 45 °C | +| D: free before | 407,605,469,184 bytes | +| D: free after | 407,577,206,784 bytes | +| Enforced D: free floor | 386,547,056,640 bytes (360 GiB) | + +The device-wide GPU-memory figure includes the persistent Triton/model-serving footprint and must not be interpreted as the incremental cost of the temporal filter. The temporal filter itself added bounded CPU/NumPy state and the process RSS grew by only 12.14 MiB during the run. + +No experiment payload was intentionally placed on C:. Worker packages, runner roots, and result artifacts remained on D:. + +## Quality result + +Raw and stabilized values were computed from artifacts produced during the same E23.2 worker run. The stabilized result was not recomputed offline for acceptance. + +| Metric | Raw P95 | Inline P95 | Reduction | +|---|---:|---:|---:| +| 2D normalized acceleration | 0.525774 | 0.395351 | 24.81% | +| 2D normalized size step | 0.267858 | 0.138933 | 48.13% | +| 3D center step | 0.893014 m | 0.591626 m | 33.75% | +| 3D half-size step | 0.060727 m | 0.016893 m | 72.18% | +| 3D yaw step | 18.0226° | 4.9804° | 72.37% | +| Unsupported semantic change | — | — | 78.05% reduction | + +Required minimums were 20% for 2D acceleration, 3D center, and 3D yaw, and 10% for unsupported semantic change. All passed. + +The result also reduced 2D tracks of at most two frames from 39 to 4 and reduced 3D cuboid gap count from 179 to 69. These are diagnostic continuity metrics, not object-truth accuracy metrics. + +## Publication and visual verification + +The accepted projection contains: + +- 601 synchronized visual frames; +- 121 exact stabilized semantic masks; +- 1323 accepted visual cuboids; +- the source point cloud and trajectory; +- five explicit detector-replacement frames; +- no copied raw sensor payloads. + +Semantic pixels were reconstructed from the immutable reference only after every one of the 121 inline mask SHA-256 values matched. The published session was then verified in the running Mission Core interface on port 8000: + +1. the saved-session card appeared as `LAB E23.2`; +2. preparation transitioned from the lower-right status pill to ready; +3. the camera and point-cloud timeline advanced; +4. 2D boxes, segmentation, and 3D cuboids could be enabled together; +5. all three layers changed with playback; +6. the source session and earlier LAB instances remained present. + +The UI cleanup delivered with this experiment removes the redundant `СЦЕНА 3D · RERUN` label and moves AI preparation state from the toolbar into the lower-right navigation/status stack. + +## Acceptance decision + +LAB E23.2 is accepted as evidence that bounded temporal stabilization can be part of the near-real-time perception publication path on the measured RTX 4090 worker. + +It remains diagnostic-only: + +- `commands_enabled = false` +- `navigation_or_safety_accepted = false` + +## Next gate + +The next experiment should stop refining the recorded view in isolation and move one boundary closer to the vehicle: + +1. use the same hashed worker package and E23.2 profile against physical K1 live ingress; +2. retain raw and stabilized diagnostic artifacts for a bounded comparison window; +3. run at least a 10-minute soak before any longer mission test; +4. measure live packet loss, camera/LiDAR clock skew, world-state age, queue replacements, RSS slope, GPU temperature, and disk growth; +5. keep commands and navigation authority disabled; +6. if the 10-minute gate passes, extend to a 60-minute soak without changing the configuration. + +Model-domain replacement and safety validation are separate workstreams. They should not be mixed into the first physical live-ingress qualification because that would make timing, synchronization, and model-quality failures impossible to distinguish. diff --git a/experiments/perception/e23_inline_temporal_profile.json b/experiments/perception/e23_inline_temporal_profile.json new file mode 100644 index 0000000..2a348f4 --- /dev/null +++ b/experiments/perception/e23_inline_temporal_profile.json @@ -0,0 +1,62 @@ +{ + "schema_version": "missioncore.e23-inline-temporal-profile/v1", + "profile_id": "lab-e23-warm-worker-inline-temporal-v1", + "mode": "inline-shadow-qualification", + "stage": "warm-worker-after-fusion-before-result-publication", + "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.32, + "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, + "maximum_rss_growth_mib": 64.0 + }, + "authority": { + "commands_enabled": false, + "navigation_or_safety_accepted": false + } +} diff --git a/experiments/perception/e23_inline_temporal_qualification.json b/experiments/perception/e23_inline_temporal_qualification.json new file mode 100644 index 0000000..530bc73 --- /dev/null +++ b/experiments/perception/e23_inline_temporal_qualification.json @@ -0,0 +1,161 @@ +{ + "schema_version": "missioncore.e23-inline-temporal-qualification/v1", + "experiment": { + "lab_id": "LAB E23.2", + "created_at_utc": "2026-07-24T05:40:18.949Z", + "state": "accepted", + "mode": "recorded-source-paced-near-live", + "speed": 1.0, + "look_ahead": false, + "authority": { + "commands_enabled": false, + "navigation_or_safety_accepted": false + } + }, + "source": { + "session_id": "20260720T065719Z_viewer_live", + "worker_session_id": "e23-ravnoves00-1784871558075385000", + "camera_source": "sensor.camera.right", + "camera_calibration_slot": "camera_1", + "camera_resolution": [ + 800, + 600 + ], + "duration_seconds": 60.0, + "selected_span_seconds": 59.994249001, + "wall_seconds": 60.2407113750005, + "camera_frames": 601, + "lidar_events": 585, + "pose_events": 600, + "camera_index_sha256": "e029815a60ad9fbfedb6169142c7449df2b119a51d1ce001f08806e04eb0be14", + "mqtt_raw_sha256": "70da0edad5cbf0e89b6e2355c0aac294f8c36a67249cd33b8918353fd88c83af", + "pacing_clock": "recorded-host-arrival" + }, + "identity": { + "pipeline": "warm-worker-inline-bounded-temporal-2d-3d-semantic/v1", + "worker_result_id": "e15-shadow-inference-32a95297664241f7102b31e064ab0c168bbd97d4f5cc152892fd378d58617ffb", + "published_result_id": "e10-integrated-perception-f3cefd9464c8b0ed19bb0c8949c1d890b55bf57353850ba155c4dc8d7d90c371", + "published_session_id": "lab-e23-2-86238c36", + "reference_result_id": "e10-integrated-perception-5cbba0b9839ae843c05607cec8448a2b43464eb81edeb11cf7c01cfec8464c75", + "profile_sha256": "86238c369aa2ec0f5de50a2c676faf9b11884e1355f8cccf416125b5f73fd38e", + "worker_package_id": "e15-worker-package-dbf55ccb75664b778a2e0f5d34284af10ec0971945bbbee67b27bc264b765b51", + "worker_package_manifest_sha256": "d3089e241286b871ff4e9db569df10222bf2feb07500e10eb641ecce88ace868", + "projection_pack_id": "e15-live-projection-713fa7344805b7a592d12e949a6db38116e5370f768bde1cf9144fab5f93f0ae", + "valid_fov_mask_sha256": "a40cee06b7c6f69b6a09a11563dcfd237f3de833b1ccd31459e66692e528ba63", + "calibration_sha256": "05f3ad9b38b3a4fc95388a8ec83da83c745e217709e51787b3d5aad0969f6fa9" + }, + "runtime": { + "gpu": "NVIDIA GeForce RTX 4090", + "python": "3.12.3", + "torch": "2.13.0+cu130", + "transformers": "4.57.6", + "numpy": "1.26.4", + "scipy": "1.16.3", + "triton_image": "nvcr.io/nvidia/tritonserver:26.06-py3@sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794", + "detector": { + "model": "YOLOX-S", + "classes": "COCO-80", + "effective_fps": 9.893899244013618, + "frames_published": 601, + "frames_consumed": 596, + "latest_wins_replacements": 5 + }, + "semantic": { + "model": "tue-mps/cityscapes_semantic_eomt_large_1024", + "precision": "fp16-autocast", + "effective_fps": 2.0086607525598117, + "frames_published": 121, + "frames_consumed": 121, + "latest_wins_replacements": 0, + "fresh_coverage_fraction": 0.988255033557047 + }, + "fusion": { + "fused_frames": 544, + "fused_fraction": 0.912751677852349, + "raw_accepted_cuboids": 867, + "stabilized_cuboids": 1323 + }, + "latency_ms": { + "detector_p95": 55.757249, + "semantic_completion_age_p95": 265.523974, + "world_state_age_p95": 185.374918, + "inline_temporal_2d_3d_p95": 1.720854, + "inline_temporal_semantic_p95": 12.306933 + }, + "resource": { + "process_cpu_percent_mean": 141.749058, + "process_rss_mib_mean": 1547.110848, + "process_rss_growth_mib": 12.144531, + "process_threads_max": 134, + "cuda_peak_allocated_mib": 2107.92529296875, + "cuda_peak_reserved_mib": 2840.0, + "device_gpu_memory_used_mib_mean": 13544.245902, + "gpu_utilization_percent_mean": 60.180328, + "gpu_utilization_percent_p95": 82.0, + "gpu_power_w_mean": 192.926721, + "gpu_temperature_c_max": 45.0, + "disk_free_bytes_before": 407605469184, + "disk_free_bytes_after": 407577206784, + "disk_free_floor_bytes": 386547056640 + } + }, + "bounded_state": { + "maximum_track_states": 128, + "peak_track_states": 19, + "semantic_history_masks": 1, + "diagnostic_sample_capacity": 4096, + "maximum_track_idle_seconds": 1.25 + }, + "quality": { + "tracking_2d_acceleration_p95": { + "raw": 0.5257735155179932, + "stabilized": 0.395351460036327, + "reduction_fraction": 0.2480574841301661 + }, + "tracking_2d_size_step_p95": { + "raw": 0.26785786216659935, + "stabilized": 0.13893264173009742, + "reduction_fraction": 0.4813195304168985 + }, + "cuboid_center_step_p95_m": { + "raw": 0.8930137938715039, + "stabilized": 0.5916256441083151, + "reduction_fraction": 0.3374955144383309 + }, + "cuboid_half_size_step_p95_m": { + "raw": 0.06072732822316884, + "stabilized": 0.016892938813881653, + "reduction_fraction": 0.7218231180564829 + }, + "cuboid_yaw_step_p95_degrees": { + "raw": 18.02256860299971, + "stabilized": 4.980429289278814, + "reduction_fraction": 0.7236559671938293 + }, + "semantic_unsupported_change_reduction_fraction": 0.7804878048780487 + }, + "publication": { + "visual_frames": 601, + "semantic_masks": 121, + "detector_replacement_frames": [ + 12, + 240, + 433, + 496, + 499 + ], + "accepted_visual_cuboids": 1323, + "source_payloads_mutated": false, + "semantic_reconstruction": "exact-inline-sha256-match", + "browser_verified": true + }, + "predecessor": { + "lab_id": "LAB E23.1", + "worker_result_id": "e15-shadow-inference-37381767fb06928951aa173c3e2b43a81b8e492916e4ac8622d7d78d7a475f8e", + "profile_sha256": "a22de018626e11a2c4d18d87d6b682846921f72c6cd9f290f20db9a71218e1e3", + "runtime_gate": "accepted", + "final_quality_gate": "rejected", + "reason": "cuboid center-step P95 reduction was 0.1710, below the required 0.20", + "published": false + } +} diff --git a/experiments/perception/prepare_e15_worker_package.py b/experiments/perception/prepare_e15_worker_package.py index 57b4e23..cf44696 100644 --- a/experiments/perception/prepare_e15_worker_package.py +++ b/experiments/perception/prepare_e15_worker_package.py @@ -16,6 +16,7 @@ from typing import Any SCHEMA = "missioncore.e15-worker-package/v1" COPIED_FILES = ( "k1link/__init__.py", + "k1link/compute/inline_temporal.py", "k1link/compute/live_perception.py", "k1link/data_plane/__init__.py", "k1link/data_plane/views.py", @@ -68,6 +69,8 @@ def _identity(source_root: Path) -> dict[str, Any]: for path, content in sorted(GENERATED_FILES.items()) }, "imports": [ + "k1link.compute.inline_temporal.TemporalStabilizer", + "k1link.compute.inline_temporal.StreamingSemanticStabilizer", "k1link.compute.live_perception.LiveSensorSynchronizer", "k1link.data_plane.DecodedPointCloudView", "k1link.data_plane.DecodedPoseView", @@ -85,9 +88,7 @@ def validate_worker_package(root: Path, *, allow_staging: bool = False) -> dict[ artifacts = manifest.get("artifacts") expected_paths = {*COPIED_FILES, *GENERATED_FILES, "manifest.json"} actual_paths = { - path.relative_to(resolved).as_posix() - for path in resolved.rglob("*") - if path.is_file() + path.relative_to(resolved).as_posix() for path in resolved.rglob("*") if path.is_file() } if ( manifest.get("schema_version") != SCHEMA diff --git a/experiments/perception/run_e12_shadow_transport_replay.py b/experiments/perception/run_e12_shadow_transport_replay.py index c4eb2c3..d89e378 100644 --- a/experiments/perception/run_e12_shadow_transport_replay.py +++ b/experiments/perception/run_e12_shadow_transport_replay.py @@ -169,7 +169,11 @@ def _wait_for(predicate: object, *, timeout_seconds: float, label: str) -> None: def run(args: argparse.Namespace) -> dict[str, object]: experiment_id = str(getattr(args, "experiment_id", _EXPERIMENT_ID)) speed = float(getattr(args, "speed", 1.0)) - if experiment_id not in {"e12", "e21"} or not math.isfinite(speed) or not 0.1 <= speed <= 10: + if ( + experiment_id not in {"e12", "e21", "e23"} + or not math.isfinite(speed) + or not 0.1 <= speed <= 10 + ): raise RuntimeError("replay experiment identity or speed is invalid") repository_root = Path(args.repository_root).expanduser().resolve() session_root = Path(args.session_root).expanduser().resolve() @@ -292,11 +296,11 @@ def run(args: argparse.Namespace) -> dict[str, object]: raise RuntimeError("selected replay interval is empty") source_span_seconds = max(0.0, (source_last_ns - source_start_ns) / 1_000_000_000) wall_seconds = completed_monotonic - started_monotonic - report_schema = ( - REPORT_SCHEMA - if experiment_id == _EXPERIMENT_ID - else "missioncore.e21-replay-source-report/v1" - ) + report_schema = { + "e12": REPORT_SCHEMA, + "e21": "missioncore.e21-replay-source-report/v1", + "e23": "missioncore.e23-replay-source-report/v1", + }[experiment_id] peak_rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss peak_rss_mib = peak_rss / (1024 * 1024 if sys.platform == "darwin" else 1024) report: dict[str, object] = { @@ -375,7 +379,7 @@ def _arguments() -> argparse.Namespace: parser.add_argument("--port", type=int, default=8012) parser.add_argument("--duration-seconds", type=float, default=15.0) parser.add_argument("--speed", type=float, default=1.0) - parser.add_argument("--experiment-id", choices=("e12", "e21"), default="e12") + parser.add_argument("--experiment-id", choices=("e12", "e21", "e23"), default="e12") parser.add_argument("--consumer-timeout-seconds", type=float, default=30.0) parser.add_argument("--completion-timeout-seconds", type=float, default=30.0) return parser.parse_args() diff --git a/experiments/perception/worker/Invoke-E15ShadowInference.ps1 b/experiments/perception/worker/Invoke-E15ShadowInference.ps1 index f180b62..722dd6b 100644 --- a/experiments/perception/worker/Invoke-E15ShadowInference.ps1 +++ b/experiments/perception/worker/Invoke-E15ShadowInference.ps1 @@ -9,6 +9,7 @@ param( [Parameter(Mandatory = $true)] [string]$ValidFovRoot, [Parameter(Mandatory = $true)] [string]$ProjectionPackRoot, [Parameter(Mandatory = $true)] [string]$PackageRoot, + [string]$StabilityProfilePath = "", [switch]$PreflightOnly, [switch]$PersistentService, [switch]$TokenStdin, @@ -84,6 +85,10 @@ $liveProfile = Resolve-DFile $LiveProfilePath "LAB E15 live profile" $e14Profile = Resolve-DFile $E14ProfilePath "Accepted E14 profile" $detectorProfile = Resolve-DFile $DetectorProfilePath "Detector profile" $semanticProfile = Resolve-DFile $SemanticProfilePath "Semantic profile" +$stabilityProfile = $null +if ($StabilityProfilePath) { + $stabilityProfile = Resolve-DFile $StabilityProfilePath "LAB E23 stability profile" +} $validFov = Resolve-DDirectory $ValidFovRoot "Valid-FOV root" $projectionPack = Resolve-DDirectory $ProjectionPackRoot "E15 projection pack" $package = Resolve-DDirectory $PackageRoot "Mission Core package root" @@ -97,6 +102,9 @@ foreach ($path in @($liveProfile, $e14Profile, $detectorProfile, $semanticProfil throw "Runner and profiles must share one immutable mount" } } +if ($stabilityProfile -and (Split-Path $stabilityProfile -Parent) -ne $runnerRoot) { + throw "Runner and LAB E23 stability profile must share one immutable mount" +} foreach ($dependency in @( "e10_fusion_runtime.py", "e15_shadow_runtime.py", @@ -115,6 +123,9 @@ foreach ($dependency in @( if (-not (Test-Path -LiteralPath (Join-Path $package "k1link\compute\live_perception.py") -PathType Leaf)) { throw "Mission Core package mount lacks live perception synchronization" } +if (-not (Test-Path -LiteralPath (Join-Path $package "k1link\compute\inline_temporal.py") -PathType Leaf)) { + throw "Mission Core package mount lacks inline temporal state" +} $job = Get-Content -LiteralPath (Join-Path $jobDirectory "job.json") -Raw | ConvertFrom-Json if ( @@ -130,6 +141,17 @@ if ( [bool]$live.authority.navigation_or_safety_accepted -or $live.transport.pyav_version -ne "18.0.0" ) { throw "LAB E15 replay-shadow authority contract changed" } +if ($stabilityProfile) { + $stability = Get-Content -LiteralPath $stabilityProfile -Raw | ConvertFrom-Json + if ( + $stability.schema_version -ne "missioncore.e23-inline-temporal-profile/v1" -or + $stability.mode -ne "inline-shadow-qualification" -or + $stability.stage -ne "warm-worker-after-fusion-before-result-publication" -or + $stability.source.calibration_sha256 -ne $live.source.calibration_sha256 -or + [bool]$stability.authority.commands_enabled -or + [bool]$stability.authority.navigation_or_safety_accepted + ) { throw "LAB E23 inline temporal authority contract changed" } +} $projection = Get-Content -LiteralPath (Join-Path $projectionPack "manifest.json") -Raw | ConvertFrom-Json if ( $projection.schema_version -ne "missioncore.e15-live-projection-pack/v1" -or @@ -137,6 +159,12 @@ if ( $projection.identity.calibration_slot -ne "camera_1" -or $projection.identity.calibration_sha256 -ne $live.source.calibration_sha256 ) { throw "LAB E15 projection pack binding changed" } +$packageManifest = Get-Content -LiteralPath (Join-Path $package "manifest.json") -Raw | ConvertFrom-Json +if ( + $packageManifest.schema_version -ne "missioncore.e15-worker-package/v1" -or + $packageManifest.package_id -ne (Split-Path $package -Leaf) -or + $packageManifest.identity_sha256 -notmatch "^[a-f0-9]{64}$" +) { throw "LAB E15 worker package manifest changed" } $mediaManifest = Get-Content -LiteralPath (Join-Path $mediaRuntime "manifest.json") -Raw | ConvertFrom-Json if ( $mediaManifest.schema_version -ne "missioncore.e15-media-runtime/v1" -or @@ -165,6 +193,7 @@ $liveProfileName = Split-Path $liveProfile -Leaf $e14ProfileName = Split-Path $e14Profile -Leaf $detectorProfileName = Split-Path $detectorProfile -Leaf $semanticProfileName = Split-Path $semanticProfile -Leaf +$stabilityProfileName = if ($stabilityProfile) { Split-Path $stabilityProfile -Leaf } else { $null } $projectionMount = "/" + (Split-Path $projectionPack -Leaf) $packageMount = "/" + (Split-Path $package -Leaf) $orchestratorSha256 = (Get-FileHash -LiteralPath $PSCommandPath -Algorithm SHA256).Hash.ToLowerInvariant() @@ -200,6 +229,11 @@ $commonRunnerArgs = @( "--environment", "/environment", "--worker-package", $packageMount ) +if ($stabilityProfileName) { + $commonRunnerArgs += @( + "--stability-profile", ("/runner/{0}" -f $stabilityProfileName) + ) +} if ($PersistentService) { if ($PreflightOnly -or $TokenStdin) { @@ -214,6 +248,14 @@ if ($PersistentService) { throw "Persistent output root must be a direct child of its guarded D: parent" } $runnerSha256 = (Get-FileHash -LiteralPath $runner -Algorithm SHA256).Hash.ToLowerInvariant() + $liveProfileSha256 = (Get-FileHash -LiteralPath $liveProfile -Algorithm SHA256).Hash.ToLowerInvariant() + $e14ProfileSha256 = (Get-FileHash -LiteralPath $e14Profile -Algorithm SHA256).Hash.ToLowerInvariant() + $detectorProfileSha256 = (Get-FileHash -LiteralPath $detectorProfile -Algorithm SHA256).Hash.ToLowerInvariant() + $semanticProfileSha256 = (Get-FileHash -LiteralPath $semanticProfile -Algorithm SHA256).Hash.ToLowerInvariant() + $stabilityProfileSha256 = if ($stabilityProfile) { + (Get-FileHash -LiteralPath $stabilityProfile -Algorithm SHA256).Hash.ToLowerInvariant() + } + else { "none" } $existingContainerId = docker ps -a --filter ("name=^{0}$" -f $PersistentContainer) --format "{{.ID}}" Assert-LastExitCode "Persistent worker container lookup" if ($existingContainerId) { @@ -221,7 +263,14 @@ if ($PersistentService) { Assert-LastExitCode "Persistent worker label inspection" if ( $labels.'missioncore.role' -ne "perception-persistent-worker" -or - $labels.'missioncore.runner.sha256' -ne $runnerSha256 + $labels.'missioncore.runner.sha256' -ne $runnerSha256 -or + $labels.'missioncore.live.sha256' -ne $liveProfileSha256 -or + $labels.'missioncore.e14.sha256' -ne $e14ProfileSha256 -or + $labels.'missioncore.detector.sha256' -ne $detectorProfileSha256 -or + $labels.'missioncore.semantic.sha256' -ne $semanticProfileSha256 -or + $labels.'missioncore.stability.sha256' -ne $stabilityProfileSha256 -or + $labels.'missioncore.worker-package.identity' -ne $packageManifest.identity_sha256 -or + $labels.'missioncore.projection.identity' -ne $projection.identity_sha256 ) { throw "Existing persistent worker has a different immutable identity" } $running = docker inspect --format "{{.State.Running}}" $PersistentContainer Assert-LastExitCode "Persistent worker state inspection" @@ -258,6 +307,13 @@ if ($PersistentService) { "run", "--detach", "--name", $PersistentContainer, "--label", "missioncore.role=perception-persistent-worker", "--label", ("missioncore.runner.sha256={0}" -f $runnerSha256), + "--label", ("missioncore.live.sha256={0}" -f $liveProfileSha256), + "--label", ("missioncore.e14.sha256={0}" -f $e14ProfileSha256), + "--label", ("missioncore.detector.sha256={0}" -f $detectorProfileSha256), + "--label", ("missioncore.semantic.sha256={0}" -f $semanticProfileSha256), + "--label", ("missioncore.stability.sha256={0}" -f $stabilityProfileSha256), + "--label", ("missioncore.worker-package.identity={0}" -f $packageManifest.identity_sha256), + "--label", ("missioncore.projection.identity={0}" -f $projection.identity_sha256), "--gpus", "all", "--network", ("container:{0}" -f $TritonContainer), "--read-only", "--security-opt", "no-new-privileges:true", "--cap-drop", "ALL", "--pids-limit", "512", "--shm-size", "4g", diff --git a/experiments/perception/worker/run_e10_integrated_perception.py b/experiments/perception/worker/run_e10_integrated_perception.py index 7a5b177..14012b9 100644 --- a/experiments/perception/worker/run_e10_integrated_perception.py +++ b/experiments/perception/worker/run_e10_integrated_perception.py @@ -212,6 +212,7 @@ def semantic_worker( completed: list[SemanticResult], failures: list[BaseException], stop_after_results: int | None, + transform_target: Any | None = None, ) -> None: try: while (envelope := queue.take()) is not None: @@ -225,6 +226,14 @@ def semantic_worker( raise RuntimeError("LAB E10 EoMT emitted an unknown category") target = target_lut[semantic].copy() target[~valid_mask] = 0 + if transform_target is not None: + target = transform_target(target) + if ( + not isinstance(target, np.ndarray) + or target.dtype != np.uint8 + or target.shape != valid_mask.shape + ): + raise RuntimeError("semantic target transform returned an invalid mask") for name, value in measured.items(): latency[name].append(float(value)) finished = time.perf_counter() diff --git a/experiments/perception/worker/run_e15_shadow_inference.py b/experiments/perception/worker/run_e15_shadow_inference.py index beb844a..cf91be8 100644 --- a/experiments/perception/worker/run_e15_shadow_inference.py +++ b/experiments/perception/worker/run_e15_shadow_inference.py @@ -19,7 +19,7 @@ import struct import sys import threading import time -from collections import Counter +from collections import Counter, deque from contextlib import suppress from dataclasses import dataclass from datetime import UTC, datetime @@ -83,6 +83,15 @@ from run_recorded_perception_epoch import ( _write_json, ) +from k1link.compute.inline_temporal import ( + PIPELINE_ID as INLINE_TEMPORAL_PIPELINE_ID, +) +from k1link.compute.inline_temporal import ( + StreamingSemanticStabilizer, + TemporalStabilizer, + read_inline_profile, + stabilize_world_state, +) from k1link.compute.live_perception import ( LIVE_RESULT_MAX_PAYLOAD_BYTES, LiveSensorSynchronizer, @@ -119,6 +128,7 @@ def arguments() -> argparse.Namespace: command.add_argument("--cache", type=Path, required=True) command.add_argument("--environment", type=Path, required=True) command.add_argument("--worker-package", type=Path, required=True) + command.add_argument("--stability-profile", type=Path) if name in {"run", "serve"}: command.add_argument("--host", default="host.docker.internal") command.add_argument("--port", type=int, default=18012) @@ -172,8 +182,7 @@ def read_live_profile(path: Path) -> tuple[dict[str, Any], str]: or source.get("source_id") != "sensor.camera.right" or source.get("resolution") != [800, 600] or source.get("calibration_slot") != "camera_1" - or authority - != {"commands_enabled": False, "navigation_or_safety_accepted": False} + or authority != {"commands_enabled": False, "navigation_or_safety_accepted": False} or transport.get("wire_schema") != "missioncore.live-perception-wire/v1" or transport.get("camera_media") != "persistent-fmp4-pyav" ): @@ -190,9 +199,7 @@ def read_live_profile(path: Path) -> tuple[dict[str, Any], str]: if ( not 100 <= float(scheduling.get("semantic_ttl_ms", 0)) <= 5000 or not 0 <= float(scheduling.get("sensor_wait_ms", -1)) <= 100 - or not 1_048_576 - <= int(transport.get("maximum_media_buffer_bytes", 0)) - <= 64 * 1024 * 1024 + or not 1_048_576 <= int(transport.get("maximum_media_buffer_bytes", 0)) <= 64 * 1024 * 1024 or not 2 <= int(transport.get("camera_metadata_capacity", 0)) <= 128 or not 1 <= int(temporal.get("buffer_capacity_per_modality", 0)) <= 256 or not 0.1 <= float(temporal.get("retention_seconds", 0)) <= 30 @@ -294,14 +301,13 @@ def read_worker_package(root: Path) -> dict[str, Any]: manifest.get("schema_version") != WORKER_PACKAGE_SCHEMA or not isinstance(identity, dict) or identity.get("schema_version") != WORKER_PACKAGE_SCHEMA - or identity.get("classification") - != "minimal-live-worker-import-projection" + or identity.get("classification") != "minimal-live-worker-import-projection" or not isinstance(identity_sha256, str) or hashlib.sha256(canonical_json(identity)).hexdigest() != identity_sha256 or manifest.get("package_id") != f"e15-worker-package-{identity_sha256}" or resolved.name != manifest.get("package_id") or not isinstance(artifacts, list) - or len(artifacts) != 11 + or len(artifacts) != 12 ): raise RuntimeError("LAB E15 worker package identity is invalid") expected = set() @@ -321,9 +327,7 @@ def read_worker_package(root: Path) -> dict[str, Any]: raise RuntimeError("LAB E15 worker package artifact changed") expected.add(relative) actual = { - path.relative_to(resolved).as_posix() - for path in resolved.rglob("*") - if path.is_file() + path.relative_to(resolved).as_posix() for path in resolved.rglob("*") if path.is_file() } if actual != expected | {"manifest.json"}: raise RuntimeError("LAB E15 worker package file set changed") @@ -386,7 +390,7 @@ class _RuntimeTelemetry: self._started = time.perf_counter() self._previous_wall = self._started self._previous_cpu = time.process_time() - self.samples: list[dict[str, Any]] = [] + self.samples: deque[dict[str, Any]] = deque(maxlen=4096) def __enter__(self) -> _RuntimeTelemetry: self._thread.start() @@ -498,6 +502,7 @@ class _RuntimeTelemetry: self._sample() def summary(self) -> dict[str, Any]: + samples = list(self.samples) numeric_fields = ( "process_cpu_percent", "process_rss_mib", @@ -506,25 +511,26 @@ class _RuntimeTelemetry: "cgroup_memory_current_mib", ) summary: dict[str, Any] = { - "sample_count": len(self.samples), + "sample_count": len(samples), + "sample_capacity": self.samples.maxlen, "interval_seconds": self._interval, } for field in numeric_fields: values = [ float(sample[field]) - for sample in self.samples + for sample in samples if isinstance(sample.get(field), int | float) ] summary[field] = _percentiles(values) - if self.samples: - quarter = max(1, len(self.samples) // 4) - early = [float(value["process_rss_mib"]) for value in self.samples[:quarter]] - late = [float(value["process_rss_mib"]) for value in self.samples[-quarter:]] + if samples: + quarter = max(1, len(samples) // 4) + early = [float(value["process_rss_mib"]) for value in samples[:quarter]] + late = [float(value["process_rss_mib"]) for value in samples[-quarter:]] summary["rss_growth_mib"] = round( float(_percentiles(late)["p95"]) - float(_percentiles(early)["p95"]), 6, ) - summary["final_queues"] = self.samples[-1]["queues"] + summary["final_queues"] = samples[-1]["queues"] else: summary["rss_growth_mib"] = 0.0 summary["final_queues"] = {} @@ -627,9 +633,7 @@ def _receiver( if state.last_ingress_sequence is not None: if sequence <= state.last_ingress_sequence: raise ShadowRuntimeError("shadow ingress sequence is not increasing") - state.ingress_sequence_gaps += max( - 0, sequence - state.last_ingress_sequence - 1 - ) + state.ingress_sequence_gaps += max(0, sequence - state.last_ingress_sequence - 1) if state.first_ingress_sequence is None: state.first_ingress_sequence = sequence state.last_ingress_sequence = sequence @@ -661,9 +665,7 @@ def _receiver( f"expected at least {expected_camera_sequence}, " f"got {camera_source_sequence}" ) - state.camera_sequence_gaps += ( - camera_source_sequence - expected_camera_sequence - ) + state.camera_sequence_gaps += camera_source_sequence - expected_camera_sequence state.last_camera_source_sequence = camera_source_sequence decoder.feed_segment( CameraFragmentMetadata( @@ -686,9 +688,7 @@ def _receiver( ), processing_started_monotonic_ns=time.monotonic_ns(), ) - sensor_decode_ms[modality].append( - (time.perf_counter() - decode_started) * 1000 - ) + sensor_decode_ms[modality].append((time.perf_counter() - decode_started) * 1000) if modality == "lidar" and isinstance(normalized, DecodedPointCloudView): synchronizer.publish_point_cloud(normalized) elif modality == "pose" and isinstance(normalized, DecodedPoseView): @@ -707,9 +707,7 @@ def _receiver( sender_thread.join(timeout=5) if sender_thread.is_alive(): assert state.failures is not None - state.failures.append( - ShadowRuntimeError("live result publisher did not stop") - ) + state.failures.append(ShadowRuntimeError("live result publisher did not stop")) if stream is not None: with suppress(Exception), write_lock: _send_client_frame(stream, 0x8, b"") @@ -760,6 +758,12 @@ def _common(args: argparse.Namespace) -> dict[str, Any]: if dependency["identity"]["profile_sha256"] != semantic_sha256: raise RuntimeError("LAB E15 semantic dependency identity changed") worker_package = read_worker_package(args.worker_package) + stability = None + stability_sha256 = None + if args.stability_profile is not None: + stability, stability_sha256 = read_inline_profile(args.stability_profile) + if stability["source"] != live["source"] or stability["authority"] != live["authority"]: + raise RuntimeError("LAB E23 inline temporal source binding changed") return { "job": job, "live": live, @@ -777,6 +781,8 @@ def _common(args: argparse.Namespace) -> dict[str, Any]: "detector_files": detector_files, "dependency": dependency, "worker_package": worker_package, + "stability": stability, + "stability_sha256": stability_sha256, } @@ -804,6 +810,7 @@ def preflight(args: argparse.Namespace) -> int: "e14_profile_sha256": common["e14_sha256"], "projection_pack": common["projection_manifest"]["pack_id"], "worker_package": common["worker_package"]["package_id"], + "stability_profile_sha256": common["stability_sha256"], "pyav": av.__version__, "lz4": getattr(lz4, "__version__", importlib.metadata.version("lz4")), "detector_files": common["detector_files"], @@ -840,8 +847,7 @@ def _load_models(args: argparse.Namespace, common: dict[str, Any]) -> _LoadedMod ) infer_semantic = e9._semantic_infer_factory(processor, semantic_model, device) target_names = { - int(key): str(value) - for key, value in common["semantic"]["target_taxonomy"].items() + int(key): str(value) for key, value in common["semantic"]["target_taxonomy"].items() } warm = np.zeros((600, 800, 3), dtype=np.uint8) _infer( @@ -894,6 +900,10 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: infer_semantic = loaded.infer_semantic semantic_files = loaded.semantic_files target_lut = loaded.target_lut + stability = common["stability"] + temporal_stabilizer = TemporalStabilizer(stability) if stability is not None else None + semantic_stabilizer = StreamingSemanticStabilizer(stability) if stability is not None else None + temporal_world_memory: dict[int, dict[str, Any]] = {} tracker = TwoStageTracker(detector["tracking"]) linear_sum_assignment(np.zeros((1, 1), dtype=np.float64)) torch.cuda.reset_peak_memory_stats() @@ -988,16 +998,19 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: "source_epoch_age_ms_unqualified", ) } + latency["temporal_2d_3d_ms"] = deque(maxlen=4096) status_counts: Counter[str] = Counter() fusion_state_counts: Counter[str] = Counter() rejection_counts: Counter[str] = Counter() fused_frames = 0 accepted_cuboids = 0 + stabilized_cuboids = 0 detector_failures = 0 history = distance_history(int(e14["association"]["distance_history_frames"])) projector = WorldStateProjector(float(e14["world_state"]["velocity_history_limit_s"])) completion_tracker = CuboidCompletionTracker(e14["cuboid_completion"]) semantic_path = output / "semantic-frames.jsonl" + raw_fusion_path = output / "raw-fusion-frames.jsonl" fusion_path = output / "fusion-frames.jsonl" world_path = output / "world-state.jsonl" gpu_path = output / "gpu-telemetry.jsonl" @@ -1007,6 +1020,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: with ( semantic_path.open("x", encoding="utf-8", newline="\n") as semantic_stream, + raw_fusion_path.open("x", encoding="utf-8", newline="\n") as raw_fusion_stream, fusion_path.open("x", encoding="utf-8", newline="\n") as fusion_stream, world_path.open("x", encoding="utf-8", newline="\n") as world_stream, gpu_path.open("x", encoding="utf-8", newline="\n") as gpu_stream, @@ -1029,6 +1043,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: }, ) as runtime_telemetry, ): + class SemanticResultStream: def __init__(self) -> None: self.count = 0 @@ -1067,6 +1082,9 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: "completed": completed_semantics, "failures": semantic_errors, "stop_after_results": None, + "transform_target": ( + None if semantic_stabilizer is None else semantic_stabilizer.update + ), }, name="lab-e15-semantic", daemon=True, @@ -1115,18 +1133,14 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: while (envelope := detector_queue.take()) is not None: started = time.perf_counter() latency["decode_age_ms"].append(float(envelope.decode_ms)) - latency["queue_wait_ms"].append( - max(0.0, (started - envelope.decoded_monotonic) * 1000) - ) + latency["queue_wait_ms"].append(max(0.0, (started - envelope.decoded_monotonic) * 1000)) try: detector_started = time.perf_counter() tensor = _preprocess(envelope.image, valid_mask, detector) output_tensor, _request_ms = _infer(args.triton_url, detector["model"], tensor) detections, _rejected = _detections(output_tensor, detector, valid_mask) tracks = tracker.update(detections, envelope.frame_index) - latency["detector_ms"].append( - (time.perf_counter() - detector_started) * 1000 - ) + latency["detector_ms"].append((time.perf_counter() - detector_started) * 1000) frame_seconds = float(envelope.timeline["session_seconds"]) current_semantic = latest.snapshot() @@ -1141,9 +1155,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: int(envelope.timeline["captured_at_epoch_ns"]), wait_seconds=float(scheduling["sensor_wait_ms"]) / 1000, ) - latency["sensor_wait_ms"].append( - (time.perf_counter() - sensor_wait_started) * 1000 - ) + latency["sensor_wait_ms"].append((time.perf_counter() - sensor_wait_started) * 1000) fusions = () points_lidar = np.empty((0, 3), dtype=np.float64) if binding.state != "fused-ready": @@ -1194,18 +1206,12 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: rejection_counts[item.status] += 1 clearance_started = time.perf_counter() clearance_state = clearance(points_lidar, e14["world_state"]["clearance"]) - latency["clearance_ms"].append( - (time.perf_counter() - clearance_started) * 1000 - ) + latency["clearance_ms"].append((time.perf_counter() - clearance_started) * 1000) world_started = time.perf_counter() - result_age = max( - 0.0, (world_started - envelope.scheduled_monotonic) * 1000 - ) + result_age = max(0.0, (world_started - envelope.scheduled_monotonic) * 1000) if result_age >= 1000: health = "unavailable" - elif result_age >= float( - live["acceptance"]["maximum_p95_world_state_age_ms"] - ): + elif result_age >= float(live["acceptance"]["maximum_p95_world_state_age_ms"]): health = "stale" elif fusion_state != "fused": health = "degraded" @@ -1229,37 +1235,58 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: "clock_qualification": "worker-ingress-only", }, ) - latency["world_state_ms"].append( - (time.perf_counter() - world_started) * 1000 - ) - result_age = max( - 0.0, (time.perf_counter() - envelope.scheduled_monotonic) * 1000 + raw_fusion_objects = [fusion_document(item) for item in fusions] + if temporal_stabilizer is None: + fusion_objects = raw_fusion_objects + else: + temporal_started = time.perf_counter() + fusion_objects = temporal_stabilizer.update( + frame_index=envelope.frame_index, + session_seconds=frame_seconds, + objects=raw_fusion_objects, + ) + world = stabilize_world_state( + world, + fusion_objects, + temporal_world_memory, + ) + latency["temporal_2d_3d_ms"].append( + (time.perf_counter() - temporal_started) * 1000 + ) + stabilized_cuboids += sum( + str(item.get("cuboid_status", "")).startswith("accepted-") + for item in fusion_objects ) + latency["world_state_ms"].append((time.perf_counter() - world_started) * 1000) + result_age = max(0.0, (time.perf_counter() - envelope.scheduled_monotonic) * 1000) world["delivery"]["result_age_ms"] = result_age latency["world_state_age_ms"].append(result_age) latency["source_epoch_age_ms_unqualified"].append( - (time.time_ns() - int(envelope.timeline["captured_at_epoch_ns"])) - / 1_000_000 + (time.time_ns() - int(envelope.timeline["captured_at_epoch_ns"])) / 1_000_000 + ) + fusion_row = { + "schema_version": FUSION_SCHEMA, + "frame_index": envelope.frame_index, + "source_frame_index": int(envelope.timeline["source_frame_index"]), + "session_seconds": frame_seconds, + "fusion_state": fusion_state, + "semantic_status": semantic_status, + "semantic_source_frame_index": ( + None if current_semantic is None else current_semantic.source_frame_index + ), + } + raw_fusion_stream.write( + json.dumps( + {**fusion_row, "objects": raw_fusion_objects}, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + "\n" ) - fusion_objects = [fusion_document(item) for item in fusions] fusion_stream.write( json.dumps( - { - "schema_version": FUSION_SCHEMA, - "frame_index": envelope.frame_index, - "source_frame_index": int( - envelope.timeline["source_frame_index"] - ), - "session_seconds": frame_seconds, - "fusion_state": fusion_state, - "semantic_status": semantic_status, - "semantic_source_frame_index": ( - None - if current_semantic is None - else current_semantic.source_frame_index - ), - "objects": fusion_objects, - }, + {**fusion_row, "objects": fusion_objects}, sort_keys=True, separators=(",", ":"), allow_nan=False, @@ -1315,18 +1342,9 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: receiver_thread.join(timeout=args.max_duration_seconds + 10) decoder_watcher.join(timeout=30) semantic_thread.join(timeout=30) - if ( - receiver_thread.is_alive() - or decoder_watcher.is_alive() - or semantic_thread.is_alive() - ): + if receiver_thread.is_alive() or decoder_watcher.is_alive() or semantic_thread.is_alive(): raise RuntimeError("LAB E15 runtime thread did not stop") - if ( - transport.failures - or decoder_watch_failures - or callback_failures - or semantic_errors - ): + if transport.failures or decoder_watch_failures or callback_failures or semantic_errors: failures = [ *(transport.failures or []), *decoder_watch_failures, @@ -1334,11 +1352,15 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: *semantic_errors, ] summary = "; ".join( - f"{type(failure).__name__}: {str(failure)[:240]}" - for failure in failures[:8] + f"{type(failure).__name__}: {str(failure)[:240]}" for failure in failures[:8] ) raise RuntimeError(f"LAB E15 runtime worker failed: {summary}") - for stream in (semantic_stream, fusion_stream, world_stream): + for stream in ( + semantic_stream, + raw_fusion_stream, + fusion_stream, + world_stream, + ): stream.flush() os.fsync(stream.fileno()) @@ -1354,22 +1376,23 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: detector_fps = int(detector_state["consumed"]) / max(source_span, run_wall, 1e-9) semantic_fps = int(semantic_state["consumed"]) / max(source_span, run_wall, 1e-9) semantic_scheduled = ( - (decoded_frame_count - 1) - // int(scheduling["semantic_sample_every_frames"]) + (decoded_frame_count - 1) // int(scheduling["semantic_sample_every_frames"]) ) + 1 fresh_coverage = status_counts["fresh"] / max(1, int(detector_state["consumed"])) fused_fraction = fused_frames / max(1, int(detector_state["consumed"])) latency_summary = {name: _percentiles(values) for name, values in latency.items()} - semantic_summary = { - name: _percentiles(values) for name, values in semantic_latency.items() - } + semantic_summary = {name: _percentiles(values) for name, values in semantic_latency.items()} sensor_decode_summary = { name: _percentiles(values) for name, values in sensor_decode_ms.items() } + runtime_summary = runtime_telemetry.summary() + temporal_track_summary = None if temporal_stabilizer is None else temporal_stabilizer.snapshot() + temporal_semantic_summary = ( + None if semantic_stabilizer is None else semantic_stabilizer.snapshot() + ) acceptance = live["acceptance"] checks = { - "minimum_camera_frames": decoded_frame_count - >= int(acceptance["minimum_camera_frames"]), + "minimum_camera_frames": decoded_frame_count >= int(acceptance["minimum_camera_frames"]), "camera_decoder_accounting": decoder.snapshot()["decoded_frames"] == transport.counts["camera-frame"], "detector_accounting": int(detector_state["consumed"]) @@ -1394,13 +1417,10 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: <= float(acceptance["semantic_maximum_p95_completion_age_ms"]), "minimum_fresh_semantic_coverage": fresh_coverage >= float(acceptance["minimum_fresh_semantic_coverage"]), - "minimum_fused_fraction": fused_fraction - >= float(acceptance["minimum_fused_fraction"]), + "minimum_fused_fraction": fused_fraction >= float(acceptance["minimum_fused_fraction"]), "maximum_p95_decode_age_ms": float(latency_summary["decode_age_ms"]["p95"]) <= float(acceptance["maximum_p95_decode_age_ms"]), - "maximum_p95_world_state_age_ms": float( - latency_summary["world_state_age_ms"]["p95"] - ) + "maximum_p95_world_state_age_ms": float(latency_summary["world_state_age_ms"]["p95"]) <= float(acceptance["maximum_p95_world_state_age_ms"]), "zero_transport_gaps": ( transport.ingress_sequence_gaps == 0 @@ -1424,10 +1444,39 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: "authority_remains_shadow_only": live["authority"] == {"commands_enabled": False, "navigation_or_safety_accepted": False}, } + if stability is not None: + temporal_acceptance = stability["acceptance"] + assert temporal_track_summary is not None + assert temporal_semantic_summary is not None + checks.update( + { + "temporal_camera_processing_p95": float(latency_summary["temporal_2d_3d_ms"]["p95"]) + <= float(temporal_acceptance["maximum_camera_frame_processing_p95_ms"]), + "temporal_semantic_processing_p95": float( + temporal_semantic_summary["processing_ms"]["p95"] + ) + <= float(temporal_acceptance["maximum_semantic_frame_processing_p95_ms"]), + "temporal_track_state_bound": int(temporal_track_summary["peak_track_states"]) + <= int(temporal_acceptance["maximum_track_states_observed"]), + "temporal_semantic_unsupported_change_reduction": float( + temporal_semantic_summary["unsupported_change_reduction_fraction"] + ) + >= float( + temporal_acceptance["minimum_semantic_unsupported_change_reduction_fraction"] + ), + "temporal_rss_growth_bound": float(runtime_summary["rss_growth_mib"]) + <= float(temporal_acceptance["maximum_rss_growth_mib"]), + "temporal_authority_remains_shadow_only": stability["authority"] + == { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + }, + } + ) accepted = all(checks.values()) identity = { "schema_version": IDENTITY_SCHEMA, - "pipeline": PIPELINE_ID, + "pipeline": (PIPELINE_ID if stability is None else INLINE_TEMPORAL_PIPELINE_ID), "session_id": transport.session_id, "bootstrap_job_id": common["job"]["job_id"], "source_id": live["source"]["source_id"], @@ -1436,6 +1485,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: "e14_sha256": common["e14_sha256"], "detector_sha256": common["detector_sha256"], "semantic_sha256": common["semantic_sha256"], + "stability_sha256": common["stability_sha256"], }, "projection_pack": { "id": common["projection_manifest"]["pack_id"], @@ -1498,14 +1548,22 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: "fused_frames": fused_frames, "fused_fraction": fused_fraction, "accepted_cuboids": accepted_cuboids, + "stabilized_cuboids": stabilized_cuboids, "state_counts": dict(fusion_state_counts), "rejection_counts": dict(rejection_counts), "synchronizer": synchronizer.snapshot(), "sensor_decode_ms": sensor_decode_summary, }, "latency_ms": latency_summary, + "temporal_stability": { + "enabled": stability is not None, + "profile_sha256": common["stability_sha256"], + "tracking_2d_3d": temporal_track_summary, + "semantic": temporal_semantic_summary, + "world_memory_objects": len(temporal_world_memory), + }, "gpu_telemetry": gpu.summary(), - "runtime_telemetry": runtime_telemetry.summary(), + "runtime_telemetry": runtime_summary, "process_cpu_seconds": time.process_time() - process_cpu_started, "process_peak_rss_mib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024, "cuda_peak_memory_allocated_mib": torch.cuda.max_memory_allocated() / 2**20, @@ -1547,12 +1605,26 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: "Cross-host source epoch age is diagnostic and excluded from acceptance.", "COCO and Cityscapes models are not forest-domain or safety validated.", "Amodal cuboids infer unobserved volume from class priors.", + *( + [ + "E23 temporal outputs are inline bounded shadow diagnostics, " + "not validated driving or safety authority." + ] + if stability is not None + else [] + ), ], } report_path = output / "run-report.json" _write_json(report_path, report) artifacts = [ _artifact(semantic_path, "e15-semantic-frames", "application/x-ndjson", SEMANTIC_SCHEMA), + _artifact( + raw_fusion_path, + "e23-raw-fusion-frames", + "application/x-ndjson", + FUSION_SCHEMA, + ), _artifact(fusion_path, "e15-fusion-frames", "application/x-ndjson", FUSION_SCHEMA), _artifact(world_path, "e15-world-state", "application/x-ndjson", WORLD_SCHEMA), _artifact(gpu_path, "worker-gpu-telemetry", "application/x-ndjson"), @@ -1590,6 +1662,8 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: "fused_fraction": fused_fraction, "world_state_age_p95_ms": latency_summary["world_state_age_ms"]["p95"], "accepted_cuboids": accepted_cuboids, + "inline_temporal": stability is not None, + "temporal_2d_3d_p95_ms": latency_summary["temporal_2d_3d_ms"]["p95"], }, sort_keys=True, ), diff --git a/src/k1link/compute/__init__.py b/src/k1link/compute/__init__.py index 50436b0..c31fd8f 100644 --- a/src/k1link/compute/__init__.py +++ b/src/k1link/compute/__init__.py @@ -39,6 +39,7 @@ from .lab_instances import ( PublishedTemporalLabInstance, publish_e21_lab_instance, publish_e22_lab_instance, + publish_e23_lab_instance, publish_integrated_lab_instance, ) from .live_perception import ( @@ -148,6 +149,7 @@ __all__ = [ "validate_integrated_perception_result", "publish_e21_lab_instance", "publish_e22_lab_instance", + "publish_e23_lab_instance", "publish_integrated_lab_instance", "validate_multirate_perception_qualification_result", "prepare_recorded_qualification_slice", diff --git a/src/k1link/compute/inline_temporal.py b/src/k1link/compute/inline_temporal.py new file mode 100644 index 0000000..8a12cb9 --- /dev/null +++ b/src/k1link/compute/inline_temporal.py @@ -0,0 +1,628 @@ +"""Bounded inline temporal state for the warm perception worker. + +This module intentionally depends only on the Python standard library and +NumPy so the exact implementation can be included in the minimal, hashed +worker package. It has no command or navigation authority. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import time +from collections import deque +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import numpy as np + +PROFILE_SCHEMA = "missioncore.e23-inline-temporal-profile/v1" +PIPELINE_ID = "warm-worker-inline-bounded-temporal-2d-3d-semantic/v1" +TARGET_CLASS_COUNT = 16 + + +@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 + + +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) + result.append( + self._observe( + state, + source, + frame_index=frame_index, + session_seconds=session_seconds, + stitched=stitched, + ) + ) + + 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-e23") + result.append(held) + + self.peak_states = max(self.peak_states, len(self.states)) + return sorted(result, key=lambda item: int(item["track_id"])) + + def snapshot(self) -> dict[str, int]: + return { + "active_track_states": len(self.states), + "peak_track_states": self.peak_states, + "alias_count": len(self.aliases), + "stitched_tracks": self.stitched_tracks, + "held_2d": self.held_2d, + "held_3d": self.held_3d, + "reset_2d": self.reset_2d, + "reset_3d": self.reset_3d, + } + + 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) + else: + self._hold_cuboid(item, state, session_seconds) + + 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-e23-v1" + item["temporal_status"] = f"e23-{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-e23-v1" + item["geometry"] = "temporally-held-last-supported-cuboid" + item["distance_smoothed_m"] = state.distance_m + item["temporal_status"] = f"e23-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) + + +class StreamingSemanticStabilizer: + """One-mask semantic hysteresis with bounded diagnostic samples.""" + + def __init__(self, profile: dict[str, Any]) -> None: + self.minimum_same_label_neighbors = int(profile["semantic"]["minimum_same_label_neighbors"]) + self.previous_raw: np.ndarray | None = None + self.previous_stabilized: np.ndarray | None = None + self.baseline_unsupported = deque(maxlen=4096) + self.stabilized_unsupported = deque(maxlen=4096) + self.processing_ms = deque(maxlen=4096) + self.frames = 0 + + def update(self, mask: np.ndarray) -> np.ndarray: + if mask.dtype != np.uint8 or mask.shape != (600, 800): + raise RuntimeError("LAB E23 semantic mask is invalid") + started = time.perf_counter() + if self.previous_raw is None or self.previous_stabilized is None: + stabilized = mask.copy() + else: + support = _same_label_neighbor_count(mask) + baseline_change = mask != self.previous_raw + unsupported_baseline = baseline_change & (support < self.minimum_same_label_neighbors) + stabilized = mask.copy() + hold = (mask != self.previous_stabilized) & ( + support < self.minimum_same_label_neighbors + ) + stabilized[hold] = self.previous_stabilized[hold] + stabilized_support = _same_label_neighbor_count(stabilized) + unsupported_stabilized = (stabilized != self.previous_stabilized) & ( + stabilized_support < self.minimum_same_label_neighbors + ) + self.baseline_unsupported.append(float(np.mean(unsupported_baseline))) + self.stabilized_unsupported.append(float(np.mean(unsupported_stabilized))) + self.previous_raw = mask.copy() + self.previous_stabilized = stabilized + self.frames += 1 + self.processing_ms.append((time.perf_counter() - started) * 1000.0) + return stabilized + + def snapshot(self) -> dict[str, Any]: + baseline = _percentiles(self.baseline_unsupported) + stabilized = _percentiles(self.stabilized_unsupported) + baseline_mean = float(baseline["mean"]) + stabilized_mean = float(stabilized["mean"]) + reduction = ( + 0.0 + if baseline_mean <= 0 + else max(0.0, (baseline_mean - stabilized_mean) / baseline_mean) + ) + return { + "frames": self.frames, + "history_masks": int(self.previous_stabilized is not None), + "diagnostic_sample_capacity": self.processing_ms.maxlen, + "baseline_unsupported_change_fraction": baseline, + "stabilized_unsupported_change_fraction": stabilized, + "unsupported_change_reduction_fraction": reduction, + "processing_ms": _percentiles(self.processing_ms), + } + + +def read_inline_profile(path: Path) -> tuple[dict[str, Any], str]: + resolved = path.expanduser().resolve(strict=True) + profile = json.loads(resolved.read_text(encoding="utf-8-sig")) + if not isinstance(profile, dict): + raise RuntimeError("LAB E23 inline temporal profile is invalid") + 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") != "inline-shadow-qualification" + or profile.get("stage") != "warm-worker-after-fusion-before-result-publication" + 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 source.get("calibration_slot") != "camera_1" + or semantic.get("mode") != "spatially-supported-streaming-hysteresis-v1" + or int(semantic.get("class_count", 0)) != TARGET_CLASS_COUNT + 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 RuntimeError("LAB E23 inline 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 RuntimeError("LAB E23 smoothing coefficient is invalid") + if any( + float(acceptance.get(key, 0)) <= 0 + for key in ( + "maximum_camera_frame_processing_p95_ms", + "maximum_semantic_frame_processing_p95_ms", + "maximum_rss_growth_mib", + ) + ): + raise RuntimeError("LAB E23 acceptance contract is invalid") + return profile, _sha256(resolved) + + +def stabilize_world_state( + 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]] = [] + active: set[int] = set() + 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)) + active.add(canonical) + objects.append(item) + for canonical in set(memory) - active: + memory.pop(canonical, None) + world["objects"] = objects + world["object_count"] = len(objects) + world.setdefault("delivery", {})["temporal_stability"] = "e23-inline-bounded" + return world + + +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: + intersection_min = np.maximum(left[:2], right[:2]) + intersection_max = np.minimum(left[2:], right[2:]) + intersection_size = np.maximum(0.0, intersection_max - intersection_min) + intersection = float(np.prod(intersection_size)) + left_area = float(np.prod(np.maximum(0.0, left[2:] - left[:2]))) + right_area = float(np.prod(np.maximum(0.0, right[2:] - right[:2]))) + 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: + x, y, z, w = quaternion + return math.atan2(2.0 * (w * z + x * y), 1.0 - 2.0 * (y * y + z * z)) + + +def _yaw_delta(value: float, reference: float) -> float: + return math.atan2(math.sin(value - reference), math.cos(value - reference)) + + +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: + for key in ( + "cuboid_center_map", + "cuboid_half_size", + "cuboid_quaternion_xyzw", + "observed_cuboid_center_map", + "observed_cuboid_half_size", + "observed_cuboid_quaternion_xyzw", + ): + item[key] = None + item["cuboid_status"] = status + item["temporal_status"] = status + + +def _percentiles(values: Any) -> dict[str, float | int]: + samples = np.asarray(list(values), dtype=np.float64) + if samples.size == 0: + return {"count": 0, "mean": 0.0, "p50": 0.0, "p95": 0.0, "max": 0.0} + return { + "count": int(samples.size), + "mean": round(float(np.mean(samples)), 6), + "p50": round(float(np.percentile(samples, 50)), 6), + "p95": round(float(np.percentile(samples, 95)), 6), + "max": round(float(np.max(samples)), 6), + } + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() diff --git a/src/k1link/compute/lab_instances.py b/src/k1link/compute/lab_instances.py index da79e5d..9118d9f 100644 --- a/src/k1link/compute/lab_instances.py +++ b/src/k1link/compute/lab_instances.py @@ -22,6 +22,7 @@ from k1link.sessions import ( publish_lab_replay_cache, ) +from .inline_temporal import StreamingSemanticStabilizer, read_inline_profile from .integrated_perception import ( IntegratedPerceptionResult, validate_integrated_perception_result, @@ -29,6 +30,7 @@ from .integrated_perception import ( from .jobs import CameraComputeJob, validate_camera_compute_job from .temporal_stability import ( TemporalStabilityBuild, + _quality_metrics, build_temporal_stability_result, ) @@ -224,9 +226,7 @@ def publish_e21_lab_instance( source_result_id=str(e21_document["result_id"]), config_sha256=str(e21_report["identity"]["profile_sha256"]), run_created_at_utc=str(e21_report["created_at_utc"]), - duration_seconds=( - validated.timeline_end_seconds - validated.timeline_start_seconds - ), + duration_seconds=(validated.timeline_end_seconds - validated.timeline_start_seconds), include_recorded_media=False, provenance={ "schema_version": "missioncore.e21-lab-publication/v1", @@ -297,20 +297,14 @@ def publish_e22_lab_instance( ) if not validated.accepted: failed = [ - name - for name, accepted in build.report["acceptance"]["checks"].items() - if not accepted + name for name, accepted in build.report["acceptance"]["checks"].items() if not accepted ] - raise SessionIntegrityError( - f"E22 temporal acceptance failed: {', '.join(failed)}" - ) + 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 + source.job.session_id if source_lab is None else source_lab.source_session_id ) publish_lab_replay_cache( store.data_dir, @@ -330,26 +324,22 @@ def publish_e22_lab_instance( 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 - ), + 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_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"], + "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"], }, ) @@ -361,6 +351,210 @@ def publish_e22_lab_instance( ) +def publish_e23_lab_instance( + *, + repository_root: Path, + reference_result_root: Path, + worker_result_root: Path, + source_report_path: Path, + profile_path: Path, + lab_session_id: str, + lab_id: str, + display_name: str, +) -> PublishedIntegratedLabInstance: + """Publish one accepted inline-temporal 1x worker run as an exact LAB replay.""" + + 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" + reference_path = reference_result_root.expanduser().resolve(strict=True) + reference_document = _read_object(reference_path / "result.json", reference_path) + reference_identity = reference_document.get("identity") + if not isinstance(reference_identity, dict) or not isinstance( + reference_identity.get("job_id"), str + ): + raise SessionIntegrityError("E23 semantic reference has no job identity") + reference = validate_integrated_perception_result( + jobs_root / reference_identity["job_id"], + reference_path, + packs_root, + ) + if not reference.accepted or reference.source_start_frame_index != 0: + raise SessionIntegrityError("E23 reference is not an accepted zero-based run") + + profile, profile_sha256 = read_inline_profile(profile_path) + worker_root = worker_result_root.expanduser().resolve(strict=True) + source_path = source_report_path.expanduser().resolve(strict=True) + worker_document, worker_report, source_report = _validate_e23_inputs( + worker_root, + source_path, + profile_sha256, + ) + frame_count = int(source_report["events_selected"]["camera-frame"]) + if frame_count != reference.frame_count: + raise SessionIntegrityError("E23 source and reference frame counts differ") + + lab_job = _publish_lab_job(reference.job, jobs_root, lab_session_id) + lab_pack = _publish_e21_pack( + reference, + lab_job, + packs_root, + lab_session_id, + frame_count, + visual_projection="accepted-e23-inline-envelope/v1", + ) + lab_result, quality = _publish_e23_visual_result( + reference=reference, + lab_job=lab_job, + lab_pack=lab_pack, + results_root=results_root, + lab_session_id=lab_session_id, + frame_count=frame_count, + worker_root=worker_root, + worker_document=worker_document, + worker_report=worker_report, + source_report=source_report, + profile=profile, + profile_sha256=profile_sha256, + ) + validated = validate_integrated_perception_result( + lab_job.job_root, + lab_result, + packs_root, + ) + if not validated.accepted or not all(quality["checks"].values()): + failed = [name for name, accepted in quality["checks"].items() if not accepted] + raise SessionIntegrityError(f"E23 inline temporal acceptance failed: {', '.join(failed)}") + + store = SessionStore(root) + source_lab = store.get_lab_instance(reference.job.session_id) + source_session_id = ( + reference.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), + ) + temporal = worker_report["metrics"]["temporal_stability"] + 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="e23-inline-temporal-stability", + result_id=validated.result_id, + source_result_id=str(worker_document["result_id"]), + config_sha256=profile_sha256, + run_created_at_utc=str(worker_report["created_at_utc"]), + duration_seconds=(validated.timeline_end_seconds - validated.timeline_start_seconds), + include_recorded_media=False, + provenance={ + "schema_version": "missioncore.e23-lab-publication/v1", + "storage_mode": "bounded-inline-worker-result-and-immutable-source-replay", + "worker_result_id": worker_document["result_id"], + "source_report_sha256": _sha256(source_path), + "reference_result_id": reference.result_id, + "source_payloads_mutated": False, + "lookahead_frames": 0, + "speed": 1.0, + "quality_reductions": quality["reductions"], + "temporal_2d_3d_p95_ms": worker_report["metrics"]["latency_ms"]["temporal_2d_3d_ms"][ + "p95" + ], + "semantic_temporal_p95_ms": temporal["semantic"]["processing_ms"]["p95"], + "peak_track_states": temporal["tracking_2d_3d"]["peak_track_states"], + "rss_growth_mib": worker_report["metrics"]["runtime_telemetry"]["rss_growth_mib"], + }, + ) + return PublishedIntegratedLabInstance( + binding=binding, + job=lab_job, + result=validated, + ) + + +def _validate_e23_inputs( + worker_root: Path, + source_report_path: Path, + profile_sha256: str, +) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: + worker_document = _read_object(worker_root / "result.json", worker_root) + worker_report = _read_object(worker_root / "run-report.json", worker_root) + source_report = _read_object(source_report_path, source_report_path.parent) + worker_identity = worker_document.get("identity") + report_identity = worker_report.get("identity") + if ( + worker_document.get("schema_version") != "missioncore.e15-shadow-inference-result/v1" + or worker_document.get("result_id") != worker_root.name + or worker_document.get("acceptance_state") != "accepted" + or worker_document.get("publication_scope") != "live-shadow-diagnostic-only" + or not isinstance(worker_identity, dict) + or worker_identity.get("pipeline") + != "warm-worker-inline-bounded-temporal-2d-3d-semantic/v1" + or worker_identity.get("profiles", {}).get("stability_sha256") != profile_sha256 + or worker_report.get("schema_version") != "missioncore.e15-shadow-inference-report/v1" + or worker_report.get("result_id") != worker_root.name + or worker_report.get("state") != "accepted" + or report_identity != worker_identity + or not all(worker_report.get("acceptance", {}).get("checks", {}).values()) + or source_report.get("schema_version") != "missioncore.e23-replay-source-report/v1" + or source_report.get("state") != "completed" + or source_report.get("session_id") != worker_identity.get("session_id") + or source_report.get("source", {}).get("speed") != 1.0 + or source_report.get("authority", {}).get("mode") != "shadow-diagnostic-only" + or source_report.get("authority", {}).get("commands_enabled") is not False + or source_report.get("authority", {}).get("navigation_or_safety_accepted") is not False + ): + raise SessionIntegrityError("E23 accepted worker/source identity is inconsistent") + selected = source_report.get("events_selected") + diagnostics = source_report.get("diagnostic_results") + if ( + not isinstance(selected, dict) + or selected.get("camera-frame") != 601 + or selected.get("lidar") != 585 + or selected.get("pose") != 600 + or not isinstance(diagnostics, dict) + or int(diagnostics.get("received", 0)) < 590 + ): + raise SessionIntegrityError("E23 source replay coverage is incomplete") + + required = { + "e15-semantic-frames": "semantic-frames.jsonl", + "e23-raw-fusion-frames": "raw-fusion-frames.jsonl", + "e15-fusion-frames": "fusion-frames.jsonl", + "e15-world-state": "world-state.jsonl", + "worker-gpu-telemetry": "gpu-telemetry.jsonl", + "worker-runtime-telemetry": "runtime-telemetry.jsonl", + "e15-run-report": "run-report.json", + } + artifacts = worker_document.get("artifacts") + descriptors = ( + { + value.get("kind"): value + for value in artifacts + if isinstance(value, dict) and value.get("kind") in required + } + if isinstance(artifacts, list) + else {} + ) + if set(descriptors) != set(required): + raise SessionIntegrityError("E23 worker artifacts are incomplete") + for kind, name in required.items(): + descriptor = descriptors[kind] + path = worker_root / name + if ( + descriptor.get("path") != name + or descriptor.get("byte_length") != path.stat().st_size + or descriptor.get("sha256") != _sha256(path) + ): + raise SessionIntegrityError("E23 worker artifact identity changed") + return worker_document, worker_report, source_report + + def _validate_e21_inputs( e21_root: Path, worker_root: Path, @@ -426,6 +620,8 @@ def _publish_e21_pack( packs_root: Path, lab_session_id: str, frame_count: int, + *, + visual_projection: str = "accepted-e21-envelope/v1", ) -> Path: source_manifest = _read_object(reference.pack_root / "manifest.json", reference.pack_root) with np.load(reference.pack_root / "lidar-pack.npz", allow_pickle=False) as arrays: @@ -438,9 +634,9 @@ def _publish_e21_pack( "cloud_offsets": arrays["cloud_offsets"][: frame_count + 1].copy(), "cloud_points_map": arrays["cloud_points_map"][:cloud_end].copy(), "pose_positions_map": arrays["pose_positions_map"][:frame_count].copy(), - "pose_quaternions_map_from_lidar": arrays[ - "pose_quaternions_map_from_lidar" - ][:frame_count].copy(), + "pose_quaternions_map_from_lidar": arrays["pose_quaternions_map_from_lidar"][ + :frame_count + ].copy(), "lidar_camera_delta_ms": arrays["lidar_camera_delta_ms"][:frame_count].copy(), "pose_point_delta_ms": arrays["pose_point_delta_ms"][:frame_count].copy(), "intrinsic_fx_fy_cx_cy": arrays["intrinsic_fx_fy_cx_cy"].copy(), @@ -460,7 +656,7 @@ def _publish_e21_pack( "point_count": int(payload["cloud_points_map"].shape[0]), "timeline_start_seconds": float(payload["session_seconds"][0]), "timeline_end_seconds": float(payload["session_seconds"][-1]), - "visual_projection": "accepted-e21-envelope/v1", + "visual_projection": visual_projection, } ) identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() @@ -580,9 +776,7 @@ def _publish_e21_visual_result( ) fusion_rows.append(normalized_fusion) world_rows.append(normalized_world) - expected_drops = int( - e21_report["metrics"]["worker"]["detector"]["queue"]["dropped_overflow"] - ) + expected_drops = int(e21_report["metrics"]["worker"]["detector"]["queue"]["dropped_overflow"]) if len(dropped_indices) != expected_drops: raise SessionIntegrityError("E21 detector replacement accounting changed") @@ -762,6 +956,329 @@ def _publish_e21_visual_result( return destination +def _publish_e23_visual_result( + *, + reference: IntegratedPerceptionResult, + lab_job: CameraComputeJob, + lab_pack: Path, + results_root: Path, + lab_session_id: str, + frame_count: int, + worker_root: Path, + worker_document: dict[str, Any], + worker_report: dict[str, Any], + source_report: dict[str, Any], + profile: dict[str, Any], + profile_sha256: str, +) -> tuple[Path, dict[str, Any]]: + with np.load(reference.arrays_path, allow_pickle=False) as arrays: + frame_times = arrays["frame_times_ns"][:frame_count].copy() + reference_semantic_indices = arrays["semantic_frame_indices"] + selected = reference_semantic_indices < frame_count + reference_indices = reference_semantic_indices[selected].copy() + reference_masks = arrays["semantic_masks"][selected].copy() + + semantic_rows = _read_jsonl(worker_root / "semantic-frames.jsonl") + if [row.get("frame_index") for row in semantic_rows] != reference_indices.tolist(): + raise SessionIntegrityError("E23 semantic frame schedule changed") + semantic_stabilizer = StreamingSemanticStabilizer(profile) + stabilized_masks = np.stack( + [semantic_stabilizer.update(mask) for mask in reference_masks] + ).astype(np.uint8, copy=False) + for row, mask in zip(semantic_rows, stabilized_masks, strict=True): + if row.get("mask_sha256") != hashlib.sha256(mask.tobytes()).hexdigest(): + raise SessionIntegrityError("E23 semantic mask does not match inline reconstruction") + row["schema_version"] = "missioncore.e10-semantic-frame/v1" + row["session_seconds"] = float(frame_times[int(row["frame_index"])]) / 1_000_000_000 + row["temporal_status"] = "e23-inline-spatially-supported-hysteresis" + + raw_worker_rows = _read_jsonl(worker_root / "raw-fusion-frames.jsonl") + stable_worker_rows = _read_jsonl(worker_root / "fusion-frames.jsonl") + fusion_source = {int(row["source_frame_index"]): row for row in stable_worker_rows} + world_source = { + int(row["source_frame_index"]): row + for row in _read_jsonl(worker_root / "world-state.jsonl") + } + if set(fusion_source) != set(world_source): + raise SessionIntegrityError("E23 fusion and world timelines differ") + fusion_rows: list[dict[str, Any]] = [] + world_rows: list[dict[str, Any]] = [] + dropped_indices: list[int] = [] + for index in range(frame_count): + session_seconds = float(frame_times[index]) / 1_000_000_000 + fusion = fusion_source.get(index) + world = world_source.get(index) + if fusion is None or world is None: + dropped_indices.append(index) + fusion_rows.append( + { + "schema_version": "missioncore.e10-fusion-frame/v1", + "frame_index": index, + "source_frame_index": index, + "session_seconds": session_seconds, + "fusion_state": "detector-dropped-latest-wins", + "semantic_source_frame_index": None, + "semantic_status": "unavailable", + "objects": [], + } + ) + world_rows.append(_dropped_world_row(index, session_seconds)) + continue + normalized_fusion = json.loads(json.dumps(fusion)) + normalized_fusion.update( + { + "schema_version": "missioncore.e10-fusion-frame/v1", + "frame_index": index, + "source_frame_index": index, + "session_seconds": session_seconds, + } + ) + normalized_world = json.loads(json.dumps(world)) + normalized_world.update( + { + "frame_index": index, + "source_frame_index": index, + "session_seconds": session_seconds, + } + ) + fusion_rows.append(normalized_fusion) + world_rows.append(normalized_world) + expected_drops = int(worker_report["metrics"]["detector"]["queue"]["dropped_overflow"]) + if len(dropped_indices) != expected_drops: + raise SessionIntegrityError("E23 detector replacement accounting changed") + + baseline = _quality_metrics(raw_worker_rows, reference_masks) + stabilized = _quality_metrics(stable_worker_rows, stabilized_masks) + reductions = { + "tracking_2d_acceleration_p95_fraction": _fraction_reduction( + baseline["tracking_2d"]["normalized_acceleration"]["p95"], + stabilized["tracking_2d"]["normalized_acceleration"]["p95"], + ), + "tracking_2d_size_step_p95_fraction": _fraction_reduction( + baseline["tracking_2d"]["normalized_size_step"]["p95"], + stabilized["tracking_2d"]["normalized_size_step"]["p95"], + ), + "cuboid_center_step_p95_fraction": _fraction_reduction( + baseline["cuboids_3d"]["center_step_m"]["p95"], + stabilized["cuboids_3d"]["center_step_m"]["p95"], + ), + "cuboid_size_step_p95_fraction": _fraction_reduction( + baseline["cuboids_3d"]["half_size_step_m"]["p95"], + stabilized["cuboids_3d"]["half_size_step_m"]["p95"], + ), + "cuboid_yaw_step_p95_fraction": _fraction_reduction( + baseline["cuboids_3d"]["yaw_step_degrees"]["p95"], + stabilized["cuboids_3d"]["yaw_step_degrees"]["p95"], + ), + "semantic_unsupported_change_fraction": float( + worker_report["metrics"]["temporal_stability"]["semantic"][ + "unsupported_change_reduction_fraction" + ] + ), + } + temporal = worker_report["metrics"]["temporal_stability"] + acceptance = profile["acceptance"] + quality_checks = { + "worker_runtime_accepted": worker_report["state"] == "accepted" + and all(worker_report["acceptance"]["checks"].values()), + "source_is_complete_1x": source_report["state"] == "completed" + and source_report["source"]["speed"] == 1.0, + "minimum_2d_acceleration_reduction": reductions["tracking_2d_acceleration_p95_fraction"] + >= float(acceptance["minimum_2d_acceleration_p95_reduction_fraction"]), + "minimum_3d_center_reduction": reductions["cuboid_center_step_p95_fraction"] + >= float(acceptance["minimum_3d_center_step_p95_reduction_fraction"]), + "minimum_3d_yaw_reduction": reductions["cuboid_yaw_step_p95_fraction"] + >= float(acceptance["minimum_3d_yaw_step_p95_reduction_fraction"]), + "minimum_semantic_unsupported_change_reduction": reductions[ + "semantic_unsupported_change_fraction" + ] + >= float(acceptance["minimum_semantic_unsupported_change_reduction_fraction"]), + "maximum_camera_frame_processing_p95": float( + worker_report["metrics"]["latency_ms"]["temporal_2d_3d_ms"]["p95"] + ) + <= float(acceptance["maximum_camera_frame_processing_p95_ms"]), + "maximum_semantic_frame_processing_p95": float(temporal["semantic"]["processing_ms"]["p95"]) + <= float(acceptance["maximum_semantic_frame_processing_p95_ms"]), + "maximum_track_states": int(temporal["tracking_2d_3d"]["peak_track_states"]) + <= int(acceptance["maximum_track_states_observed"]), + "maximum_rss_growth": float(worker_report["metrics"]["runtime_telemetry"]["rss_growth_mib"]) + <= float(acceptance["maximum_rss_growth_mib"]), + } + quality = { + "schema_version": "missioncore.e23-inline-quality/v1", + "baseline": baseline, + "stabilized": stabilized, + "reductions": reductions, + "checks": quality_checks, + } + if not all(quality_checks.values()): + failed = [name for name, accepted in quality_checks.items() if not accepted] + raise SessionIntegrityError(f"E23 inline temporal quality failed: {', '.join(failed)}") + + box_offsets = [0] + centers: list[list[float]] = [] + half_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"]) + half_sizes.append(item["cuboid_half_size"]) + quaternions.append(item["cuboid_quaternion_xyzw"]) + colors.append(_cuboid_color(item)) + box_offsets.append(len(centers)) + + worker_identity = worker_document["identity"] + configuration = { + "pipeline": "e23-inline-temporal-envelope-visual-projection/v1", + "profile_sha256": profile_sha256, + "profile": profile, + "worker_result_id": worker_document["result_id"], + "source_report": { + "schema_version": source_report["schema_version"], + "session_id": source_report["session_id"], + "speed": source_report["source"]["speed"], + }, + "semantic_mask_materialization": { + "mode": "inline-reconstruction-from-immutable-reference-exact-sha256", + "reference_result_id": reference.result_id, + "matched_masks": len(semantic_rows), + }, + "quality": quality, + } + selection = { + "frame_count": frame_count, + "source_start_frame_index": 0, + "source_end_frame_index": frame_count - 1, + "timeline_start_seconds": float(frame_times[0]) / 1_000_000_000, + "timeline_end_seconds": float(frame_times[-1]) / 1_000_000_000, + "timeline_sha256": hashlib.sha256(frame_times.tobytes()).hexdigest(), + } + identity = { + "schema_version": "missioncore.e10-integrated-perception-identity/v1", + "job_id": lab_job.job_id, + "input_sha256": lab_job.input_sha256, + "session_id": lab_session_id, + "source_id": lab_job.source_id, + "lidar_pack_id": lab_pack.name, + "selection": selection, + "configuration": configuration, + "models": worker_identity["models"], + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + result_id = f"e10-integrated-perception-{identity_sha256}" + destination = results_root / result_id + if destination.exists(): + existing = _read_object(destination / "result.json", destination) + if existing.get("identity") != identity: + raise SessionIntegrityError("E23 LAB visual result id collides") + return destination, quality + + staging = _staging_directory(results_root, result_id) + 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) + np.savez_compressed( + arrays_path, + frame_times_ns=frame_times.astype(np.int64, copy=False), + semantic_frame_indices=reference_indices.astype(np.int64, copy=False), + semantic_masks=stabilized_masks.astype(np.uint8, copy=False), + support_offsets=np.zeros(frame_count + 1, dtype=np.int64), + support_points=np.empty((0, 3), dtype=np.float32), + support_colors=np.empty((0, 3), dtype=np.uint8), + 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(half_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)), + ) + shutil.copyfile(worker_root / "gpu-telemetry.jsonl", gpu_path) + report = { + "schema_version": "missioncore.e10-integrated-perception-report/v1", + "result_id": result_id, + "created_at_utc": worker_report["created_at_utc"], + "state": "accepted", + "ground_truth": False, + "identity": identity, + "acceptance": { + "accepted": True, + "navigation_or_safety_accepted": False, + "checks": quality_checks, + }, + "metrics": { + **worker_report["metrics"], + "quality": quality, + "visual_projection": { + "frames": frame_count, + "semantic_masks": len(semantic_rows), + "detector_replacement_frames": dropped_indices, + "accepted_cuboids": len(centers), + }, + }, + "runtime": worker_report.get("runtime", {}), + "limitations": [ + "This is the accepted E23 recorded 1x inline worker gate, not a physical K1 run.", + "Latest-wins detector replacements are explicit empty visual frames.", + "Semantic pixels are reconstructed only after exact inline SHA-256 matches.", + "LiDAR support points remain in the immutable source scene and are not duplicated.", + "Navigation and safety authority remain disabled.", + ], + } + write_json_atomic(report_path, report) + artifacts = [ + _artifact_descriptor( + "e10-semantic-frames", + semantic_path, + "missioncore.e10-semantic-frame/v1", + ), + _artifact_descriptor( + "e10-fusion-frames", + fusion_path, + "missioncore.e10-fusion-frame/v1", + ), + _artifact_descriptor( + "e10-world-state", + world_path, + "missioncore.live-perception-world-state/v1", + ), + _artifact_descriptor("e10-transient-perception", arrays_path, None), + _artifact_descriptor("worker-gpu-telemetry", gpu_path, None), + _artifact_descriptor( + "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": worker_report["created_at_utc"], + "ground_truth": False, + "publication_scope": "recorded-integrated-realtime-qualification-only", + "acceptance_state": "accepted", + "frames_processed": frame_count, + "artifacts": artifacts, + }, + ) + _publish_directory(staging, destination) + finally: + _remove_staging(staging) + return destination, quality + + def _dropped_world_row(frame_index: int, session_seconds: float) -> dict[str, Any]: return { "schema_version": "missioncore.live-perception-world-state/v1", @@ -799,6 +1316,13 @@ def _cuboid_color(item: dict[str, Any]) -> list[int]: return [64 + digest[0] % 176, 64 + digest[1] % 176, 64 + digest[2] % 176, 88] +def _fraction_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 _read_jsonl(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] with path.open(encoding="utf-8") as stream: diff --git a/src/k1link/device_plugins/xgrids_k1/cli.py b/src/k1link/device_plugins/xgrids_k1/cli.py index 97fe81c..45c44d0 100644 --- a/src/k1link/device_plugins/xgrids_k1/cli.py +++ b/src/k1link/device_plugins/xgrids_k1/cli.py @@ -21,6 +21,7 @@ from k1link.compute import ( prepare_camera_compute_job, publish_e21_lab_instance, publish_e22_lab_instance, + publish_e23_lab_instance, publish_integrated_lab_instance, ) from k1link.device_plugins.xgrids_k1.analyze import ( @@ -497,6 +498,89 @@ def publish_e22_lab( ) +@lab_app.command("publish-e23") +def publish_e23_lab( + reference: Annotated[ + Path, + typer.Option( + exists=True, + file_okay=False, + readable=True, + resolve_path=True, + help="Accepted zero-based E10 result containing exact source masks.", + ), + ], + worker_result: Annotated[ + Path, + typer.Option( + "--worker-result", + exists=True, + file_okay=False, + readable=True, + resolve_path=True, + help="Accepted E23 inline-temporal worker result.", + ), + ], + source_report: Annotated[ + Path, + typer.Option( + "--source-report", + exists=True, + dir_okay=False, + readable=True, + resolve_path=True, + help="Exact E23 1x replay source report.", + ), + ], + profile: Annotated[ + Path, + typer.Option( + exists=True, + dir_okay=False, + readable=True, + resolve_path=True, + help="Bounded E23 inline-temporal 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 E23.2'."), + ], + display_name: Annotated[ + str, + typer.Option("--display-name", help="Operator-facing saved-session title."), + ], +) -> None: + """Publish an accepted inline-temporal 1x run as an immutable LAB replay.""" + + repository_root = Path(__file__).resolve().parents[4] + try: + published = publish_e23_lab_instance( + repository_root=repository_root, + reference_result_root=reference, + worker_result_root=worker_result, + source_report_path=source_report, + profile_path=profile, + lab_session_id=session_id, + lab_id=lab_id, + display_name=display_name, + ) + except (OSError, SessionIntegrityError, RuntimeError, ValueError) as exc: + console.print(f"[red]E23 LAB publication failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + console.print( + "[green]E23 LAB instance published.[/green] " + f"session={published.binding.session_id}; " + f"source={published.binding.source_session_id}; " + f"result={published.binding.result_id}; " + "speed=1.0; source_payloads_mutated=false" + ) + + @app.command("serve") def serve_console( port: Annotated[ diff --git a/tests/test_e15_worker_package.py b/tests/test_e15_worker_package.py index 5554bf1..db2b98d 100644 --- a/tests/test_e15_worker_package.py +++ b/tests/test_e15_worker_package.py @@ -28,11 +28,11 @@ def test_e15_worker_package_is_minimal_hash_addressed_projection(tmp_path: Path) manifest = module.validate_worker_package(package) assert package.name == f"e15-worker-package-{manifest['identity_sha256']}" - assert manifest["identity"]["classification"] == ( - "minimal-live-worker-import-projection" - ) - assert len(manifest["artifacts"]) == 11 + assert manifest["identity"]["classification"] == ("minimal-live-worker-import-projection") + assert len(manifest["artifacts"]) == 12 + assert (package / "k1link" / "compute" / "inline_temporal.py").is_file() assert not (package / "k1link" / "device_plugins" / "xgrids_k1" / "mqtt").exists() - assert "observation" not in ( - package / "k1link" / "device_plugins" / "xgrids_k1" / "__init__.py" - ).read_text() + assert ( + "observation" + not in (package / "k1link" / "device_plugins" / "xgrids_k1" / "__init__.py").read_text() + ) diff --git a/tests/test_e23_inline_temporal.py b/tests/test_e23_inline_temporal.py new file mode 100644 index 0000000..a9ac591 --- /dev/null +++ b/tests/test_e23_inline_temporal.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest + +from k1link.compute.inline_temporal import ( + StreamingSemanticStabilizer, + TemporalStabilizer, + read_inline_profile, + stabilize_world_state, +) + + +def _profile() -> dict[str, object]: + root = Path(__file__).resolve().parents[1] + profile, digest = read_inline_profile( + root / "experiments" / "perception" / "e23_inline_temporal_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_e23_profile_pins_inline_stage_and_has_no_control_authority() -> None: + profile = _profile() + + assert profile["stage"] == "warm-worker-after-fusion-before-result-publication" + assert profile["bounds"]["maximum_track_states"] == 128 + assert profile["bounds"]["semantic_history_masks"] == 1 + assert profile["authority"] == { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + } + + +def test_e23_inline_tracking_is_bounded_and_updates_world_state() -> 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])], + ) + world = stabilize_world_state( + { + "objects": [ + { + "track_id": 11, + "velocity_status": "observed", + } + ], + "delivery": {}, + }, + second, + {}, + ) + + assert first[0]["track_id"] == 10 + assert second[0]["track_id"] == 10 + assert second[0]["cuboid_status"] == "accepted-temporally-stabilized-e23-v1" + assert world["delivery"]["temporal_stability"] == "e23-inline-bounded" + assert world["objects"][0]["track_id"] == 10 + assert stabilizer.snapshot()["peak_track_states"] <= 128 + + +def test_e23_semantic_state_keeps_only_one_mask_and_rejects_islands() -> None: + stabilizer = StreamingSemanticStabilizer(_profile()) + first = np.zeros((600, 800), dtype=np.uint8) + second = first.copy() + second[20, 20] = 4 + second[100:110, 100:110] = 7 + + stabilizer.update(first) + output = stabilizer.update(second) + snapshot = stabilizer.snapshot() + + assert output[20, 20] == 0 + assert np.all(output[102:108, 102:108] == 7) + assert snapshot["history_masks"] == 1 + assert snapshot["frames"] == 2 + assert snapshot["unsupported_change_reduction_fraction"] > 0 + + +def test_e23_profile_rejects_non_object_json(tmp_path: Path) -> None: + profile_path = tmp_path / "profile.json" + profile_path.write_text("[]", encoding="utf-8") + + with pytest.raises(RuntimeError, match="profile is invalid"): + read_inline_profile(profile_path)