From 631ba50f8826d535f4446d3d2a2c828d02d1a2d6 Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 18:54:43 +0530 Subject: [PATCH 1/9] docs: make README honest about implemented vs planned features MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase A1 of the 10x roadmap. The README advertised several capabilities as shipped that aren't wired into the live pipeline; this marks them as planned and fixes concrete drift. - Add an "Implementation status" table up front (source of truth: βœ… vs πŸ”­) - Mark πŸ”­ planned: cross-camera homography (H-PROJ), pixel extrapolation (EXTRAP), appearance re-ID, sensor-trust scoring, adaptive-Kalman-by-area, GPS/IMU fusion, DeepSORT/centroid fallback chain, compass ribbon, threat ring - Reframe the homography section as forward-looking design - Note only the orange WORLD ghost path is emitted today - Fix mobile handshake (no target_fps is sent; sensor_data is received, not fused) - Update Testing section for the new frontend tests + comprehensive CI gate - pyproject: allow Python 3.13 (requires-python <3.14; verified locally) - .env.example: remove the duplicated JWT/AUTH block No code behaviour change; backend gate green (ruff/mypy/57 tests). Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 108 ++++++++++++++++++++++++++++++++----------- backend/.env.example | 7 +-- pyproject.toml | 2 +- 3 files changed, 84 insertions(+), 33 deletions(-) diff --git a/README.md b/README.md index ccfabc2..fb9c78b 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,38 @@ The system runs a **singleton perception pipeline**: detection, tracking, and fu --- +## βœ… Implementation status + +OVERWATCH ships a working perception core today. Several headline capabilities are +**designed and on the roadmap but not yet wired into the live pipeline** β€” this table is +the source of truth. Rows marked πŸ”­ are *planned*, not implemented; the sections below +that describe them are forward-looking design. + +| Capability | Status | +|---|---| +| YOLOv8 person detection (TensorRT / ONNX / PyTorch) | βœ… Implemented | +| Per-camera Hungarian / IoU tracking | βœ… Implemented | +| 6-state Kalman world fusion (confidence-adaptive) | βœ… Implemented | +| Cross-camera merge by world-space distance | βœ… Implemented | +| World-projection ghost predictions (orange `WORLD`) | βœ… Implemented | +| msgpack zero-copy WebSocket broadcast | βœ… Implemented | +| Canvas tactical HUD (brackets, diamonds, velocity, ghosts) | βœ… Implemented | +| Mobile phone camera source + standalone page | βœ… Implemented | +| Optional JWT auth Β· SSL Β· atomic Jetson deploy | βœ… Implemented | +| Cross-camera homography ghosts (green `H-PROJ`) | πŸ”­ Planned | +| Pixel-extrapolation ghosts (red `EXTRAP`) | πŸ”­ Planned | +| Appearance re-ID (HSV histograms) wired into tracking / fusion | πŸ”­ Planned | +| Sensor-trust scoring | πŸ”­ Planned | +| Adaptive Kalman noise by bbox area | πŸ”­ Planned | +| GPS + IMU fusion into the world model | πŸ”­ Planned | +| DeepSORT / centroid tracker fallback chain | πŸ”­ Planned | + +> Config and helper scaffolding for the πŸ”­ items already exists (`compute_appearance()`, +> the `HOMOGRAPHY_*` / `GPS_REFERENCE_*` env vars), but those paths are **not yet active** +> in the pipeline. The build order is tracked in [`docs/superpowers/specs/`](docs/superpowers/specs/). + +--- + ## πŸš€ Features ### Core perception @@ -90,13 +122,13 @@ The system runs a **singleton perception pipeline**: detection, tracking, and fu |---|---| | **Person detection** | YOLOv8n with NMS-level class filter (`classes=[0]`) β€” person-only | | **TensorRT FP16** | `.engine` export on Jetson β€” ~8 MiB, sub-10 ms inference | -| **Hungarian tracking** | `scipy.optimize.linear_sum_assignment` β€” `0.6 Γ— IoU + 0.4 Γ— cosine appearance` cost | -| **Tracker fallback chain** | DeepSORT (MobileNet) β†’ Hungarian (scipy) β†’ Centroid | -| **Adaptive Kalman filter** | 6-state `[x, y, z, vx, vy, vz]` β€” measurement noise scales by confidence, bbox area, sensor trust | -| **Cross-camera re-ID** | 64-dim HSV histogram descriptors, L2-normalized, EMA-smoothed (Ξ± = 0.3) | -| **Sensor trust scoring** | Per-sensor trust ∈ [0.1, 1.0] β€” increases for consistent measurements, decays for innovation outliers | -| **Cross-camera homography** | Self-calibrating ground-plane H from shared foot-point observations via `cv2.findHomography` + RANSAC | -| **3-path ghost predictions** | (A) homography projection from any source camera (green), (B) pixel extrapolation with adaptive budget (red), (C) world-coordinate pinhole projection fallback (orange) | +| **Hungarian tracking** | `scipy.optimize.linear_sum_assignment` β€” cost `0.6 Γ— IoU + 0.4 Γ— cosine appearance` (appearance term is inactive until re-ID lands, so tracking is effectively pure-IoU today) | +| **Tracker fallback chain** πŸ”­ | *Planned* β€” only Hungarian (with a greedy fallback) is active today; DeepSORT/Centroid are not implemented | +| **Adaptive Kalman filter** | 6-state `[x, y, z, vx, vy, vz]` β€” measurement noise scales by **confidence** (bbox-area & sensor-trust scaling πŸ”­ planned) | +| **Cross-camera re-ID** πŸ”­ | *Planned* β€” 64-dim HSV histogram descriptors (`compute_appearance()` exists but isn't yet wired into detection/tracking) | +| **Sensor trust scoring** πŸ”­ | *Planned* β€” per-sensor trust ∈ [0.1, 1.0]; the Kalman update accepts the param but it is fixed at 1.0 today | +| **Cross-camera homography** πŸ”­ | *Planned* β€” self-calibrating ground-plane H via `cv2.findHomography` + RANSAC (no homography code in the pipeline yet) | +| **Ghost predictions** | Path C β€” world-coordinate pinhole projection (orange `WORLD`) is **active**. Path A (homography/green `H-PROJ`) and Path B (pixel extrapolation/red `EXTRAP`) are πŸ”­ planned | ### Platform @@ -104,7 +136,7 @@ The system runs a **singleton perception pipeline**: detection, tracking, and fu |---|---| | **Multi-camera** | Up to 4 concurrent streams (physical MJPEG/RTSP + mobile virtual cameras) | | **Mobile streaming** | Phone browsers β†’ `getUserMedia` β†’ binary JPEG over WebSocket β†’ `VirtualCamera` | -| **GPS + IMU fusion** | Mobile geolocation β†’ equirectangular projection; `DeviceOrientationEvent` β†’ camera rotation | +| **GPS + IMU fusion** πŸ”­ | *Planned* β€” the mobile client sends GPS (`watchPosition`) + IMU (`DeviceOrientationEvent`), but the backend currently receives and discards `sensor_data`; fusion into the world model is not wired yet | | **AR overlays** | Canvas-based: cyan detection brackets, amber track boxes, green/orange/red ghost predictions | | **Binary protocol** | msgpack-serialized snapshots β€” zero-copy broadcast to all viewers | | **SSL/TLS** | Self-signed certificates with SAN for LAN IP access (required for `getUserMedia`) | @@ -320,13 +352,21 @@ python scripts/check_status.py ## πŸ§ͺ Testing -The backend has 57 unit tests covering domain primitives, Kalman filtering, coordinate transforms, tracking, and configuration. CI runs them on every push and PR. +**Backend** β€” 57 unit tests covering domain primitives, Kalman filtering, coordinate transforms, tracking, and configuration: ```bash python -m pytest backend/tests/unit -v ``` -Tests are pure-Python and do not require CUDA, ultralytics, or torch. They use `pytest.importorskip("cv2")` where OpenCV is needed. +Tests are pure-Python and do not require CUDA, ultralytics, or torch (heavy deps are import-skipped). + +**Frontend** β€” Jest + React Testing Library smoke tests for domain helpers and components: + +```bash +cd frontend && npm run test:ci +``` + +**CI** runs the full gate on every push and PR β€” a backend job (`ruff` + `mypy` + `pytest` with a coverage floor) and a frontend job (`jest`/RTL + `eslint` via the production build). --- @@ -357,9 +397,9 @@ When `AUTH_ENABLED=true`, both endpoints require a `?token=` query paramete ``` Client β†’ { "type": "register", "role": "camera_source", "camera_id": null } -Server β†’ { "type": "registered", "camera_id": 0, "target_fps": 15 } -Client β†’ [binary JPEG frames at target FPS] -Client β†’ { "type": "sensor_data", "gps": {...}, "orientation": {...} } +Server β†’ { "type": "registered", "camera_id": 0 } +Client β†’ [binary JPEG frames] +Client β†’ { "type": "sensor_data", "gps": {...}, "orientation": {...} } # received, not yet fused πŸ”­ ``` --- @@ -419,19 +459,19 @@ Client β†’ { "type": "sensor_data", "gps": {...}, "orientation": {...} } ## 🎯 AR overlay system β€” EagleEye-inspired tactical HUD -The frontend renders a tactical HUD inspired by Anduril's EagleEye UI β€” diamond IFF markers, compass ribbon, threat rings β€” implemented entirely in HTML5 Canvas. (Visual style only; rendered from open code, no Anduril assets used.) +The frontend renders a tactical HUD inspired by Anduril's EagleEye UI, implemented entirely in HTML5 Canvas β€” corner brackets, diamond IFF markers, velocity vectors, ghost chevrons, and HUD corner ticks. (Compass ribbon and threat ring are πŸ”­ planned, not in the current renderer. Visual style only; rendered from open code, no Anduril assets used.) | Layer | Color | Elements | |---|---|---| | **Detections** | Slate-blue `#64b5f6` | Diamond markers, corner brackets, `PERSON` confidence pill | | **Tracks** | Amber `#ffd740` | Diamond/chevron markers, velocity vector arrows, track ID callouts | -| **Predictions (H-PROJ)** | Green `#00ff82` solid | Homography-projected ghost β€” accurate, real-time cross-camera | -| **Predictions (EXTRAP)** | Red `#ff5050` dashed | Pixel-extrapolated ghost β€” time-decaying dead-reckoning | -| **Predictions (WORLD)** | Orange `#ff9800` dashed | World-coordinate projection β€” pinhole-model fallback | -| **Compass ribbon** | β€” | Heading ribbon with N/E/S/W and bearing tick marks | -| **Threat ring** | Per-IFF color | Inner ring around feed showing bearing to off-screen predictions | +| **Predictions (H-PROJ)** πŸ”­ | Green `#00ff82` solid | *Planned* β€” homography-projected ghost (the renderer supports this color, but the backend never emits this method yet) | +| **Predictions (EXTRAP)** πŸ”­ | Red `#ff5050` dashed | *Planned* β€” pixel-extrapolated ghost (renderer-ready; not emitted by the backend yet) | +| **Predictions (WORLD)** | Orange `#ff9800` dashed | βœ… Active β€” world-coordinate pinhole projection; the only ghost path emitted today | +| **Compass ribbon** πŸ”­ | β€” | *Planned* β€” not in the current canvas renderer | +| **Threat ring** πŸ”­ | Per-IFF color | *Planned* β€” not in the current canvas renderer | -Detection overlays show what the model sees *right now*. Track overlays show persistent identity across frames. Predictions show cross-camera projections β€” green for homography (most accurate), orange for world-model fallback (rough but always available), red for pixel extrapolation (last resort). +Detection overlays show what the model sees *right now*. Track overlays show persistent identity across frames. Predictions show cross-camera projections β€” **today only the orange `WORLD` path is emitted**; green (homography) and red (extrapolation) are πŸ”­ planned for Phase B. --- @@ -446,25 +486,35 @@ Each fused world object maintains a 6-state Kalman filter `[x, y, z, vx, vy, vz] Objects from different cameras are matched when: - Euclidean distance < 2 m - Same `class_id` -- Appearance cosine similarity > 0.5 (when feature vectors available) +- πŸ”­ *Planned:* appearance cosine similarity > 0.5 (today the match is distance + class only; the appearance gate is not yet wired in) -### Sensor trust +### Sensor trust πŸ”­ *(planned)* -Each camera/sensor earns trust through consistency: +The intended design earns per-sensor trust through consistency: - **Consistent measurements** β†’ trust increases (capped at 1.0) - **Innovation outliers** β†’ trust decays (floored at 0.1) -### Appearance re-ID +Today the Kalman update accepts a `sensor_trust` argument but it is fixed at `1.0`. + +### Appearance re-ID πŸ”­ *(planned)* -- 64-dimensional HSV histogram descriptors computed per detection (~0.1 ms each) +- 64-dimensional HSV histogram descriptors (`compute_appearance()` is implemented…) - L2-normalized for cosine similarity - Exponential moving average (Ξ± = 0.3) for descriptor stability across frames +…but `compute_appearance()` is **not yet called** in the pipeline, so `Detection.appearance` +is always `None` and re-ID is dormant until Phase B. + --- -## πŸ“ Cross-camera homography β€” how it works +## πŸ“ Cross-camera homography β€” planned design πŸ”­ + +> **Status: not yet implemented.** This section is the roadmap for the green `H-PROJ` +> ghost path. Today only Path C (world projection, orange `WORLD`) is active β€” there is no +> `cv2.findHomography` / foot-point collection in the pipeline yet. The text below describes +> how the homography path *will* work once built (Phase B of the roadmap). -The signature feature is **ghost prediction**: when Camera 0 can't see a person but Camera 1 can, the system renders a ghost overlay on Camera 0's feed showing where that person is. +The planned signature feature is **homography ghost prediction**: when Camera 0 can't see a person but Camera 1 can, the system will render a ghost overlay on Camera 0's feed showing where that person is. ### The problem with naive extrapolation @@ -526,6 +576,10 @@ The mobile client: ### Cross-camera prediction +> Rows that mention homography / Path A / Path B describe πŸ”­ **planned** behavior (see the +> [Implementation status](#-implementation-status) table). Path C (world projection, orange) +> is the only path active today. + | Edge case | Behavior | Mitigation | |---|---|---| | **No homography learned yet** | Path A fails silently; falls through to Path B (extrap) or Path C (world projection). Ghost appears orange instead of green. | Walk through overlapping camera FOVs to collect β‰₯ 4 foot-point pairs. Homography auto-learns within ~5 s of co-visibility. | diff --git a/backend/.env.example b/backend/.env.example index 970db21..aa9e488 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -90,10 +90,7 @@ HOMOGRAPHY_MOVEMENT_THRESHOLD=1.0 GPS_REFERENCE_LAT= GPS_REFERENCE_LNG= -# Security (additive, default-off) +# --- Security (additive, default-off) --- # CORS_ORIGINS='["https://app.example.com"]' # JSON list; default ["*"] # MAX_WS_CLIENTS=100 - -# Optional JWT auth (default off) -# AUTH_ENABLED=true -# JWT_SECRET=replace-with-strong-random-secret-min-32-chars +# (JWT auth is configured in the "JWT Authentication" block above.) diff --git a/pyproject.toml b/pyproject.toml index 6d3c30b..2fef04e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta" name = "overwatch" version = "2.0.0" description = "Multi-camera perception and tracking system" -requires-python = ">=3.10,<3.13" +requires-python = ">=3.10,<3.14" [tool.pytest.ini_options] minversion = "7.0" From 32cee74b58a0121a3447bf3f2b9b2700260f69a4 Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 18:59:12 +0530 Subject: [PATCH 2/9] fix(world-model): auto-default calibration so objects exist without CAMERA_POSITIONS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase A2 of the 10x roadmap. Previously, with no CAMERA_POSITIONS configured, pixel_to_world() returned None for every camera, so the world model produced ZERO world objects and ZERO predictions β€” viewers saw only raw detections and per-camera tracks, and the "fusion" layer silently no-opped. - Add WorldModelRepositoryImpl._ensure_calibration(): lazily synthesize a default calibration (cameras spread along x-axis, height 2.5 m, focal 800) when a camera has none, with a one-time warning. Explicit CAMERA_POSITIONS still take precedence. - Wire it into _process_track and generate_predictions so both track fusion and view-only prediction cameras work out of the box. - Tests: 4 new (objects created without config; lazy creation for unseen camera; explicit positions win; predictions don't raise for a second camera). - Docs: backend/ARCHITECTURE.md no longer claims CAMERA_POSITIONS is required. Backend gate green: ruff + mypy clean, 61 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/ARCHITECTURE.md | 5 +- .../app/infrastructure/world_model_adapter.py | 53 ++++++++++++++++- .../unit/test_world_model_auto_calibration.py | 57 +++++++++++++++++++ 3 files changed, 112 insertions(+), 3 deletions(-) create mode 100644 backend/tests/unit/test_world_model_auto_calibration.py diff --git a/backend/ARCHITECTURE.md b/backend/ARCHITECTURE.md index 0a03f4a..e0476d1 100644 --- a/backend/ARCHITECTURE.md +++ b/backend/ARCHITECTURE.md @@ -99,10 +99,13 @@ CAM_0_URL=0 # Local camera device CAM_1_URL=http://192.168.1.100:8080/video ``` -**Camera Positions** (required for world coordinates): +**Camera Positions** (recommended for accurate world coordinates; auto-defaulted if omitted): ```env CAMERA_POSITIONS=[[0, 0, 2], [5, 0, 2], [0, 5, 2]] ``` +> If omitted, the world model synthesizes a default calibration per camera (spread along +> the x-axis) so it still produces world objects out of the box, logging a one-time warning. +> Set real positions for accurate cross-camera coordinates. ## Running diff --git a/backend/app/infrastructure/world_model_adapter.py b/backend/app/infrastructure/world_model_adapter.py index 6a8ea4f..a6a641a 100644 --- a/backend/app/infrastructure/world_model_adapter.py +++ b/backend/app/infrastructure/world_model_adapter.py @@ -245,7 +245,13 @@ def __init__(self, config_repo: ConfigurationRepository): self._person_height = config_repo.get_float("person_height_meters", 1.7) self._max_age = config_repo.get_float("world_object_max_age_seconds", 5.0) self._prediction_horizon = config_repo.get_float("prediction_horizon_seconds", 5.0) - + + # Default calibration (used when CAMERA_POSITIONS is not configured) + self._default_focal_length = 800.0 + self._default_camera_height = 2.5 + self._default_camera_spacing = 3.0 + self._warned_default_calibration = False + # Initialize default calibrations if positions provided self._init_default_calibrations() @@ -274,7 +280,43 @@ def _init_default_calibrations(self) -> None: ) self._transformer.set_calibration(calibration) logger.info(f"Camera {i} calibrated at position ({pos[0]}, {pos[1]}, {pos[2]})") - + + def _ensure_calibration(self, camera_id: int) -> None: + """Ensure ``camera_id`` has a calibration, synthesizing a default if not. + + Without ``CAMERA_POSITIONS`` the world model would otherwise produce no + world objects or predictions at all (``pixel_to_world`` returns ``None`` + for an uncalibrated camera). The default spreads cameras along the x-axis + so multi-camera setups stay distinct; set ``CAMERA_POSITIONS`` for accuracy. + """ + if camera_id in self._transformer._calibrations: + return + + if not self._warned_default_calibration: + logger.warning( + "No CAMERA_POSITIONS configured; using auto-default camera " + "calibration. World coordinates are approximate β€” set " + "CAMERA_POSITIONS for accuracy." + ) + self._warned_default_calibration = True + + position = Point3D( + camera_id * self._default_camera_spacing, 0.0, self._default_camera_height + ) + self._transformer.set_calibration( + CameraCalibration( + camera_id=camera_id, + position=position, + rotation=(0.0, 0.0, 0.0), + focal_length=self._default_focal_length, + image_center=(640.0, 360.0), + ) + ) + logger.info( + f"Camera {camera_id}: auto-default calibration at " + f"({position.x}, {position.y}, {position.z})" + ) + async def initialize(self) -> None: """Initialize the world model.""" logger.info("World model initialized") @@ -302,6 +344,10 @@ async def _process_track( timestamp: datetime ) -> None: """Process a single track update.""" + # Ensure the camera has a calibration (auto-default if unconfigured), + # otherwise pixel_to_world returns None and no world object is created. + self._ensure_calibration(camera_id) + # Estimate depth from bbox height bbox_height = track.bbox.height if bbox_height > 20: @@ -467,6 +513,9 @@ def get_world_objects(self) -> List[WorldObject]: def generate_predictions(self, camera_id: int) -> List[PredictedTarget]: """Generate predictions for a camera view.""" + # A view-only camera still needs a calibration to project world objects. + self._ensure_calibration(camera_id) + predictions = [] now = datetime.now() diff --git a/backend/tests/unit/test_world_model_auto_calibration.py b/backend/tests/unit/test_world_model_auto_calibration.py new file mode 100644 index 0000000..9df3db1 --- /dev/null +++ b/backend/tests/unit/test_world_model_auto_calibration.py @@ -0,0 +1,57 @@ +"""World model must produce objects even without CAMERA_POSITIONS configured. + +Regression: previously, with no CAMERA_POSITIONS, ``pixel_to_world`` returned +None for every camera, so the world model produced ZERO world objects and ZERO +predictions β€” viewers saw only raw detections/tracks. Auto-default calibration +fixes this so single-camera setups work out of the box. +""" +import asyncio +from unittest.mock import Mock + +import pytest + +pytest.importorskip("cv2") + + +def _make_repo(camera_positions=None): + from app.infrastructure.world_model_adapter import WorldModelRepositoryImpl + config = Mock() + config.get.return_value = {} + config.get_int.return_value = 4 + config.get_float.return_value = 1.7 + config.get_list.return_value = [] if camera_positions is None else camera_positions + return WorldModelRepositoryImpl(config) + + +def test_world_objects_created_without_camera_positions(make_track): + repo = _make_repo(camera_positions=[]) + track = make_track(track_id=1, camera_id=0) + objects = asyncio.run(repo.update({0: [track]})) + assert len(objects) == 1 + assert objects[0].class_name == "person" + + +def test_auto_default_calibration_created_lazily_for_unseen_camera(make_track): + repo = _make_repo(camera_positions=[]) + assert repo._transformer._calibrations.get(2) is None + track = make_track(track_id=1, camera_id=2) + asyncio.run(repo.update({2: [track]})) + assert repo._transformer._calibrations.get(2) is not None + + +def test_explicit_camera_positions_take_precedence(make_track): + repo = _make_repo(camera_positions=[[5.0, 0.0, 2.0]]) + calib = repo._transformer._calibrations.get(0) + assert calib is not None + assert (calib.position.x, calib.position.y, calib.position.z) == (5.0, 0.0, 2.0) + + +def test_predictions_available_for_second_camera_without_config(make_track): + """With two cameras and no config, an object seen only by cam 0 should yield + a world-projection prediction for cam 1.""" + repo = _make_repo(camera_positions=[]) + asyncio.run(repo.update({0: [make_track(track_id=1, camera_id=0)]})) + # cam 1 has a default calibration created on demand; ask for its predictions + repo._ensure_calibration(1) + preds = repo.generate_predictions(camera_id=1) + assert isinstance(preds, list) # may be empty if behind camera, but must not raise From b15212c9ef7060e6a9ec4c2d005607c9d8d1900e Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 19:05:45 +0530 Subject: [PATCH 3/9] refactor(tracking): associate by track id, not positional index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase A3 of the 10x roadmap. CameraTracker.update() previously mapped Hungarian row indices back to tracks via `list(self.tracks.keys())[idx]` inside its loops β€” O(n^2) and implicitly coupled to dict iteration order matching the cost-matrix build order. Correct today, but fragile. - _associate_detections now returns (matched[(track_id, det_idx)], unmatched_track_ids, unmatched_det_indices); update() consumes track ids directly. Behaviour-preserving; removes the ordering coupling and the per-match key-list rebuilds. - Add 6 characterization tests for the CameraTracker lifecycle (create β†’ confirm β†’ coast β†’ remove), including multi-track id-mapping correctness. Coverage: tracking_adapter 46%β†’85%, total 42%β†’49%. ruff/mypy clean, 67 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../app/infrastructure/tracking_adapter.py | 87 ++++++++-------- backend/tests/unit/test_camera_tracker.py | 99 +++++++++++++++++++ 2 files changed, 146 insertions(+), 40 deletions(-) create mode 100644 backend/tests/unit/test_camera_tracker.py diff --git a/backend/app/infrastructure/tracking_adapter.py b/backend/app/infrastructure/tracking_adapter.py index 1f2c20c..f051840 100644 --- a/backend/app/infrastructure/tracking_adapter.py +++ b/backend/app/infrastructure/tracking_adapter.py @@ -128,29 +128,26 @@ def update(self, detections: List[Detection]) -> List[Track]: """Update tracks with new detections.""" # Predict existing tracks self._predict_tracks() - - # Associate detections with tracks - matched_tracks, matched_dets, unmatched_tracks, unmatched_dets = \ + + # Associate detections with tracks (results are keyed by track id) + matched, unmatched_track_ids, unmatched_det_indices = \ self._associate_detections(detections) - + # Update matched tracks - for track_idx, det_idx in zip(matched_tracks, matched_dets): - track_id = list(self.tracks.keys())[track_idx] - detection = detections[det_idx] - self._update_track(track_id, detection) - + for track_id, det_idx in matched: + self._update_track(track_id, detections[det_idx]) + # Mark unmatched tracks as coasting - for track_idx in unmatched_tracks: - track_id = list(self.tracks.keys())[track_idx] + for track_id in unmatched_track_ids: self._coast_track(track_id) - + # Create new tracks for unmatched detections - for det_idx in unmatched_dets: + for det_idx in unmatched_det_indices: self._create_track(detections[det_idx]) - + # Remove old tracks self._remove_old_tracks() - + return list(self.tracks.values()) def _predict_tracks(self) -> None: @@ -179,40 +176,50 @@ def _predict_tracks(self) -> None: def _associate_detections( self, detections: List[Detection] - ) -> Tuple[List[int], List[int], List[int], List[int]]: - """Associate detections with existing tracks.""" - if not self.tracks or not detections: - return [], [], list(range(len(self.tracks))), list(range(len(detections))) - - track_list = list(self.tracks.values()) - - # Compute cost matrix + ) -> Tuple[List[Tuple[int, int]], List[int], List[int]]: + """Associate detections with existing tracks. + + Returns ``(matched, unmatched_track_ids, unmatched_det_indices)`` where + ``matched`` is a list of ``(track_id, detection_index)`` pairs. Resolving + the Hungarian row indices to track *ids* here (rather than handing + positional indices back to the caller) keeps association robust to dict + ordering and avoids rebuilding ``list(self.tracks.keys())`` per match. + """ + track_ids = list(self.tracks.keys()) + if not track_ids or not detections: + return [], list(track_ids), list(range(len(detections))) + + track_list = [self.tracks[tid] for tid in track_ids] + + # Compute cost matrix (rows align with track_ids order) cost_matrix = compute_cost_matrix( track_list, detections, appearance_weight=self.appearance_weight ) - + # Run Hungarian algorithm row_ind, col_ind = hungarian_assignment(cost_matrix) - + # Filter by IoU threshold - matched_tracks = [] - matched_dets = [] - + matched: List[Tuple[int, int]] = [] + matched_rows: set[int] = set() + matched_cols: set[int] = set() + for r, c in zip(row_ind, col_ind): - track = track_list[r] - det = detections[c] - iou = compute_iou(track.bbox, det.bbox) - + iou = compute_iou(track_list[r].bbox, detections[c].bbox) if iou >= self.iou_threshold: - matched_tracks.append(r) - matched_dets.append(c) - - # Find unmatched - unmatched_tracks = [i for i in range(len(track_list)) if i not in matched_tracks] - unmatched_dets = [i for i in range(len(detections)) if i not in matched_dets] - - return matched_tracks, matched_dets, unmatched_tracks, unmatched_dets + matched.append((track_ids[r], c)) + matched_rows.add(r) + matched_cols.add(c) + + unmatched_track_ids = [ + track_ids[i] for i in range(len(track_ids)) if i not in matched_rows + ] + unmatched_det_indices = [ + i for i in range(len(detections)) if i not in matched_cols + ] + + return matched, unmatched_track_ids, unmatched_det_indices def _update_track(self, track_id: int, detection: Detection) -> None: """Update a track with a matched detection.""" diff --git a/backend/tests/unit/test_camera_tracker.py b/backend/tests/unit/test_camera_tracker.py new file mode 100644 index 0000000..9e5f1a1 --- /dev/null +++ b/backend/tests/unit/test_camera_tracker.py @@ -0,0 +1,99 @@ +"""Characterization + robustness tests for CameraTracker.update lifecycle. + +These pin the per-camera tracker behaviour (create β†’ confirm β†’ coast β†’ remove) +and, importantly, that the matched/unmatched mapping resolves to the *correct +track ids* when multiple tracks exist β€” guarding the association refactor that +removes the positional-index β†’ dict-ordering coupling. +""" +from datetime import datetime + +from app.domain.entities import BoundingBox, Detection, TrackingState +from app.infrastructure.tracking_adapter import CameraTracker + + +def _tracker(min_hits: int = 3, max_age: int = 3, iou_threshold: float = 0.25) -> CameraTracker: + return CameraTracker( + camera_id=0, + max_age=max_age, + min_hits=min_hits, + iou_threshold=iou_threshold, + appearance_weight=0.4, + ) + + +def _det(x: float, det_id: str = "d") -> Detection: + return Detection( + detection_id=det_id, + camera_id=0, + bbox=BoundingBox(x, 0.0, x + 100.0, 200.0), + confidence=0.9, + class_id=0, + class_name="person", + timestamp=datetime.now(), + ) + + +def test_new_detection_creates_tentative_track(): + t = _tracker() + tracks = t.update([_det(0)]) + assert len(tracks) == 1 + assert tracks[0].state == TrackingState.TENTATIVE + assert tracks[0].track_id == 1 + + +def test_repeated_match_confirms_and_keeps_same_id(): + t = _tracker(min_hits=3) + for _ in range(3): + t.update([_det(0)]) + tracks = list(t.tracks.values()) + assert len(tracks) == 1 + assert tracks[0].track_id == 1 + assert tracks[0].state == TrackingState.CONFIRMED + + +def test_unmatched_confirmed_track_coasts(): + t = _tracker(min_hits=2) + t.update([_det(0)]) + t.update([_det(0)]) # confirmed now + assert t.tracks[1].state == TrackingState.CONFIRMED + conf_before = t.tracks[1].confidence + # Next frame: a detection far away β€” does not match track 1 + t.update([_det(1000)]) + assert t.tracks[1].state == TrackingState.COASTING + assert t.tracks[1].confidence < conf_before + + +def test_multi_track_mapping_resolves_correct_ids(): + """Two confirmed tracks; a single detection matches only the first. + The matched/unmatched split must map to the right track ids.""" + t = _tracker(min_hits=2, max_age=10) + # Confirm two well-separated tracks + for _ in range(2): + t.update([_det(0, "a"), _det(500, "b")]) + assert {tid: trk.state for tid, trk in t.tracks.items()} == { + 1: TrackingState.CONFIRMED, + 2: TrackingState.CONFIRMED, + } + # Frame with only the left detection: track 1 should update, track 2 coast + t.update([_det(0, "a")]) + assert t.tracks[1].state == TrackingState.CONFIRMED + assert t.tracks[1].time_since_update == 0 + assert t.tracks[2].state == TrackingState.COASTING + assert t.tracks[2].time_since_update >= 1 + + +def test_stale_track_is_removed_after_max_age(): + t = _tracker(min_hits=1, max_age=2) + t.update([_det(0)]) # track 1 created + confirmed (min_hits=1) + assert 1 in t.tracks + # No matching detections for several frames -> time_since_update grows + for _ in range(4): + t.update([_det(1000, "far")]) + assert 1 not in t.tracks # removed once time_since_update > max_age + + +def test_two_detections_create_two_distinct_tracks(): + t = _tracker() + tracks = t.update([_det(0, "a"), _det(500, "b")]) + assert len(tracks) == 2 + assert {trk.track_id for trk in tracks} == {1, 2} From 22e5dc7d2eb4ef020163e6339c6b7abde02f05cf Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 19:09:05 +0530 Subject: [PATCH 4/9] test(ws): cover broadcast serialization; raise coverage floor to 50% MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase A4 of the 10x roadmap β€” completes Phase A (the honest & solid base). - Add msgpack round-trip tests for WebSocketCommunicationRepository._serialize_* (snapshot envelope, JPEG frame encoding, detection/track/world-object/prediction shapes, empty snapshot). This pins the exact wire format the frontend decodes. - CI: install fastapi + msgpack (needed to import the ws adapter) and raise the coverage floor 40% -> 50%. websocket_adapter 0%->46%, total 49%->52%. ruff/mypy clean, 74 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- .github/workflows/ci.yml | 3 +- .../unit/test_websocket_serialization.py | 122 ++++++++++++++++++ 2 files changed, 124 insertions(+), 1 deletion(-) create mode 100644 backend/tests/unit/test_websocket_serialization.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7f74dc5..7f7f36f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,6 +20,7 @@ jobs: python -m pip install --upgrade pip pip install pytest pytest-cov ruff mypy pip install numpy pydantic pydantic-settings opencv-python-headless + pip install fastapi msgpack - name: Lint (ruff) run: python -m ruff check backend scripts - name: Type-check (mypy) @@ -27,7 +28,7 @@ jobs: - name: Unit tests + coverage run: | python -m pytest backend/tests/unit -q \ - --cov=app --cov-report=term-missing --cov-fail-under=40 + --cov=app --cov-report=term-missing --cov-fail-under=50 env: PYTHONPATH: backend diff --git a/backend/tests/unit/test_websocket_serialization.py b/backend/tests/unit/test_websocket_serialization.py new file mode 100644 index 0000000..bb05ced --- /dev/null +++ b/backend/tests/unit/test_websocket_serialization.py @@ -0,0 +1,122 @@ +"""Serialization tests for the WebSocket broadcast payload. + +The pipeline serializes each PerceptionSnapshot to msgpack once and broadcasts the +bytes to every viewer; the frontend decodes this exact shape. These tests pin the +wire format via a msgpack round-trip. +""" +from datetime import datetime + +import numpy as np +import pytest + +pytest.importorskip("fastapi") +pytest.importorskip("msgpack") +pytest.importorskip("cv2") + +import msgpack + +from app.domain.entities import ( + BoundingBox, CameraFrame, Detection, PerceptionSnapshot, Point3D, + PredictedTarget, PredictionMethod, Track, TrackingState, Velocity3D, + WorldObject, +) +from app.infrastructure.websocket_adapter import WebSocketCommunicationRepository + + +def _snapshot() -> PerceptionSnapshot: + ts = datetime.now() + frame = CameraFrame( + camera_id=0, + frame_data=np.zeros((4, 4, 3), dtype=np.uint8), + timestamp=ts, + frame_number=1, + ) + det = Detection( + detection_id="det_1", camera_id=0, + bbox=BoundingBox(10, 20, 110, 220), confidence=0.9, + class_id=0, class_name="person", timestamp=ts, + ) + trk = Track( + track_id=7, camera_id=0, bbox=BoundingBox(10, 20, 110, 220), + confidence=0.8, class_id=0, class_name="person", + state=TrackingState.CONFIRMED, age=5, hits=4, velocity=(1.0, -2.0), + ) + obj = WorldObject( + object_id=3, position=Point3D(1.0, 2.0, 3.0), + velocity=Velocity3D(0.1, 0.2, 0.3), class_id=0, class_name="person", + confidence=0.7, last_seen_camera=0, last_update=ts, + ) + pred = PredictedTarget( + object_id=3, camera_id=1, predicted_bbox=BoundingBox(5, 5, 55, 105), + confidence=0.6, time_since_seen=1.2, velocity_projection=(0.5, 0.5), + source_camera=0, prediction_method=PredictionMethod.WORLD_PROJECTION, + ) + return PerceptionSnapshot( + timestamp=ts, generation=42, + world_objects=[obj], + camera_frames={0: frame}, + detections={0: [det]}, + tracks={0: [trk]}, + predictions={1: [pred]}, + ) + + +def _roundtrip(snapshot: PerceptionSnapshot) -> dict: + repo = WebSocketCommunicationRepository(max_clients=10) + payload = repo._serialize_snapshot(snapshot) + assert isinstance(payload, (bytes, bytearray)) + return msgpack.unpackb(payload, raw=False) + + +def test_snapshot_envelope_shape(): + msg = _roundtrip(_snapshot()) + assert msg["type"] == "snapshot" + assert msg["generation"] == 42 + assert set(msg) >= { + "type", "timestamp", "generation", "camera_frames", + "world_objects", "detections", "tracks", "predictions", "metrics", + } + + +def test_camera_frame_encoded_to_jpeg_bytes(): + msg = _roundtrip(_snapshot()) + frame_bytes = msg["camera_frames"]["0"] + assert isinstance(frame_bytes, (bytes, bytearray)) + assert bytes(frame_bytes[:2]) == b"\xff\xd8" # JPEG SOI marker + + +def test_detection_serialization_shape(): + det = _roundtrip(_snapshot())["detections"]["0"][0] + assert det["detection_id"] == "det_1" + assert det["bbox"] == [10, 20, 110, 220] + assert det["class_name"] == "person" + assert det["center"] == [60.0, 120.0] + + +def test_track_serialization_shape(): + trk = _roundtrip(_snapshot())["tracks"]["0"][0] + assert trk["track_id"] == 7 + assert trk["state"] == "CONFIRMED" + assert list(trk["velocity"]) == [1.0, -2.0] + + +def test_world_object_serialization_shape(): + obj = _roundtrip(_snapshot())["world_objects"][0] + assert obj["object_id"] == 3 + assert obj["position"] == {"x": 1.0, "y": 2.0, "z": 3.0} + assert obj["last_seen_camera"] == 0 + + +def test_prediction_serialization_method_value(): + pred = _roundtrip(_snapshot())["predictions"]["1"][0] + assert pred["object_id"] == 3 + assert pred["method"] == "world_projection" + assert pred["source_camera"] == 0 + + +def test_empty_snapshot_serializes_cleanly(): + empty = PerceptionSnapshot(timestamp=datetime.now(), generation=0) + msg = _roundtrip(empty) + assert msg["world_objects"] == [] + assert msg["camera_frames"] == {} + assert msg["detections"] == {} From 47e0338d19ee2965e178f3b608a65f92fa61fc96 Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 19:40:37 +0530 Subject: [PATCH 5/9] feat(reid): wire appearance descriptors into detection + cross-camera matching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase B1 β€” the first feature build. Activates appearance re-ID, which was scaffolded but dormant (compute_appearance was never called; Detection.appearance was always None, so tracking was pure-IoU and cross-camera matching distance-only). - detection: extract module-level compute_hsv_appearance() and populate Detection.appearance per detection (64-dim L2-normalized HSV histogram; toggle via APPEARANCE_REID_ENABLED). This also activates the appearance term already present in the tracking cost matrix. - world model: gate cross-camera association on appearance cosine similarity (CROSS_CAMERA_APPEARANCE_THRESHOLD, default 0.5) so differently-dressed people at the same ground position stay separate; EMA-smooth (alpha=0.3) the fused descriptor. - config: add appearance_reid_enabled + cross_camera_appearance_threshold as real Settings fields (env-wired) + .env.example docs. - README: flip "Appearance re-ID" from planned to implemented. - Tests: 7 new (descriptor shape/similarity/degenerate; appearance-gated matching; distance fallback; config wiring). Coverage 52%->55%. ruff/mypy clean, 81 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 18 +-- backend/.env.example | 6 + backend/app/infrastructure/config_adapter.py | 6 +- .../app/infrastructure/detection_adapter.py | 69 +++++++----- .../app/infrastructure/world_model_adapter.py | 71 ++++++++++-- backend/tests/unit/test_appearance_reid.py | 103 ++++++++++++++++++ 6 files changed, 221 insertions(+), 52 deletions(-) create mode 100644 backend/tests/unit/test_appearance_reid.py diff --git a/README.md b/README.md index fb9c78b..f687731 100644 --- a/README.md +++ b/README.md @@ -102,7 +102,7 @@ that describe them are forward-looking design. | Optional JWT auth Β· SSL Β· atomic Jetson deploy | βœ… Implemented | | Cross-camera homography ghosts (green `H-PROJ`) | πŸ”­ Planned | | Pixel-extrapolation ghosts (red `EXTRAP`) | πŸ”­ Planned | -| Appearance re-ID (HSV histograms) wired into tracking / fusion | πŸ”­ Planned | +| Appearance re-ID (HSV histograms) wired into tracking / fusion | βœ… Implemented | | Sensor-trust scoring | πŸ”­ Planned | | Adaptive Kalman noise by bbox area | πŸ”­ Planned | | GPS + IMU fusion into the world model | πŸ”­ Planned | @@ -122,10 +122,10 @@ that describe them are forward-looking design. |---|---| | **Person detection** | YOLOv8n with NMS-level class filter (`classes=[0]`) β€” person-only | | **TensorRT FP16** | `.engine` export on Jetson β€” ~8 MiB, sub-10 ms inference | -| **Hungarian tracking** | `scipy.optimize.linear_sum_assignment` β€” cost `0.6 Γ— IoU + 0.4 Γ— cosine appearance` (appearance term is inactive until re-ID lands, so tracking is effectively pure-IoU today) | +| **Hungarian tracking** | `scipy.optimize.linear_sum_assignment` β€” cost `0.6 Γ— IoU + 0.4 Γ— cosine appearance`; appearance descriptors are now computed per detection, so the appearance term is active | | **Tracker fallback chain** πŸ”­ | *Planned* β€” only Hungarian (with a greedy fallback) is active today; DeepSORT/Centroid are not implemented | | **Adaptive Kalman filter** | 6-state `[x, y, z, vx, vy, vz]` β€” measurement noise scales by **confidence** (bbox-area & sensor-trust scaling πŸ”­ planned) | -| **Cross-camera re-ID** πŸ”­ | *Planned* β€” 64-dim HSV histogram descriptors (`compute_appearance()` exists but isn't yet wired into detection/tracking) | +| **Cross-camera re-ID** | 64-dim HSV histogram descriptors, L2-normalized, EMA-smoothed (Ξ± = 0.3); computed per detection and used to gate cross-camera association | | **Sensor trust scoring** πŸ”­ | *Planned* β€” per-sensor trust ∈ [0.1, 1.0]; the Kalman update accepts the param but it is fixed at 1.0 today | | **Cross-camera homography** πŸ”­ | *Planned* β€” self-calibrating ground-plane H via `cv2.findHomography` + RANSAC (no homography code in the pipeline yet) | | **Ghost predictions** | Path C β€” world-coordinate pinhole projection (orange `WORLD`) is **active**. Path A (homography/green `H-PROJ`) and Path B (pixel extrapolation/red `EXTRAP`) are πŸ”­ planned | @@ -486,7 +486,7 @@ Each fused world object maintains a 6-state Kalman filter `[x, y, z, vx, vy, vz] Objects from different cameras are matched when: - Euclidean distance < 2 m - Same `class_id` -- πŸ”­ *Planned:* appearance cosine similarity > 0.5 (today the match is distance + class only; the appearance gate is not yet wired in) +- Appearance cosine similarity β‰₯ 0.5 when both observations carry a descriptor (differently-dressed people at the same ground position stay separate) ### Sensor trust πŸ”­ *(planned)* @@ -496,14 +496,14 @@ The intended design earns per-sensor trust through consistency: Today the Kalman update accepts a `sensor_trust` argument but it is fixed at `1.0`. -### Appearance re-ID πŸ”­ *(planned)* +### Appearance re-ID -- 64-dimensional HSV histogram descriptors (`compute_appearance()` is implemented…) +- 64-dimensional HSV histogram descriptors computed per detection (~0.1 ms each) - L2-normalized for cosine similarity -- Exponential moving average (Ξ± = 0.3) for descriptor stability across frames +- Exponential moving average (Ξ± = 0.3) on the fused world object for descriptor stability -…but `compute_appearance()` is **not yet called** in the pipeline, so `Detection.appearance` -is always `None` and re-ID is dormant until Phase B. +Used to gate cross-camera association: two people at the same ground position but with +different appearance are kept as distinct world objects. --- diff --git a/backend/.env.example b/backend/.env.example index aa9e488..258cfb4 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -79,6 +79,12 @@ PERSON_HEIGHT_METERS=1.7 PREDICTION_HORIZON_SECONDS=5.0 WORLD_OBJECT_MAX_AGE_SECONDS=5.0 +# --- Cross-camera appearance re-ID --- +# Compute HSV histogram descriptors per detection and gate cross-camera matching. +APPEARANCE_REID_ENABLED=true +# Min cosine similarity [0..1] to merge two cameras' observations into one object. +CROSS_CAMERA_APPEARANCE_THRESHOLD=0.5 + # --- Homography Configuration --- HOMOGRAPHY_MIN_PAIRS=4 HOMOGRAPHY_MAX_PAIRS=100 diff --git a/backend/app/infrastructure/config_adapter.py b/backend/app/infrastructure/config_adapter.py index eace532..145e6d9 100644 --- a/backend/app/infrastructure/config_adapter.py +++ b/backend/app/infrastructure/config_adapter.py @@ -99,7 +99,11 @@ class Settings(BaseSettings): person_height_meters: float = Field(default=1.7, ge=0.5, le=3.0) prediction_horizon_seconds: float = Field(default=5.0, ge=1.0, le=30.0) world_object_max_age_seconds: float = Field(default=5.0, ge=1.0, le=60.0) - + + # Cross-camera appearance re-ID + appearance_reid_enabled: bool = Field(default=True) + cross_camera_appearance_threshold: float = Field(default=0.5, ge=0.0, le=1.0) + # Homography settings homography_min_pairs: int = Field(default=4, ge=3) homography_max_pairs: int = Field(default=100, ge=10) diff --git a/backend/app/infrastructure/detection_adapter.py b/backend/app/infrastructure/detection_adapter.py index e641474..819203c 100644 --- a/backend/app/infrastructure/detection_adapter.py +++ b/backend/app/infrastructure/detection_adapter.py @@ -23,6 +23,34 @@ logger = logging.getLogger(__name__) +def compute_hsv_appearance( + frame: NDArray[np.uint8], bbox: BoundingBox +) -> Optional[AppearanceDescriptor]: + """Compute a 64-dim L2-normalized HSV histogram descriptor for a bbox crop. + + 32 hue + 16 saturation + 16 value bins. Returns ``None`` for an out-of-bounds + or degenerate crop. ~0.1 ms per call. Used for cross-camera re-identification. + """ + try: + x1, y1, x2, y2 = int(bbox.x1), int(bbox.y1), int(bbox.x2), int(bbox.y2) + h, w = frame.shape[:2] + x1, y1 = max(0, x1), max(0, y1) + x2, y2 = min(w, x2), min(h, y2) + if x2 <= x1 or y2 <= y1: + return None + crop = frame[y1:y2, x1:x2] + hsv = cv2.cvtColor(crop, cv2.COLOR_BGR2HSV) + hist_h = cv2.calcHist([hsv], [0], None, [32], [0, 180]).flatten() + hist_s = cv2.calcHist([hsv], [1], None, [16], [0, 256]).flatten() + hist_v = cv2.calcHist([hsv], [2], None, [16], [0, 256]).flatten() + feature = np.concatenate([hist_h, hist_s, hist_v]) + norm = float(np.linalg.norm(feature)) + 1e-6 + return AppearanceDescriptor(vector=(feature / norm).astype(np.float32)) + except Exception as e: + logger.error(f"Appearance computation error: {e}") + return None + + class YOLODetector: """YOLOv8 detector with proper error handling.""" @@ -36,6 +64,7 @@ def __init__(self, config_repo: ConfigurationRepository): self._iou_threshold = config_repo.get_float("iou_threshold", 0.45) self._max_detections = config_repo.get_int("max_detections", 100) self._detection_classes = config_repo.get_list("detection_classes", [0]) + self._appearance_enabled = config_repo.get_bool("appearance_reid_enabled", True) self._is_initialized = False # Try to import ultralytics @@ -167,6 +196,12 @@ def detect(self, frame: NDArray[np.uint8]) -> List[Detection]: confidence=float(person_kp[j, 2]) )) + # Cross-camera re-ID descriptor (HSV histogram of the crop) + appearance = ( + compute_hsv_appearance(frame, bbox) + if self._appearance_enabled else None + ) + detection = Detection( detection_id=f"det_{timestamp.timestamp()}_{idx}_{id(boxes)}", camera_id=-1, # Will be set by caller @@ -175,7 +210,8 @@ def detect(self, frame: NDArray[np.uint8]) -> List[Detection]: class_id=int(cls_id), class_name=self._model.names.get(cls_id, 'unknown'), timestamp=timestamp, - keypoints=keypoints + keypoints=keypoints, + appearance=appearance, ) detections.append(detection) @@ -251,32 +287,5 @@ async def compute_appearance( frame: NDArray[np.uint8], bbox: BoundingBox ) -> Optional[AppearanceDescriptor]: - """Compute HSV histogram appearance descriptor.""" - try: - x1, y1, x2, y2 = int(bbox.x1), int(bbox.y1), int(bbox.x2), int(bbox.y2) - h, w = frame.shape[:2] - - # Clamp to frame bounds - x1, y1 = max(0, x1), max(0, y1) - x2, y2 = min(w, x2), min(h, y2) - - if x2 <= x1 or y2 <= y1: - return None - - crop = frame[y1:y2, x1:x2] - hsv = cv2.cvtColor(crop, cv2.COLOR_BGR2HSV) - - # 32 hue + 16 sat + 16 val = 64-dim histogram - hist_h = cv2.calcHist([hsv], [0], None, [32], [0, 180]).flatten() - hist_s = cv2.calcHist([hsv], [1], None, [16], [0, 256]).flatten() - hist_v = cv2.calcHist([hsv], [2], None, [16], [0, 256]).flatten() - - feature = np.concatenate([hist_h, hist_s, hist_v]) - norm = np.linalg.norm(feature) + 1e-6 - normalized = (feature / norm).astype(np.float32) - - return AppearanceDescriptor(vector=normalized) - - except Exception as e: - logger.error(f"Appearance computation error: {e}") - return None + """Compute an HSV histogram appearance descriptor (module-level helper).""" + return compute_hsv_appearance(frame, bbox) diff --git a/backend/app/infrastructure/world_model_adapter.py b/backend/app/infrastructure/world_model_adapter.py index a6a641a..33a782b 100644 --- a/backend/app/infrastructure/world_model_adapter.py +++ b/backend/app/infrastructure/world_model_adapter.py @@ -14,7 +14,7 @@ from app.application.ports import WorldModelRepository, ConfigurationRepository from app.domain.entities import ( Track, WorldObject, PredictedTarget, CameraCalibration, - Point3D, Velocity3D, BoundingBox, PredictionMethod + Point3D, Velocity3D, BoundingBox, PredictionMethod, AppearanceDescriptor ) @@ -252,6 +252,12 @@ def __init__(self, config_repo: ConfigurationRepository): self._default_camera_spacing = 3.0 self._warned_default_calibration = False + # Cross-camera appearance re-ID + self._appearance_match_threshold = config_repo.get_float( + "cross_camera_appearance_threshold", 0.5 + ) + self._appearance_ema_alpha = 0.3 + # Initialize default calibrations if positions provided self._init_default_calibrations() @@ -378,8 +384,10 @@ async def _process_track( object_id = self._track_to_object[track_key] self._update_existing_object(object_id, track, world_pos, timestamp) else: - # Check for nearby objects (cross-camera matching) - existing_id = self._find_matching_object(world_pos, track.class_id) + # Check for nearby objects (cross-camera matching, appearance-gated) + existing_id = self._find_matching_object( + world_pos, track.class_id, track.appearance + ) if existing_id: self._track_to_object[track_key] = existing_id self._update_existing_object(existing_id, track, world_pos, timestamp) @@ -429,6 +437,9 @@ def _update_existing_object( obj.camera_pixel_positions[track.camera_id] = track.bbox.center obj.camera_pixel_velocities[track.camera_id] = track.velocity obj.camera_last_seen[track.camera_id] = timestamp + + # EMA-smooth the appearance descriptor for re-ID stability + obj.appearance = self._blend_appearance(obj.appearance, track.appearance) def _create_new_object( self, @@ -450,6 +461,7 @@ def _create_new_object( confidence=track.confidence, last_seen_camera=track.camera_id, last_update=timestamp, + appearance=track.appearance, source_tracks={track.camera_id: track.track_id}, camera_pixel_positions={track.camera_id: track.bbox.center}, camera_pixel_velocities={track.camera_id: track.velocity}, @@ -465,25 +477,60 @@ def _create_new_object( logger.debug(f"Created new world object {object_id}") + def _blend_appearance( + self, + old: Optional[AppearanceDescriptor], + new: Optional[AppearanceDescriptor], + ) -> Optional[AppearanceDescriptor]: + """EMA-blend two appearance descriptors (alpha weights the new observation).""" + if new is None: + return old + if old is None: + return new + a = self._appearance_ema_alpha + blended = (1.0 - a) * old.vector + a * new.vector + norm = float(np.linalg.norm(blended)) + 1e-6 + return AppearanceDescriptor(vector=(blended / norm).astype(np.float32)) + def _find_matching_object( self, world_pos: Point3D, - class_id: int + class_id: int, + appearance: Optional[AppearanceDescriptor] = None, ) -> Optional[int]: - """Find existing object that matches position.""" + """Find an existing world object matching this observation. + + Candidates must share the class and lie within ``distance_threshold`` metres. + When both the candidate and the observation carry an appearance descriptor, + a cosine similarity below ``_appearance_match_threshold`` rejects the match β€” + so two differently-dressed people at the same spot stay separate. With no + appearance available it falls back to nearest-within-threshold. + """ + distance_threshold = 2.0 # metres best_match = None - min_distance = float('inf') - threshold = 2.0 # meters - + best_score = -1.0 + for obj_id, obj in self._world_objects.items(): if obj.class_id != class_id: continue - + distance = obj.position.distance_to(world_pos) - if distance < min_distance and distance < threshold: - min_distance = distance + if distance >= distance_threshold: + continue + + if appearance is not None and obj.appearance is not None: + similarity = obj.appearance.cosine_similarity(appearance) + if similarity < self._appearance_match_threshold: + continue + score = similarity + else: + # No appearance to compare β€” prefer the closest candidate. + score = 1.0 - (distance / distance_threshold) + + if score > best_score: + best_score = score best_match = obj_id - + return best_match def _cleanup_old_objects(self, current_time: datetime) -> None: diff --git a/backend/tests/unit/test_appearance_reid.py b/backend/tests/unit/test_appearance_reid.py new file mode 100644 index 0000000..48122af --- /dev/null +++ b/backend/tests/unit/test_appearance_reid.py @@ -0,0 +1,103 @@ +"""Phase B1 β€” appearance re-ID: HSV descriptors + cross-camera appearance gating.""" +from datetime import datetime +from unittest.mock import Mock + +import numpy as np +import pytest + +pytest.importorskip("cv2") + +from app.domain.entities import ( + AppearanceDescriptor, BoundingBox, Point3D, Velocity3D, WorldObject, +) + + +def _solid(b: int, g: int, r: int, h: int = 40, w: int = 20) -> np.ndarray: + img = np.zeros((h, w, 3), dtype=np.uint8) + img[:, :] = (b, g, r) + return img + + +# ---------------------------------------------------------------- descriptor + +def test_descriptor_is_64d_and_l2_normalized(): + from app.infrastructure.detection_adapter import compute_hsv_appearance + d = compute_hsv_appearance(_solid(0, 0, 255), BoundingBox(0, 0, 20, 40)) + assert d is not None + assert d.vector.shape == (64,) + assert float(np.linalg.norm(d.vector)) == pytest.approx(1.0, abs=1e-5) + + +def test_same_color_more_similar_than_different_color(): + from app.infrastructure.detection_adapter import compute_hsv_appearance + red = compute_hsv_appearance(_solid(0, 0, 255), BoundingBox(0, 0, 20, 40)) + red2 = compute_hsv_appearance(_solid(0, 0, 255), BoundingBox(0, 0, 20, 40)) + blue = compute_hsv_appearance(_solid(255, 0, 0), BoundingBox(0, 0, 20, 40)) + assert red.cosine_similarity(red2) == pytest.approx(1.0, abs=1e-3) + assert red.cosine_similarity(blue) < red.cosine_similarity(red2) + + +def test_degenerate_bbox_returns_none(): + from app.infrastructure.detection_adapter import compute_hsv_appearance + assert compute_hsv_appearance(_solid(0, 0, 255), BoundingBox(0, 0, 0.4, 0.4)) is None + + +# ---------------------------------------------------- cross-camera matching + +def _repo(): + from app.infrastructure.world_model_adapter import WorldModelRepositoryImpl + config = Mock() + config.get.return_value = {} + config.get_int.return_value = 4 + config.get_float.return_value = 0.5 + config.get_list.return_value = [] + repo = WorldModelRepositoryImpl(config) + repo._appearance_match_threshold = 0.5 + return repo + + +def _descriptor(vec): + v = np.array(vec, dtype=np.float32) + return AppearanceDescriptor(vector=v / (np.linalg.norm(v) + 1e-6)) + + +def _insert(repo, obj_id, pos, appearance): + repo._world_objects[obj_id] = WorldObject( + object_id=obj_id, position=pos, velocity=Velocity3D(0, 0, 0), + class_id=0, class_name="person", confidence=0.9, + last_seen_camera=0, last_update=datetime.now(), appearance=appearance, + ) + + +def test_similar_appearance_within_distance_matches(): + repo = _repo() + a = _descriptor([1, 0, 0, 0]) + _insert(repo, 1, Point3D(0, 0, 0), a) + assert repo._find_matching_object(Point3D(0.5, 0, 0), 0, a) == 1 + + +def test_dissimilar_appearance_does_not_match(): + """Two people at (nearly) the same spot but different clothing stay separate.""" + repo = _repo() + _insert(repo, 1, Point3D(0, 0, 0), _descriptor([1, 0, 0, 0])) + orthogonal = _descriptor([0, 1, 0, 0]) # cosine 0 < 0.5 threshold + assert repo._find_matching_object(Point3D(0.3, 0, 0), 0, orthogonal) is None + + +def test_no_appearance_falls_back_to_distance(): + repo = _repo() + _insert(repo, 1, Point3D(0, 0, 0), None) + assert repo._find_matching_object(Point3D(0.5, 0, 0), 0, None) == 1 + assert repo._find_matching_object(Point3D(5, 0, 0), 0, None) is None + + +def test_reid_config_fields_are_wired(): + """Guard against doc drift: the new keys must actually flow through Settings.""" + from app.infrastructure.config_adapter import ( + PydanticConfigurationRepository, Settings, + ) + repo = PydanticConfigurationRepository( + Settings(cross_camera_appearance_threshold=0.7, appearance_reid_enabled=False) + ) + assert repo.get_float("cross_camera_appearance_threshold", 0.5) == 0.7 + assert repo.get_bool("appearance_reid_enabled", True) is False From 454ebb23b54182f499855fc9a12b9c101627fa9b Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 19:49:13 +0530 Subject: [PATCH 6/9] feat(homography): cross-camera H-PROJ ghost predictions (Path A) Phase B2. Implements the headline cross-camera homography feature: when one camera sees a person and another doesn't, project the foot point between views to render a green H-PROJ ghost. - New app/infrastructure/homography.py: HomographyEstimator accumulates foot-point correspondences per camera pair, estimates H via cv2.findHomography + RANSAC once >= min_pairs, re-estimates periodically, projects points, caps the pair buffer. - World model integration: - store per-camera foot points on WorldObject (camera_foot_points) - collect correspondences from objects co-visible on a tick (_collect_correspondences) - generate_predictions tries Path A (homography, HOMOGRAPHY method) first, falling back to Path C (world projection) when no homography exists - wired from the existing HOMOGRAPHY_* Settings - README: flip homography / H-PROJ from planned to implemented. - Tests: 8 new (estimator: known-H recovery, identity, directionality, min-pairs, buffer cap, same-cam ignore; integration: predictions emit HOMOGRAPHY, correspondence collection for co-visible objects). Coverage 55%->58%. ruff/mypy clean, 89 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 27 ++-- backend/app/domain/entities.py | 4 +- backend/app/infrastructure/homography.py | 103 +++++++++++++++ .../app/infrastructure/world_model_adapter.py | 94 ++++++++++++- backend/tests/unit/test_homography.py | 124 ++++++++++++++++++ 5 files changed, 332 insertions(+), 20 deletions(-) create mode 100644 backend/app/infrastructure/homography.py create mode 100644 backend/tests/unit/test_homography.py diff --git a/README.md b/README.md index f687731..c934d95 100644 --- a/README.md +++ b/README.md @@ -100,7 +100,7 @@ that describe them are forward-looking design. | Canvas tactical HUD (brackets, diamonds, velocity, ghosts) | βœ… Implemented | | Mobile phone camera source + standalone page | βœ… Implemented | | Optional JWT auth Β· SSL Β· atomic Jetson deploy | βœ… Implemented | -| Cross-camera homography ghosts (green `H-PROJ`) | πŸ”­ Planned | +| Cross-camera homography ghosts (green `H-PROJ`) | βœ… Implemented | | Pixel-extrapolation ghosts (red `EXTRAP`) | πŸ”­ Planned | | Appearance re-ID (HSV histograms) wired into tracking / fusion | βœ… Implemented | | Sensor-trust scoring | πŸ”­ Planned | @@ -127,8 +127,8 @@ that describe them are forward-looking design. | **Adaptive Kalman filter** | 6-state `[x, y, z, vx, vy, vz]` β€” measurement noise scales by **confidence** (bbox-area & sensor-trust scaling πŸ”­ planned) | | **Cross-camera re-ID** | 64-dim HSV histogram descriptors, L2-normalized, EMA-smoothed (Ξ± = 0.3); computed per detection and used to gate cross-camera association | | **Sensor trust scoring** πŸ”­ | *Planned* β€” per-sensor trust ∈ [0.1, 1.0]; the Kalman update accepts the param but it is fixed at 1.0 today | -| **Cross-camera homography** πŸ”­ | *Planned* β€” self-calibrating ground-plane H via `cv2.findHomography` + RANSAC (no homography code in the pipeline yet) | -| **Ghost predictions** | Path C β€” world-coordinate pinhole projection (orange `WORLD`) is **active**. Path A (homography/green `H-PROJ`) and Path B (pixel extrapolation/red `EXTRAP`) are πŸ”­ planned | +| **Cross-camera homography** | Self-calibrating ground-plane H from shared foot-point observations via `cv2.findHomography` + RANSAC; projects Path-A green `H-PROJ` ghosts | +| **Ghost predictions** | Path A (homography/green `H-PROJ`) and Path C (world projection/orange `WORLD`) are **active**; Path B (pixel extrapolation/red `EXTRAP`) is πŸ”­ planned | ### Platform @@ -465,13 +465,13 @@ The frontend renders a tactical HUD inspired by Anduril's EagleEye UI, implement |---|---|---| | **Detections** | Slate-blue `#64b5f6` | Diamond markers, corner brackets, `PERSON` confidence pill | | **Tracks** | Amber `#ffd740` | Diamond/chevron markers, velocity vector arrows, track ID callouts | -| **Predictions (H-PROJ)** πŸ”­ | Green `#00ff82` solid | *Planned* β€” homography-projected ghost (the renderer supports this color, but the backend never emits this method yet) | +| **Predictions (H-PROJ)** | Green `#00ff82` solid | βœ… Active β€” homography-projected ghost; emitted once a camera-pair homography has been learned | | **Predictions (EXTRAP)** πŸ”­ | Red `#ff5050` dashed | *Planned* β€” pixel-extrapolated ghost (renderer-ready; not emitted by the backend yet) | | **Predictions (WORLD)** | Orange `#ff9800` dashed | βœ… Active β€” world-coordinate pinhole projection; the only ghost path emitted today | | **Compass ribbon** πŸ”­ | β€” | *Planned* β€” not in the current canvas renderer | | **Threat ring** πŸ”­ | Per-IFF color | *Planned* β€” not in the current canvas renderer | -Detection overlays show what the model sees *right now*. Track overlays show persistent identity across frames. Predictions show cross-camera projections β€” **today only the orange `WORLD` path is emitted**; green (homography) and red (extrapolation) are πŸ”­ planned for Phase B. +Detection overlays show what the model sees *right now*. Track overlays show persistent identity across frames. Predictions show cross-camera projections β€” green `H-PROJ` (homography, most accurate) and orange `WORLD` (world-model fallback) are emitted today; red `EXTRAP` (pixel extrapolation) is πŸ”­ planned. --- @@ -507,14 +507,13 @@ different appearance are kept as distinct world objects. --- -## πŸ“ Cross-camera homography β€” planned design πŸ”­ +## πŸ“ Cross-camera homography β€” how it works -> **Status: not yet implemented.** This section is the roadmap for the green `H-PROJ` -> ghost path. Today only Path C (world projection, orange `WORLD`) is active β€” there is no -> `cv2.findHomography` / foot-point collection in the pipeline yet. The text below describes -> how the homography path *will* work once built (Phase B of the roadmap). +> **Status: implemented** (`app/infrastructure/homography.py`). The estimator self-calibrates +> from foot-point correspondences and projects Path-A green `H-PROJ` ghosts, falling back to +> Path C (world projection) when no homography has been learned for a camera pair yet. -The planned signature feature is **homography ghost prediction**: when Camera 0 can't see a person but Camera 1 can, the system will render a ghost overlay on Camera 0's feed showing where that person is. +The signature feature is **homography ghost prediction**: when Camera 0 can't see a person but Camera 1 can, the system renders a ghost overlay on Camera 0's feed showing where that person is. ### The problem with naive extrapolation @@ -576,9 +575,9 @@ The mobile client: ### Cross-camera prediction -> Rows that mention homography / Path A / Path B describe πŸ”­ **planned** behavior (see the -> [Implementation status](#-implementation-status) table). Path C (world projection, orange) -> is the only path active today. +> Path A (homography) and Path C (world projection) are active; rows mentioning Path B +> (pixel extrapolation) describe πŸ”­ **planned** behavior (see the +> [Implementation status](#-implementation-status) table). | Edge case | Behavior | Mitigation | |---|---|---| diff --git a/backend/app/domain/entities.py b/backend/app/domain/entities.py index 43ffe95..3a8947d 100644 --- a/backend/app/domain/entities.py +++ b/backend/app/domain/entities.py @@ -206,7 +206,9 @@ class WorldObject: camera_pixel_positions: Dict[int, Tuple[float, float]] = field(default_factory=dict) camera_pixel_velocities: Dict[int, Tuple[float, float]] = field(default_factory=dict) camera_last_seen: Dict[int, datetime] = field(default_factory=dict) - + # Bottom-center (foot) pixel point per camera, for cross-camera homography + camera_foot_points: Dict[int, Tuple[float, float]] = field(default_factory=dict) + @property def time_since_update(self) -> float: return (datetime.now() - self.last_update).total_seconds() diff --git a/backend/app/infrastructure/homography.py b/backend/app/infrastructure/homography.py new file mode 100644 index 0000000..8774f05 --- /dev/null +++ b/backend/app/infrastructure/homography.py @@ -0,0 +1,103 @@ +"""Cross-camera ground-plane homography estimation (Phase B2). + +Learns a projective transform H_{src->dst} between two cameras' image planes from +foot-point correspondences of the same person observed by both cameras, then projects +a foot point from one camera into another to render a cross-camera "ghost" prediction. + +The estimator is self-calibrating: correspondences accumulate as the world model +re-identifies the same object across cameras; once enough pairs exist for a camera +pair, ``cv2.findHomography`` + RANSAC computes the transform, re-estimated periodically. +""" +from __future__ import annotations + +import logging +from collections import defaultdict +from typing import Dict, List, Optional, Tuple + +import numpy as np +from numpy.typing import NDArray + +try: + import cv2 +except ImportError: # pragma: no cover - cv2 always present in runtime/CI + cv2 = None # type: ignore[assignment] + +logger = logging.getLogger(__name__) + +Point = Tuple[float, float] +CamPair = Tuple[int, int] + + +class HomographyEstimator: + """Accumulates foot-point correspondences and estimates per-camera-pair homographies.""" + + def __init__( + self, + min_pairs: int = 4, + max_pairs: int = 100, + ransac_threshold: float = 12.0, + reestimate_every: int = 5, + ) -> None: + # findHomography needs at least 4 points. + self._min_pairs = max(4, int(min_pairs)) + self._max_pairs = max(self._min_pairs, int(max_pairs)) + self._ransac_threshold = float(ransac_threshold) + self._reestimate_every = max(1, int(reestimate_every)) + + self._pairs: Dict[CamPair, List[Tuple[Point, Point]]] = defaultdict(list) + self._homographies: Dict[CamPair, NDArray[np.float64]] = {} + self._since_estimate: Dict[CamPair, int] = defaultdict(int) + + def add_correspondence( + self, src_cam: int, dst_cam: int, src_pt: Point, dst_pt: Point + ) -> None: + """Record a matched foot point seen as ``src_pt`` in ``src_cam`` and ``dst_pt`` + in ``dst_cam``. Triggers (re-)estimation once enough pairs are available.""" + if src_cam == dst_cam: + return + key = (src_cam, dst_cam) + pairs = self._pairs[key] + pairs.append((src_pt, dst_pt)) + if len(pairs) > self._max_pairs: + pairs.pop(0) # keep the most recent observations + self._since_estimate[key] += 1 + + ready = len(pairs) >= self._min_pairs + first_time = key not in self._homographies + if ready and (first_time or self._since_estimate[key] >= self._reestimate_every): + self._estimate(key) + + def _estimate(self, key: CamPair) -> None: + if cv2 is None: + return + pairs = self._pairs[key] + src = np.array([p[0] for p in pairs], dtype=np.float64) + dst = np.array([p[1] for p in pairs], dtype=np.float64) + try: + H, _mask = cv2.findHomography(src, dst, cv2.RANSAC, self._ransac_threshold) + except Exception as e: # pragma: no cover - defensive + logger.warning(f"findHomography failed for cam{key[0]}->cam{key[1]}: {e}") + return + if H is not None and np.isfinite(H).all(): + self._homographies[key] = np.asarray(H, dtype=np.float64) + self._since_estimate[key] = 0 + logger.debug( + f"H learned: cam{key[0]}->cam{key[1]} from {len(pairs)} pairs" + ) + + def has_homography(self, src_cam: int, dst_cam: int) -> bool: + return (src_cam, dst_cam) in self._homographies + + def source_cameras_for(self, dst_cam: int) -> List[int]: + """Cameras that currently have a valid homography projecting into ``dst_cam``.""" + return [a for (a, b) in self._homographies if b == dst_cam] + + def project(self, src_cam: int, dst_cam: int, point: Point) -> Optional[Point]: + """Project a ``src_cam`` ground-plane point into ``dst_cam`` pixel space.""" + H = self._homographies.get((src_cam, dst_cam)) + if H is None: + return None + v = H @ np.array([point[0], point[1], 1.0]) + if abs(v[2]) < 1e-9: + return None + return (float(v[0] / v[2]), float(v[1] / v[2])) diff --git a/backend/app/infrastructure/world_model_adapter.py b/backend/app/infrastructure/world_model_adapter.py index 33a782b..c89d435 100644 --- a/backend/app/infrastructure/world_model_adapter.py +++ b/backend/app/infrastructure/world_model_adapter.py @@ -16,6 +16,7 @@ Track, WorldObject, PredictedTarget, CameraCalibration, Point3D, Velocity3D, BoundingBox, PredictionMethod, AppearanceDescriptor ) +from app.infrastructure.homography import HomographyEstimator logger = logging.getLogger(__name__) @@ -258,6 +259,13 @@ def __init__(self, config_repo: ConfigurationRepository): ) self._appearance_ema_alpha = 0.3 + # Cross-camera homography (Path A green H-PROJ ghost predictions) + self._homography = HomographyEstimator( + min_pairs=config_repo.get_int("homography_min_pairs", 4), + max_pairs=config_repo.get_int("homography_max_pairs", 100), + ransac_threshold=config_repo.get_float("homography_ransac_threshold", 12.0), + ) + # Initialize default calibrations if positions provided self._init_default_calibrations() @@ -334,7 +342,10 @@ async def update(self, tracks: Dict[int, List[Track]]) -> List[WorldObject]: for camera_id, camera_tracks in tracks.items(): for track in camera_tracks: await self._process_track(camera_id, track, now) - + + # Feed cross-camera homography from objects co-visible this tick + self._collect_correspondences(now) + # Clean up old objects self._cleanup_old_objects(now) @@ -437,6 +448,7 @@ def _update_existing_object( obj.camera_pixel_positions[track.camera_id] = track.bbox.center obj.camera_pixel_velocities[track.camera_id] = track.velocity obj.camera_last_seen[track.camera_id] = timestamp + obj.camera_foot_points[track.camera_id] = (track.bbox.center[0], track.bbox.y2) # EMA-smooth the appearance descriptor for re-ID stability obj.appearance = self._blend_appearance(obj.appearance, track.appearance) @@ -465,7 +477,8 @@ def _create_new_object( source_tracks={track.camera_id: track.track_id}, camera_pixel_positions={track.camera_id: track.bbox.center}, camera_pixel_velocities={track.camera_id: track.velocity}, - camera_last_seen={track.camera_id: timestamp} + camera_last_seen={track.camera_id: timestamp}, + camera_foot_points={track.camera_id: (track.bbox.center[0], track.bbox.y2)}, ) self._world_objects[object_id] = obj @@ -557,7 +570,70 @@ def _cleanup_old_objects(self, current_time: datetime) -> None: def get_world_objects(self) -> List[WorldObject]: """Get all current world objects.""" return list(self._world_objects.values()) - + + def _collect_correspondences(self, now: datetime) -> None: + """Feed foot-point correspondences for objects co-visible on this tick. + + An object seen by two cameras at the same instant gives one matched + ground-plane point per camera pair, which the homography estimator uses + to self-calibrate the camera-to-camera transform. + """ + for obj in self._world_objects.values(): + cams = [ + c for c, seen in obj.camera_last_seen.items() + if seen == now and c in obj.camera_foot_points + ] + if len(cams) < 2: + continue + for src in cams: + for dst in cams: + if src == dst: + continue + self._homography.add_correspondence( + src, dst, + obj.camera_foot_points[src], + obj.camera_foot_points[dst], + ) + + def _try_homography_prediction( + self, camera_id: int, obj: WorldObject, time_since_seen: float + ) -> Optional[PredictedTarget]: + """Project ``obj``'s foot point from a source camera into ``camera_id`` via + a learned homography. Returns a HOMOGRAPHY prediction, or None if no usable + homography/foot-point exists (caller then falls back to world projection).""" + for src_cam in self._homography.source_cameras_for(camera_id): + foot = obj.camera_foot_points.get(src_cam) + if foot is None: + continue + projected = self._homography.project(src_cam, camera_id, foot) + if projected is None: + continue + + depth = max(abs(obj.position.z), 0.5) + bbox_height = min(500, max(50, 500 / depth)) + bbox_width = bbox_height * 0.4 + fx, fy = projected + try: + # Foot point is the bottom-centre; the body extends upward. + bbox = BoundingBox(fx - bbox_width / 2, fy - bbox_height, fx + bbox_width / 2, fy) + except ValueError: + continue + + confidence = obj.confidence * max( + 0.1, 1.0 - time_since_seen / self._prediction_horizon + ) + return PredictedTarget( + object_id=obj.object_id, + camera_id=camera_id, + predicted_bbox=bbox, + confidence=confidence, + time_since_seen=time_since_seen, + velocity_projection=(obj.velocity.vx, obj.velocity.vy), + source_camera=src_cam, + prediction_method=PredictionMethod.HOMOGRAPHY, + ) + return None + def generate_predictions(self, camera_id: int) -> List[PredictedTarget]: """Generate predictions for a camera view.""" # A view-only camera still needs a calibration to project world objects. @@ -574,8 +650,16 @@ def generate_predictions(self, camera_id: int) -> List[PredictedTarget]: time_since_seen = (now - obj.last_update).total_seconds() if time_since_seen > self._prediction_horizon: continue - - # Try world-to-pixel projection + + # Path A: cross-camera homography (green H-PROJ) β€” most accurate. + homography_pred = self._try_homography_prediction( + camera_id, obj, time_since_seen + ) + if homography_pred is not None: + predictions.append(homography_pred) + continue + + # Path C: world-to-pixel projection (orange WORLD) β€” always-available fallback. pixel = self._transformer.world_to_pixel(camera_id, obj.position) if pixel is None: diff --git a/backend/tests/unit/test_homography.py b/backend/tests/unit/test_homography.py new file mode 100644 index 0000000..b379036 --- /dev/null +++ b/backend/tests/unit/test_homography.py @@ -0,0 +1,124 @@ +"""Phase B2 β€” cross-camera ground-plane homography estimator.""" +import numpy as np +import pytest + +pytest.importorskip("cv2") + +from app.infrastructure.homography import HomographyEstimator + + +def _apply_h(H, pt): + v = H @ np.array([pt[0], pt[1], 1.0]) + return (v[0] / v[2], v[1] / v[2]) + + +def test_no_homography_until_min_pairs(): + est = HomographyEstimator(min_pairs=4) + for i in range(3): + est.add_correspondence(0, 1, (i, i), (i, i)) + assert not est.has_homography(0, 1) + assert est.project(0, 1, (1, 1)) is None + + +def test_recovers_known_homography(): + H_true = np.array([ + [1.2, 0.1, 30.0], + [0.05, 1.1, -20.0], + [0.0001, 0.0002, 1.0], + ]) + est = HomographyEstimator(min_pairs=4) + for s in [(10, 10), (200, 20), (30, 220), (250, 240), (120, 130)]: + est.add_correspondence(0, 1, s, _apply_h(H_true, s)) + assert est.has_homography(0, 1) + test_pt = (160, 90) + expected = _apply_h(H_true, test_pt) + got = est.project(0, 1, test_pt) + assert got is not None + assert got[0] == pytest.approx(expected[0], abs=1e-2) + assert got[1] == pytest.approx(expected[1], abs=1e-2) + + +def test_identity_when_src_equals_dst(): + est = HomographyEstimator(min_pairs=4) + for s in [(0, 0), (100, 0), (0, 100), (100, 100), (50, 50)]: + est.add_correspondence(0, 1, s, s) + got = est.project(0, 1, (42, 17)) + assert got is not None + assert got[0] == pytest.approx(42, abs=1e-3) + assert got[1] == pytest.approx(17, abs=1e-3) + + +def test_directional_homographies_are_independent(): + est = HomographyEstimator(min_pairs=4) + for s in [(0, 0), (100, 0), (0, 100), (100, 100)]: + est.add_correspondence(0, 1, s, (s[0] + 10, s[1])) + assert est.has_homography(0, 1) + assert not est.has_homography(1, 0) + assert est.source_cameras_for(1) == [0] + + +def test_same_camera_correspondence_ignored(): + est = HomographyEstimator(min_pairs=4) + for s in [(0, 0), (100, 0), (0, 100), (100, 100)]: + est.add_correspondence(2, 2, s, s) + assert not est.has_homography(2, 2) + + +def test_pair_buffer_capped_at_max_pairs(): + est = HomographyEstimator(min_pairs=4, max_pairs=10) + for i in range(50): + est.add_correspondence(0, 1, (i, i), (i, i)) + assert len(est._pairs[(0, 1)]) <= 10 + + +# ---------------------------------------------- world-model integration + +def _world_repo(): + from unittest.mock import Mock + from app.infrastructure.world_model_adapter import WorldModelRepositoryImpl + config = Mock() + config.get.return_value = {} + config.get_int.return_value = 4 + config.get_float.return_value = 0.5 + config.get_list.return_value = [] + return WorldModelRepositoryImpl(config) + + +def test_generate_predictions_uses_homography_when_available(): + from datetime import datetime + from app.domain.entities import ( + PredictionMethod, Point3D, Velocity3D, WorldObject, + ) + repo = _world_repo() + for s in [(0, 0), (100, 0), (0, 100), (100, 100), (50, 50)]: + repo._homography.add_correspondence(0, 1, s, s) # identity cam0 -> cam1 + assert repo._homography.has_homography(0, 1) + + now = datetime.now() + repo._world_objects[1] = WorldObject( + object_id=1, position=Point3D(1, 0, 3), velocity=Velocity3D(0, 0, 0), + class_id=0, class_name="person", confidence=0.9, + last_seen_camera=0, last_update=now, + source_tracks={0: 5}, camera_foot_points={0: (200, 300)}, + ) + preds = repo.generate_predictions(camera_id=1) + assert len(preds) == 1 + assert preds[0].prediction_method == PredictionMethod.HOMOGRAPHY + assert preds[0].source_camera == 0 + + +def test_collect_correspondences_feeds_estimator_for_covisible_object(): + from datetime import datetime + from app.domain.entities import Point3D, Velocity3D, WorldObject + repo = _world_repo() + now = datetime.now() + repo._world_objects[1] = WorldObject( + object_id=1, position=Point3D(1, 0, 3), velocity=Velocity3D(0, 0, 0), + class_id=0, class_name="person", confidence=0.9, last_seen_camera=1, + last_update=now, + camera_last_seen={0: now, 1: now}, + camera_foot_points={0: (10, 20), 1: (30, 40)}, + ) + repo._collect_correspondences(now) + assert len(repo._homography._pairs[(0, 1)]) == 1 + assert len(repo._homography._pairs[(1, 0)]) == 1 From 0d6c540f764f64af866311206adcc3aafe576c63 Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 19:54:56 +0530 Subject: [PATCH 7/9] feat(predictions): pixel-extrapolation ghosts (red EXTRAP / Path B) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase B3 β€” the third and final ghost-prediction path. When a camera loses a person it was previously tracking (but the world object is still alive via another camera), dead-reckon a red EXTRAP ghost from this camera's last-known pixel position along its pixel velocity, capped by an adaptive budget. - world model: refine the prediction skip from "ever seen by this camera" to "seen within a live window" (~2 frames), so a camera that LOST a target becomes eligible for a ghost. generate_predictions now tries Path A (homography) -> Path B (extrapolation) -> Path C (world projection). - _try_extrapolation_prediction: slide last pixel by velocity * time * fps, capped at min(250, 80 + 40*t) px; zero-velocity stays put; no pixel history -> None. - README: flip EXTRAP / red ghost from planned to implemented; all 3 paths now active. - Tests: 6 new (no-history None, moves in velocity direction, budget cap, zero-velocity stays, integration emits EXTRAP for a lost camera, live camera skipped). Coverage 58%. ruff/mypy clean, 95 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 13 +-- .../app/infrastructure/world_model_adapter.py | 69 +++++++++++- backend/tests/unit/test_extrapolation.py | 105 ++++++++++++++++++ 3 files changed, 177 insertions(+), 10 deletions(-) create mode 100644 backend/tests/unit/test_extrapolation.py diff --git a/README.md b/README.md index c934d95..6840221 100644 --- a/README.md +++ b/README.md @@ -101,7 +101,7 @@ that describe them are forward-looking design. | Mobile phone camera source + standalone page | βœ… Implemented | | Optional JWT auth Β· SSL Β· atomic Jetson deploy | βœ… Implemented | | Cross-camera homography ghosts (green `H-PROJ`) | βœ… Implemented | -| Pixel-extrapolation ghosts (red `EXTRAP`) | πŸ”­ Planned | +| Pixel-extrapolation ghosts (red `EXTRAP`) | βœ… Implemented | | Appearance re-ID (HSV histograms) wired into tracking / fusion | βœ… Implemented | | Sensor-trust scoring | πŸ”­ Planned | | Adaptive Kalman noise by bbox area | πŸ”­ Planned | @@ -128,7 +128,7 @@ that describe them are forward-looking design. | **Cross-camera re-ID** | 64-dim HSV histogram descriptors, L2-normalized, EMA-smoothed (Ξ± = 0.3); computed per detection and used to gate cross-camera association | | **Sensor trust scoring** πŸ”­ | *Planned* β€” per-sensor trust ∈ [0.1, 1.0]; the Kalman update accepts the param but it is fixed at 1.0 today | | **Cross-camera homography** | Self-calibrating ground-plane H from shared foot-point observations via `cv2.findHomography` + RANSAC; projects Path-A green `H-PROJ` ghosts | -| **Ghost predictions** | Path A (homography/green `H-PROJ`) and Path C (world projection/orange `WORLD`) are **active**; Path B (pixel extrapolation/red `EXTRAP`) is πŸ”­ planned | +| **Ghost predictions** | All three paths active β€” Path A (homography/green `H-PROJ`), Path B (pixel extrapolation/red `EXTRAP`), Path C (world projection/orange `WORLD`) | ### Platform @@ -466,12 +466,12 @@ The frontend renders a tactical HUD inspired by Anduril's EagleEye UI, implement | **Detections** | Slate-blue `#64b5f6` | Diamond markers, corner brackets, `PERSON` confidence pill | | **Tracks** | Amber `#ffd740` | Diamond/chevron markers, velocity vector arrows, track ID callouts | | **Predictions (H-PROJ)** | Green `#00ff82` solid | βœ… Active β€” homography-projected ghost; emitted once a camera-pair homography has been learned | -| **Predictions (EXTRAP)** πŸ”­ | Red `#ff5050` dashed | *Planned* β€” pixel-extrapolated ghost (renderer-ready; not emitted by the backend yet) | +| **Predictions (EXTRAP)** | Red `#ff5050` dashed | βœ… Active β€” pixel-extrapolated dead-reckoning when a camera loses a previously-seen target | | **Predictions (WORLD)** | Orange `#ff9800` dashed | βœ… Active β€” world-coordinate pinhole projection; the only ghost path emitted today | | **Compass ribbon** πŸ”­ | β€” | *Planned* β€” not in the current canvas renderer | | **Threat ring** πŸ”­ | Per-IFF color | *Planned* β€” not in the current canvas renderer | -Detection overlays show what the model sees *right now*. Track overlays show persistent identity across frames. Predictions show cross-camera projections β€” green `H-PROJ` (homography, most accurate) and orange `WORLD` (world-model fallback) are emitted today; red `EXTRAP` (pixel extrapolation) is πŸ”­ planned. +Detection overlays show what the model sees *right now*. Track overlays show persistent identity across frames. Predictions show cross-camera projections β€” green `H-PROJ` (homography, most accurate), orange `WORLD` (world-model fallback), and red `EXTRAP` (pixel extrapolation, when a camera has lost a previously-seen target) are all emitted. --- @@ -575,9 +575,8 @@ The mobile client: ### Cross-camera prediction -> Path A (homography) and Path C (world projection) are active; rows mentioning Path B -> (pixel extrapolation) describe πŸ”­ **planned** behavior (see the -> [Implementation status](#-implementation-status) table). +> All three prediction paths β€” Path A (homography), Path B (pixel extrapolation), and +> Path C (world projection) β€” are active. See the [Implementation status](#-implementation-status) table. | Edge case | Behavior | Mitigation | |---|---|---| diff --git a/backend/app/infrastructure/world_model_adapter.py b/backend/app/infrastructure/world_model_adapter.py index c89d435..2dc734c 100644 --- a/backend/app/infrastructure/world_model_adapter.py +++ b/backend/app/infrastructure/world_model_adapter.py @@ -266,6 +266,13 @@ def __init__(self, config_repo: ConfigurationRepository): ransac_threshold=config_repo.get_float("homography_ransac_threshold", 12.0), ) + # Pixel-extrapolation ghosts (Path B red EXTRAP) + fps = config_repo.get_float("target_fps", 24.0) + self._extrap_fps = fps if fps > 0 else 24.0 + # A camera seen within this window is "live"; longer means it has lost the + # object and becomes eligible for a ghost prediction. + self._live_track_seconds = 2.0 / self._extrap_fps + # Initialize default calibrations if positions provided self._init_default_calibrations() @@ -634,6 +641,51 @@ def _try_homography_prediction( ) return None + def _try_extrapolation_prediction( + self, camera_id: int, obj: WorldObject, time_since_seen: float + ) -> Optional[PredictedTarget]: + """Dead-reckon a ghost from this camera's last-known pixel position, sliding it + along the per-camera pixel velocity with an adaptive budget. Returns None when + this camera has no pixel history for the object (so the caller falls through).""" + last_pixel = obj.camera_pixel_positions.get(camera_id) + if last_pixel is None: + return None + + vx, vy = obj.camera_pixel_velocities.get(camera_id, (0.0, 0.0)) + speed = (vx * vx + vy * vy) ** 0.5 + budget = min(250.0, 80.0 + 40.0 * time_since_seen) + if speed < 1e-6: + cx, cy = last_pixel + else: + disp = min(budget, speed * time_since_seen * self._extrap_fps) + cx = last_pixel[0] + (vx / speed) * disp + cy = last_pixel[1] + (vy / speed) * disp + + depth = max(abs(obj.position.z), 0.5) + bbox_height = min(500, max(50, 500 / depth)) + bbox_width = bbox_height * 0.4 + try: + bbox = BoundingBox( + cx - bbox_width / 2, cy - bbox_height / 2, + cx + bbox_width / 2, cy + bbox_height / 2, + ) + except ValueError: + return None + + confidence = obj.confidence * max( + 0.1, 1.0 - time_since_seen / self._prediction_horizon + ) + return PredictedTarget( + object_id=obj.object_id, + camera_id=camera_id, + predicted_bbox=bbox, + confidence=confidence, + time_since_seen=time_since_seen, + velocity_projection=(vx, vy), + source_camera=camera_id, + prediction_method=PredictionMethod.EXTRAPOLATION, + ) + def generate_predictions(self, camera_id: int) -> List[PredictedTarget]: """Generate predictions for a camera view.""" # A view-only camera still needs a calibration to project world objects. @@ -643,10 +695,12 @@ def generate_predictions(self, camera_id: int) -> List[PredictedTarget]: now = datetime.now() for obj in self._world_objects.values(): - # Skip if already seen by this camera - if camera_id in obj.source_tracks: + # Skip objects this camera is tracking live (seen within the live window). + last_seen_here = obj.camera_last_seen.get(camera_id) + if last_seen_here is not None and \ + (now - last_seen_here).total_seconds() < self._live_track_seconds: continue - + time_since_seen = (now - obj.last_update).total_seconds() if time_since_seen > self._prediction_horizon: continue @@ -659,6 +713,15 @@ def generate_predictions(self, camera_id: int) -> List[PredictedTarget]: predictions.append(homography_pred) continue + # Path B: pixel extrapolation (red EXTRAP) β€” dead-reckon from this camera's + # own last-known pixel position. Only fires if it saw the object before. + extrap_pred = self._try_extrapolation_prediction( + camera_id, obj, time_since_seen + ) + if extrap_pred is not None: + predictions.append(extrap_pred) + continue + # Path C: world-to-pixel projection (orange WORLD) β€” always-available fallback. pixel = self._transformer.world_to_pixel(camera_id, obj.position) diff --git a/backend/tests/unit/test_extrapolation.py b/backend/tests/unit/test_extrapolation.py new file mode 100644 index 0000000..c565bbd --- /dev/null +++ b/backend/tests/unit/test_extrapolation.py @@ -0,0 +1,105 @@ +"""Phase B3 β€” pixel-extrapolation ghosts (red EXTRAP, Path B).""" +from datetime import datetime, timedelta +from unittest.mock import Mock + +import pytest + +pytest.importorskip("cv2") + +from app.domain.entities import ( + PredictionMethod, Point3D, Velocity3D, WorldObject, +) + + +def _world_repo(): + from app.infrastructure.world_model_adapter import WorldModelRepositoryImpl + config = Mock() + config.get.return_value = {} + config.get_int.return_value = 4 + config.get_float.return_value = 24.0 # fps and other floats + config.get_list.return_value = [] + repo = WorldModelRepositoryImpl(config) + repo._prediction_horizon = 5.0 + return repo + + +def _obj(**kw): + base = dict( + object_id=1, position=Point3D(0, 0, 3), velocity=Velocity3D(0, 0, 0), + class_id=0, class_name="person", confidence=0.9, + last_seen_camera=0, last_update=datetime.now(), + ) + base.update(kw) + return WorldObject(**base) + + +def _center(pred): + b = pred.predicted_bbox + return ((b.x1 + b.x2) / 2, (b.y1 + b.y2) / 2) + + +def test_extrapolation_none_without_pixel_history(): + repo = _world_repo() + obj = _obj() # no camera_pixel_positions for cam 1 + assert repo._try_extrapolation_prediction(1, obj, 0.2) is None + + +def test_extrapolation_moves_in_velocity_direction(): + repo = _world_repo() + obj = _obj( + camera_pixel_positions={1: (100.0, 100.0)}, + camera_pixel_velocities={1: (5.0, 0.0)}, # moving +x + ) + pred = repo._try_extrapolation_prediction(1, obj, time_since_seen=0.2) + assert pred is not None + assert pred.prediction_method == PredictionMethod.EXTRAPOLATION + cx, cy = _center(pred) + assert cx > 100.0 + assert cy == pytest.approx(100.0, abs=1e-6) + + +def test_extrapolation_displacement_capped_by_budget(): + repo = _world_repo() + obj = _obj( + camera_pixel_positions={1: (100.0, 100.0)}, + camera_pixel_velocities={1: (100000.0, 0.0)}, # absurd speed + ) + pred = repo._try_extrapolation_prediction(1, obj, time_since_seen=0.2) + cx, _ = _center(pred) + budget = min(250.0, 80.0 + 40.0 * 0.2) + assert cx - 100.0 <= budget + 1e-6 + + +def test_zero_velocity_stays_at_last_pixel(): + repo = _world_repo() + obj = _obj( + camera_pixel_positions={1: (140.0, 160.0)}, + camera_pixel_velocities={1: (0.0, 0.0)}, + ) + cx, cy = _center(repo._try_extrapolation_prediction(1, obj, 1.0)) + assert (cx, cy) == pytest.approx((140.0, 160.0)) + + +def test_generate_predictions_emits_extrapolation_for_lost_camera(): + repo = _world_repo() + now = datetime.now() + repo._world_objects[1] = _obj( + last_update=now, # object still fresh (seen by someone now) + source_tracks={1: 9}, + camera_pixel_positions={1: (100.0, 100.0)}, + camera_pixel_velocities={1: (5.0, 0.0)}, + camera_last_seen={1: now - timedelta(seconds=100)}, # cam 1 lost it long ago + ) + methods = [p.prediction_method for p in repo.generate_predictions(camera_id=1)] + assert PredictionMethod.EXTRAPOLATION in methods + + +def test_live_camera_is_skipped(): + repo = _world_repo() + now = datetime.now() + repo._world_objects[1] = _obj( + last_update=now, + camera_pixel_positions={1: (100.0, 100.0)}, + camera_last_seen={1: now}, # seen this instant -> live -> no ghost for cam 1 + ) + assert repo.generate_predictions(camera_id=1) == [] From 67b8a6836eac2421a98c2c35807956547333c8f9 Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 20:01:15 +0530 Subject: [PATCH 8/9] feat(fusion): fuse mobile GPS/IMU into camera calibration Phase B4. The mobile client streams GPS (watchPosition) + IMU (DeviceOrientationEvent), but the backend was receiving and discarding the sensor_data message. Now it's fused. - New app/infrastructure/geo.py: gps_to_local() equirectangular projection. - World model: update_camera_sensor() converts GPS -> local position (relative to GPS_REFERENCE_* or the first fix as origin) and DeviceOrientation alpha/beta/gamma -> camera rotation, overriding the auto-default calibration. Added to the WorldModelRepository port (concrete no-op default). - main.py: /ws/camera handler forwards sensor_data to the world model instead of pass. - README: flip GPS/IMU fusion from planned to implemented. - Tests: 8 new (geo: origin/north/east projection; fusion: first-fix origin, north offset, configured reference, orientation->rotation, no-data no-op). Coverage 58%->59%. ruff/mypy clean, 103 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 6 +- backend/app/application/ports.py | 13 ++++ backend/app/infrastructure/geo.py | 21 ++++++ .../app/infrastructure/world_model_adapter.py | 66 ++++++++++++++++++- backend/main.py | 13 +++- backend/tests/unit/test_geo.py | 27 ++++++++ backend/tests/unit/test_gps_fusion.py | 60 +++++++++++++++++ 7 files changed, 200 insertions(+), 6 deletions(-) create mode 100644 backend/app/infrastructure/geo.py create mode 100644 backend/tests/unit/test_geo.py create mode 100644 backend/tests/unit/test_gps_fusion.py diff --git a/README.md b/README.md index 6840221..fcd7c25 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,7 @@ that describe them are forward-looking design. | Appearance re-ID (HSV histograms) wired into tracking / fusion | βœ… Implemented | | Sensor-trust scoring | πŸ”­ Planned | | Adaptive Kalman noise by bbox area | πŸ”­ Planned | -| GPS + IMU fusion into the world model | πŸ”­ Planned | +| GPS + IMU fusion into the world model | βœ… Implemented | | DeepSORT / centroid tracker fallback chain | πŸ”­ Planned | > Config and helper scaffolding for the πŸ”­ items already exists (`compute_appearance()`, @@ -136,7 +136,7 @@ that describe them are forward-looking design. |---|---| | **Multi-camera** | Up to 4 concurrent streams (physical MJPEG/RTSP + mobile virtual cameras) | | **Mobile streaming** | Phone browsers β†’ `getUserMedia` β†’ binary JPEG over WebSocket β†’ `VirtualCamera` | -| **GPS + IMU fusion** πŸ”­ | *Planned* β€” the mobile client sends GPS (`watchPosition`) + IMU (`DeviceOrientationEvent`), but the backend currently receives and discards `sensor_data`; fusion into the world model is not wired yet | +| **GPS + IMU fusion** | Mobile geolocation (`watchPosition`) β†’ equirectangular projection into the local frame; `DeviceOrientationEvent` β†’ camera rotation. Fused into the camera calibration (`GPS_REFERENCE_*` or first-fix origin) | | **AR overlays** | Canvas-based: cyan detection brackets, amber track boxes, green/orange/red ghost predictions | | **Binary protocol** | msgpack-serialized snapshots β€” zero-copy broadcast to all viewers | | **SSL/TLS** | Self-signed certificates with SAN for LAN IP access (required for `getUserMedia`) | @@ -399,7 +399,7 @@ When `AUTH_ENABLED=true`, both endpoints require a `?token=` query paramete Client β†’ { "type": "register", "role": "camera_source", "camera_id": null } Server β†’ { "type": "registered", "camera_id": 0 } Client β†’ [binary JPEG frames] -Client β†’ { "type": "sensor_data", "gps": {...}, "orientation": {...} } # received, not yet fused πŸ”­ +Client β†’ { "type": "sensor_data", "gps": {...}, "orientation": {...} } # fused into camera calibration ``` --- diff --git a/backend/app/application/ports.py b/backend/app/application/ports.py index 0287e25..92ad850 100644 --- a/backend/app/application/ports.py +++ b/backend/app/application/ports.py @@ -157,6 +157,19 @@ def get_camera_calibration(self, camera_id: int) -> Optional[CameraCalibration]: """Get calibration for a camera.""" pass + def update_camera_sensor( + self, + camera_id: int, + gps: Optional[Dict[str, Any]] = None, + orientation: Optional[Dict[str, Any]] = None, + ) -> None: + """Fuse mobile GPS/IMU sensor data into a camera's calibration. + + Optional capability; the default is a no-op so adapters that don't support + sensor fusion remain valid. + """ + return None + class FrameEncoderRepository(ABC): """Repository for frame encoding.""" diff --git a/backend/app/infrastructure/geo.py b/backend/app/infrastructure/geo.py new file mode 100644 index 0000000..f610b3f --- /dev/null +++ b/backend/app/infrastructure/geo.py @@ -0,0 +1,21 @@ +"""Geographic helpers for fusing mobile GPS into the local world frame.""" +from __future__ import annotations + +import math +from typing import Tuple + +EARTH_RADIUS_M = 6_371_000.0 + + +def gps_to_local( + lat: float, lng: float, ref_lat: float, ref_lng: float +) -> Tuple[float, float]: + """Equirectangular projection of ``(lat, lng)`` to local ``(x_east, y_north)`` + metres relative to a reference point. Accurate for the small areas a backpack + multi-camera rig covers; the reference is the local-frame origin. + """ + d_lat = math.radians(lat - ref_lat) + d_lng = math.radians(lng - ref_lng) + x_east = d_lng * math.cos(math.radians(ref_lat)) * EARTH_RADIUS_M + y_north = d_lat * EARTH_RADIUS_M + return (x_east, y_north) diff --git a/backend/app/infrastructure/world_model_adapter.py b/backend/app/infrastructure/world_model_adapter.py index 2dc734c..426ca89 100644 --- a/backend/app/infrastructure/world_model_adapter.py +++ b/backend/app/infrastructure/world_model_adapter.py @@ -4,7 +4,8 @@ """ from __future__ import annotations import logging -from typing import List, Dict, Optional, Tuple +import math +from typing import Any, List, Dict, Optional, Tuple from dataclasses import dataclass, field from datetime import datetime @@ -17,6 +18,7 @@ Point3D, Velocity3D, BoundingBox, PredictionMethod, AppearanceDescriptor ) from app.infrastructure.homography import HomographyEstimator +from app.infrastructure.geo import gps_to_local logger = logging.getLogger(__name__) @@ -273,6 +275,13 @@ def __init__(self, config_repo: ConfigurationRepository): # object and becomes eligible for a ghost prediction. self._live_track_seconds = 2.0 / self._extrap_fps + # GPS/IMU fusion reference (None -> the first GPS fix becomes the local origin) + ref_lat = config_repo.get("gps_reference_lat") + ref_lng = config_repo.get("gps_reference_lng") + self._gps_ref: Optional[Tuple[float, float]] = None + if isinstance(ref_lat, (int, float)) and isinstance(ref_lng, (int, float)): + self._gps_ref = (float(ref_lat), float(ref_lng)) + # Initialize default calibrations if positions provided self._init_default_calibrations() @@ -766,3 +775,58 @@ def update_camera_calibration(self, calibration: CameraCalibration) -> None: def get_camera_calibration(self, camera_id: int) -> Optional[CameraCalibration]: """Get calibration for a camera.""" return self._transformer._calibrations.get(camera_id) + + def update_camera_sensor( + self, + camera_id: int, + gps: Optional[Dict[str, Any]] = None, + orientation: Optional[Dict[str, Any]] = None, + ) -> None: + """Fuse mobile GPS/IMU into ``camera_id``'s calibration. + + GPS -> local position (equirectangular projection); DeviceOrientation + (alpha/beta/gamma degrees) -> camera rotation (roll, pitch, yaw radians). + Overrides the auto-default calibration for that camera. Missing fields are + kept from any existing calibration. + """ + position: Optional[Point3D] = None + if gps: + lat, lng = gps.get("latitude"), gps.get("longitude") + if isinstance(lat, (int, float)) and isinstance(lng, (int, float)): + if self._gps_ref is None: + self._gps_ref = (float(lat), float(lng)) # first fix = local origin + x, y = gps_to_local( + float(lat), float(lng), self._gps_ref[0], self._gps_ref[1] + ) + alt = gps.get("altitude") + z = float(alt) if isinstance(alt, (int, float)) else self._default_camera_height + position = Point3D(x, y, z) + + rotation: Optional[Tuple[float, float, float]] = None + if orientation: + alpha = orientation.get("alpha") or 0.0 # compass heading -> yaw + beta = orientation.get("beta") or 0.0 # front-back tilt -> pitch + gamma = orientation.get("gamma") or 0.0 # left-right tilt -> roll + rotation = ( + math.radians(float(gamma)), + math.radians(float(beta)), + math.radians(float(alpha)), + ) + + if position is None and rotation is None: + return + + existing = self._transformer._calibrations.get(camera_id) + accuracy = gps.get("accuracy") if gps else None + self._transformer.set_calibration(CameraCalibration( + camera_id=camera_id, + position=position or ( + existing.position if existing + else Point3D(0.0, 0.0, self._default_camera_height) + ), + rotation=rotation or (existing.rotation if existing else (0.0, 0.0, 0.0)), + focal_length=existing.focal_length if existing else self._default_focal_length, + image_center=existing.image_center if existing else (640.0, 360.0), + gps_accuracy=float(accuracy) if isinstance(accuracy, (int, float)) else 5.0, + )) + logger.info(f"Camera {camera_id}: calibration updated from GPS/IMU sensor data") diff --git a/backend/main.py b/backend/main.py index 51db5e1..a9cbd27 100644 --- a/backend/main.py +++ b/backend/main.py @@ -350,8 +350,17 @@ async def camera_websocket_endpoint(websocket: WebSocket): # Send pong to keep connection alive await websocket.send_text(json.dumps({"type": "pong"})) elif msg_type == "sensor_data": - # Handle sensor data if needed - pass + # Fuse mobile GPS/IMU into this camera's calibration. + try: + container.world_model_repo.update_camera_sensor( + camera_id, + gps=control.get("gps"), + orientation=control.get("orientation"), + ) + except Exception as e: + logger.warning( + f"Camera {camera_id}: sensor_data fusion error: {e}" + ) # Check for timeout (no frames for 60 seconds) if time.time() - last_frame_time > 60: diff --git a/backend/tests/unit/test_geo.py b/backend/tests/unit/test_geo.py new file mode 100644 index 0000000..e346a80 --- /dev/null +++ b/backend/tests/unit/test_geo.py @@ -0,0 +1,27 @@ +"""Phase B4 β€” GPS -> local equirectangular projection.""" +import math + +import pytest + +from app.infrastructure.geo import gps_to_local + + +def test_reference_point_maps_to_origin(): + x, y = gps_to_local(40.0, -74.0, 40.0, -74.0) + assert x == pytest.approx(0.0, abs=1e-6) + assert y == pytest.approx(0.0, abs=1e-6) + + +def test_north_offset_is_positive_y(): + x, y = gps_to_local(40.001, -74.0, 40.0, -74.0) + assert x == pytest.approx(0.0, abs=1e-6) + assert y == pytest.approx(111.19, abs=1.0) # ~111 m per 0.001 deg latitude + assert y > 0 + + +def test_east_offset_is_positive_x_scaled_by_cos_lat(): + x, y = gps_to_local(40.0, -73.999, 40.0, -74.0) + expected = math.radians(0.001) * math.cos(math.radians(40.0)) * 6_371_000.0 + assert x == pytest.approx(expected, abs=1e-3) + assert y == pytest.approx(0.0, abs=1e-6) + assert x > 0 diff --git a/backend/tests/unit/test_gps_fusion.py b/backend/tests/unit/test_gps_fusion.py new file mode 100644 index 0000000..ae0e6aa --- /dev/null +++ b/backend/tests/unit/test_gps_fusion.py @@ -0,0 +1,60 @@ +"""Phase B4 β€” fuse mobile GPS/IMU sensor data into camera calibration.""" +import math +from unittest.mock import Mock + +import pytest + +pytest.importorskip("cv2") + + +def _repo(reference=None): + from app.infrastructure.world_model_adapter import WorldModelRepositoryImpl + config = Mock() + ref = { + "gps_reference_lat": reference[0] if reference else None, + "gps_reference_lng": reference[1] if reference else None, + } + config.get.side_effect = lambda k, d=None: ref.get(k, {} if d is None else d) + config.get_int.return_value = 4 + config.get_float.return_value = 24.0 + config.get_list.return_value = [] + return WorldModelRepositoryImpl(config) + + +def test_first_gps_fix_becomes_local_origin(): + repo = _repo() + repo.update_camera_sensor(0, gps={"latitude": 40.0, "longitude": -74.0, "altitude": 2.0}) + calib = repo.get_camera_calibration(0) + assert calib is not None + assert calib.position.x == pytest.approx(0.0, abs=1e-6) + assert calib.position.y == pytest.approx(0.0, abs=1e-6) + assert calib.position.z == pytest.approx(2.0) + + +def test_second_fix_offset_north_gives_positive_y(): + repo = _repo() + repo.update_camera_sensor(0, gps={"latitude": 40.0, "longitude": -74.0}) # origin + repo.update_camera_sensor(1, gps={"latitude": 40.001, "longitude": -74.0}) + c1 = repo.get_camera_calibration(1) + assert c1.position.y > 100 # ~111 m north of the origin fix + assert c1.position.x == pytest.approx(0.0, abs=1e-3) + + +def test_configured_reference_is_used(): + repo = _repo(reference=(40.0, -74.0)) + repo.update_camera_sensor(0, gps={"latitude": 40.001, "longitude": -74.0}) + assert repo.get_camera_calibration(0).position.y > 100 + + +def test_orientation_sets_rotation_yaw(): + repo = _repo() + repo.update_camera_sensor(0, orientation={"alpha": 90.0, "beta": 0.0, "gamma": 0.0}) + calib = repo.get_camera_calibration(0) + assert calib is not None + assert calib.rotation[2] == pytest.approx(math.radians(90.0)) # yaw from alpha + + +def test_no_sensor_data_is_noop(): + repo = _repo() + repo.update_camera_sensor(0) + assert repo.get_camera_calibration(0) is None From 8d4d8604cc19a835e89d981de95e20e2634aa230 Mon Sep 17 00:00:00 2001 From: mandarwagh9 Date: Fri, 19 Jun 2026 20:06:08 +0530 Subject: [PATCH 9/9] feat(fusion): sensor-trust scoring + adaptive Kalman by bbox area MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase B5 β€” the final roadmap feature. The Kalman update accepted a sensor_trust param but it was always 1.0, and measurement noise scaled only by confidence. - KalmanFilter.update now scales R by confidence * sensor_trust * bbox-area factor, and returns the innovation magnitude. - World model tracks per-camera trust in [0.1, 1.0]: consistent measurements (low innovation) raise it, outliers decay it; _bbox_area_factor maps bbox area to a [0.1, 1.0] reliability factor (larger = closer = lower noise). Both feed kf.update. - config: sensor_trust_innovation_threshold + bbox_reference_area Settings (env-wired) + .env.example docs. - README: flip sensor-trust + adaptive-Kalman-by-area to implemented; refresh the status note (remaining planned items are the deferred DeepSORT/compass/threat ring). - Tests: 8 new (innovation return, area-factor effect, trust up/down/clamped, default, area-factor bounds, config wiring). Coverage 59%->60%. ruff/mypy clean, 111 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 22 ++--- backend/.env.example | 7 ++ backend/app/infrastructure/config_adapter.py | 4 + .../app/infrastructure/world_model_adapter.py | 62 +++++++++++-- backend/tests/unit/test_sensor_trust.py | 91 +++++++++++++++++++ 5 files changed, 167 insertions(+), 19 deletions(-) create mode 100644 backend/tests/unit/test_sensor_trust.py diff --git a/README.md b/README.md index fcd7c25..2d7c84f 100644 --- a/README.md +++ b/README.md @@ -103,14 +103,14 @@ that describe them are forward-looking design. | Cross-camera homography ghosts (green `H-PROJ`) | βœ… Implemented | | Pixel-extrapolation ghosts (red `EXTRAP`) | βœ… Implemented | | Appearance re-ID (HSV histograms) wired into tracking / fusion | βœ… Implemented | -| Sensor-trust scoring | πŸ”­ Planned | -| Adaptive Kalman noise by bbox area | πŸ”­ Planned | +| Sensor-trust scoring | βœ… Implemented | +| Adaptive Kalman noise by bbox area | βœ… Implemented | | GPS + IMU fusion into the world model | βœ… Implemented | | DeepSORT / centroid tracker fallback chain | πŸ”­ Planned | -> Config and helper scaffolding for the πŸ”­ items already exists (`compute_appearance()`, -> the `HOMOGRAPHY_*` / `GPS_REFERENCE_*` env vars), but those paths are **not yet active** -> in the pipeline. The build order is tracked in [`docs/superpowers/specs/`](docs/superpowers/specs/). +> The remaining πŸ”­ items (DeepSORT/centroid fallback chain, compass ribbon, threat ring) +> are deferred β€” they're not blockers for the core perception pipeline. The roadmap and +> build order are tracked in [`docs/superpowers/specs/`](docs/superpowers/specs/). --- @@ -124,9 +124,9 @@ that describe them are forward-looking design. | **TensorRT FP16** | `.engine` export on Jetson β€” ~8 MiB, sub-10 ms inference | | **Hungarian tracking** | `scipy.optimize.linear_sum_assignment` β€” cost `0.6 Γ— IoU + 0.4 Γ— cosine appearance`; appearance descriptors are now computed per detection, so the appearance term is active | | **Tracker fallback chain** πŸ”­ | *Planned* β€” only Hungarian (with a greedy fallback) is active today; DeepSORT/Centroid are not implemented | -| **Adaptive Kalman filter** | 6-state `[x, y, z, vx, vy, vz]` β€” measurement noise scales by **confidence** (bbox-area & sensor-trust scaling πŸ”­ planned) | +| **Adaptive Kalman filter** | 6-state `[x, y, z, vx, vy, vz]` β€” measurement noise scales by confidence, bbox area, and sensor trust | | **Cross-camera re-ID** | 64-dim HSV histogram descriptors, L2-normalized, EMA-smoothed (Ξ± = 0.3); computed per detection and used to gate cross-camera association | -| **Sensor trust scoring** πŸ”­ | *Planned* β€” per-sensor trust ∈ [0.1, 1.0]; the Kalman update accepts the param but it is fixed at 1.0 today | +| **Sensor trust scoring** | Per-sensor trust ∈ [0.1, 1.0] β€” rises on consistent measurements, decays on innovation outliers; scales Kalman measurement noise | | **Cross-camera homography** | Self-calibrating ground-plane H from shared foot-point observations via `cv2.findHomography` + RANSAC; projects Path-A green `H-PROJ` ghosts | | **Ghost predictions** | All three paths active β€” Path A (homography/green `H-PROJ`), Path B (pixel extrapolation/red `EXTRAP`), Path C (world projection/orange `WORLD`) | @@ -488,13 +488,13 @@ Objects from different cameras are matched when: - Same `class_id` - Appearance cosine similarity β‰₯ 0.5 when both observations carry a descriptor (differently-dressed people at the same ground position stay separate) -### Sensor trust πŸ”­ *(planned)* +### Sensor trust -The intended design earns per-sensor trust through consistency: -- **Consistent measurements** β†’ trust increases (capped at 1.0) +Each camera/sensor earns trust through consistency: +- **Consistent measurements** (low Kalman innovation) β†’ trust increases (capped at 1.0) - **Innovation outliers** β†’ trust decays (floored at 0.1) -Today the Kalman update accepts a `sensor_trust` argument but it is fixed at `1.0`. +Trust scales the Kalman measurement noise, so flaky sensors are automatically down-weighted. ### Appearance re-ID diff --git a/backend/.env.example b/backend/.env.example index 258cfb4..09126a5 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -85,6 +85,13 @@ APPEARANCE_REID_ENABLED=true # Min cosine similarity [0..1] to merge two cameras' observations into one object. CROSS_CAMERA_APPEARANCE_THRESHOLD=0.5 +# --- Sensor trust + adaptive Kalman --- +# Kalman innovation (metres) above which a measurement is treated as an outlier +# and the source camera's trust decays (consistent measurements raise it). +SENSOR_TRUST_INNOVATION_THRESHOLD=1.0 +# Reference bbox area (px^2); larger detections get lower measurement noise. +BBOX_REFERENCE_AREA=40000.0 + # --- Homography Configuration --- HOMOGRAPHY_MIN_PAIRS=4 HOMOGRAPHY_MAX_PAIRS=100 diff --git a/backend/app/infrastructure/config_adapter.py b/backend/app/infrastructure/config_adapter.py index 145e6d9..11af369 100644 --- a/backend/app/infrastructure/config_adapter.py +++ b/backend/app/infrastructure/config_adapter.py @@ -104,6 +104,10 @@ class Settings(BaseSettings): appearance_reid_enabled: bool = Field(default=True) cross_camera_appearance_threshold: float = Field(default=0.5, ge=0.0, le=1.0) + # Sensor trust + adaptive Kalman + sensor_trust_innovation_threshold: float = Field(default=1.0, ge=0.0) + bbox_reference_area: float = Field(default=40000.0, gt=0.0) + # Homography settings homography_min_pairs: int = Field(default=4, ge=3) homography_max_pairs: int = Field(default=100, ge=10) diff --git a/backend/app/infrastructure/world_model_adapter.py b/backend/app/infrastructure/world_model_adapter.py index 426ca89..752f92e 100644 --- a/backend/app/infrastructure/world_model_adapter.py +++ b/backend/app/infrastructure/world_model_adapter.py @@ -64,29 +64,38 @@ def update( self, measurement: Point3D, confidence: float = 1.0, - sensor_trust: float = 1.0 - ) -> None: - """Update with measurement.""" + sensor_trust: float = 1.0, + area_factor: float = 1.0, + ) -> float: + """Update with a measurement and return the innovation magnitude (metres). + + Measurement noise R scales inversely with a quality factor combining + detection confidence, per-sensor trust, and bbox area (larger bbox = + closer = more reliable depth). Higher quality => smaller R => the filter + trusts the measurement more. + """ # Measurement matrix (we measure position only) H = np.array([ [1, 0, 0, 0, 0, 0], [0, 1, 0, 0, 0, 0], [0, 0, 1, 0, 0, 0] ]) - + # Adaptive measurement noise - quality = max(confidence, 0.1) * max(sensor_trust, 0.1) + quality = max(confidence, 0.1) * max(sensor_trust, 0.1) * max(area_factor, 0.1) R = np.eye(3) * (self.r_base / quality) - + # Kalman gain S = H @ self.covariance @ H.T + R K = self.covariance @ H.T @ np.linalg.inv(S) - + # Update z = np.array([measurement.x, measurement.y, measurement.z]) y = z - H @ self.state + innovation = float(np.linalg.norm(y)) self.state = self.state + K @ y self.covariance = (np.eye(6) - K @ H) @ self.covariance + return innovation @property def position(self) -> Point3D: @@ -282,6 +291,14 @@ def __init__(self, config_repo: ConfigurationRepository): if isinstance(ref_lat, (int, float)) and isinstance(ref_lng, (int, float)): self._gps_ref = (float(ref_lat), float(ref_lng)) + # Per-sensor trust scoring + adaptive Kalman by bbox area + self._sensor_trust: Dict[int, float] = {} + self._trust_innovation_threshold = config_repo.get_float( + "sensor_trust_innovation_threshold", 1.0 + ) + self._trust_step = 0.05 + self._bbox_reference_area = config_repo.get_float("bbox_reference_area", 40000.0) + # Initialize default calibrations if positions provided self._init_default_calibrations() @@ -439,7 +456,15 @@ def _update_existing_object( kf = self._kalman_filters[object_id] dt = max(0.0, (timestamp - obj.last_update).total_seconds()) kf.predict(dt) - kf.update(world_pos, confidence=track.confidence) + trust = self._sensor_trust.get(track.camera_id, 1.0) + area_factor = self._bbox_area_factor(track.bbox.area) + innovation = kf.update( + world_pos, + confidence=track.confidence, + sensor_trust=trust, + area_factor=area_factor, + ) + self._update_sensor_trust(track.camera_id, innovation) obj.position = kf.position obj.velocity = kf.velocity @@ -506,6 +531,27 @@ def _create_new_object( logger.debug(f"Created new world object {object_id}") + def _bbox_area_factor(self, area: float) -> float: + """Map a detection bbox area to a [0.1, 1.0] reliability factor: a larger + bbox means the person is closer, so depth is more reliable and noise lower.""" + if self._bbox_reference_area <= 0: + return 1.0 + return max(0.1, min(1.0, area / self._bbox_reference_area)) + + def _update_sensor_trust(self, camera_id: int, innovation: float) -> None: + """Nudge a camera's trust up on a consistent measurement (low innovation) and + down on an outlier (high innovation), clamped to [0.1, 1.0].""" + trust = self._sensor_trust.get(camera_id, 1.0) + if innovation <= self._trust_innovation_threshold: + trust = min(1.0, trust + self._trust_step) + else: + trust = max(0.1, trust - self._trust_step) + self._sensor_trust[camera_id] = trust + + def get_sensor_trust(self, camera_id: int) -> float: + """Current trust score for a camera (defaults to 1.0 before any update).""" + return self._sensor_trust.get(camera_id, 1.0) + def _blend_appearance( self, old: Optional[AppearanceDescriptor], diff --git a/backend/tests/unit/test_sensor_trust.py b/backend/tests/unit/test_sensor_trust.py new file mode 100644 index 0000000..c9e7ac3 --- /dev/null +++ b/backend/tests/unit/test_sensor_trust.py @@ -0,0 +1,91 @@ +"""Phase B5 β€” sensor trust scoring + adaptive Kalman by bbox area.""" +from unittest.mock import Mock + +import pytest + +pytest.importorskip("cv2") + +from app.domain.entities import Point3D +from app.infrastructure.world_model_adapter import KalmanFilter + + +def _repo(): + from app.infrastructure.world_model_adapter import WorldModelRepositoryImpl + config = Mock() + config.get.return_value = {} + config.get_int.return_value = 4 + config.get_float.return_value = 1.0 + config.get_list.return_value = [] + repo = WorldModelRepositoryImpl(config) + repo._trust_innovation_threshold = 1.0 + repo._bbox_reference_area = 40000.0 + return repo + + +# --------------------------------------------------------- adaptive Kalman + +def test_update_returns_innovation_magnitude(): + kf = KalmanFilter() + kf.state[0:3] = [0.0, 0.0, 0.0] + inn = kf.update(Point3D(3.0, 4.0, 0.0)) # |(3,4,0)| = 5 + assert inn == pytest.approx(5.0, abs=1e-6) + + +def test_higher_area_factor_trusts_measurement_more(): + hi = KalmanFilter() + hi.state[0:3] = [0, 0, 0] + lo = KalmanFilter() + lo.state[0:3] = [0, 0, 0] + hi.update(Point3D(10, 0, 0), area_factor=1.0) + lo.update(Point3D(10, 0, 0), area_factor=0.1) + assert hi.position.x > lo.position.x + + +# ------------------------------------------------------------ sensor trust + +def test_consistent_measurement_increases_trust(): + repo = _repo() + repo._sensor_trust[0] = 0.5 + repo._update_sensor_trust(0, innovation=0.01) + assert repo._sensor_trust[0] > 0.5 + + +def test_outlier_decreases_trust(): + repo = _repo() + repo._sensor_trust[0] = 0.5 + repo._update_sensor_trust(0, innovation=100.0) + assert repo._sensor_trust[0] < 0.5 + + +def test_trust_clamped_to_unit_interval(): + repo = _repo() + repo._sensor_trust[0] = 1.0 + for _ in range(10): + repo._update_sensor_trust(0, innovation=0.0) + assert repo._sensor_trust[0] <= 1.0 + repo._sensor_trust[1] = 0.1 + for _ in range(10): + repo._update_sensor_trust(1, innovation=1000.0) + assert repo._sensor_trust[1] >= 0.1 + + +def test_default_trust_is_one(): + assert _repo().get_sensor_trust(99) == 1.0 + + +def test_bbox_area_factor_bounds(): + repo = _repo() + assert repo._bbox_area_factor(0.0) == pytest.approx(0.1) + assert repo._bbox_area_factor(1e9) == pytest.approx(1.0) + assert 0.1 < repo._bbox_area_factor(20000.0) < 1.0 + + +def test_trust_config_wired(): + from app.infrastructure.config_adapter import ( + PydanticConfigurationRepository, Settings, + ) + repo = PydanticConfigurationRepository( + Settings(sensor_trust_innovation_threshold=2.5, bbox_reference_area=10000.0) + ) + assert repo.get_float("sensor_trust_innovation_threshold", 1.0) == 2.5 + assert repo.get_float("bbox_reference_area", 40000.0) == 10000.0