| 1 | """Tests for the ALPR movement-intelligence engine.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | from datetime import datetime, timedelta, timezone |
| 5 | |
| 6 | from app.services.analytics import ( |
| 7 | camera_transitions, detect_dwell, find_cotravel, first_last_seen, |
| 8 | movement_summary, possible_duplicate_plates, reconstruct_route, |
| 9 | ) |
| 10 | |
| 11 | UTC = timezone.utc |
| 12 | |
| 13 | |
| 14 | def _r(plate, cam, minute, lat=None, lon=None): |
| 15 | return {"plate_norm": plate, "camera_id": cam, |
| 16 | "read_ts": datetime(2026, 6, 1, 8, minute, 0, tzinfo=UTC), |
| 17 | "lat": lat, "lon": lon, "confidence": 0.95} |
| 18 | |
| 19 | |
| 20 | def test_first_last_seen(): |
| 21 | reads = [_r("A", "c1", 0), _r("A", "c2", 10), _r("A", "c1", 5)] |
| 22 | fl = first_last_seen(reads) |
| 23 | assert fl["count"] == 3 |
| 24 | assert fl["first_seen"].minute == 0 |
| 25 | assert fl["last_seen"].minute == 10 |
| 26 | |
| 27 | |
| 28 | def test_route_reconstruction_and_speed(): |
| 29 | reads = [_r("A", "c1", 0, 35.1495, -90.0490), _r("A", "c2", 12, 35.1180, -89.9710)] |
| 30 | legs = reconstruct_route(reads) |
| 31 | assert len(legs) == 1 |
| 32 | assert legs[0].distance_km and legs[0].distance_km > 5 |
| 33 | assert not legs[0].implausible |
| 34 | |
| 35 | |
| 36 | def test_impossible_travel_flagged(): |
| 37 | reads = [_r("A", "c1", 0, 35.1495, -90.0490), _r("A", "c2", 1, 35.1180, -89.9710)] |
| 38 | anomalies = possible_duplicate_plates(reads, impossible_speed_kmh=250.0) |
| 39 | assert len(anomalies) == 1 |
| 40 | assert anomalies[0]["reason"] == "impossible_travel_speed" |
| 41 | |
| 42 | |
| 43 | def test_dwell_detection(): |
| 44 | reads = [_r("A", "c1", 0), _r("A", "c1", 5), _r("A", "c1", 10), _r("A", "c2", 40)] |
| 45 | episodes = detect_dwell(reads, gap_seconds=1800) |
| 46 | dwell_c1 = [e for e in episodes if e.camera_id == "c1"] |
| 47 | assert dwell_c1 and dwell_c1[0].reads == 3 |
| 48 | assert dwell_c1[0].duration_s == 600 |
| 49 | |
| 50 | |
| 51 | def test_camera_transitions(): |
| 52 | by_vehicle = { |
| 53 | "A": [_r("A", "c1", 0), _r("A", "c2", 5)], |
| 54 | "B": [_r("B", "c1", 1), _r("B", "c2", 6)], |
| 55 | } |
| 56 | edges = camera_transitions(by_vehicle) |
| 57 | assert {"from": "c1", "to": "c2", "weight": 2} in edges |
| 58 | |
| 59 | |
| 60 | def test_cotravel(): |
| 61 | by_vehicle = { |
| 62 | "A": [_r("A", "c1", 0), _r("A", "c2", 10)], |
| 63 | "B": [_r("B", "c1", 0), _r("B", "c2", 10)], |
| 64 | "C": [_r("C", "c3", 30)], |
| 65 | } |
| 66 | pairs = find_cotravel(by_vehicle, "A", window_seconds=120, min_shared=2) |
| 67 | assert pairs and pairs[0]["plate_norm"] == "B" |
| 68 | assert pairs[0]["shared_sightings"] == 2 |
| 69 | |
| 70 | |
| 71 | def test_movement_summary_integration(): |
| 72 | reads = [_r("A", "c1", 0, 35.1495, -90.0490), _r("A", "c2", 12, 35.1180, -89.9710), |
| 73 | _r("A", "c1", 600 % 60 or 30, 35.1495, -90.0490)] |
| 74 | m = movement_summary("A", reads) |
| 75 | assert m.plate_norm == "A" |
| 76 | assert m.total_reads == 3 |
| 77 | assert m.distinct_cameras == 2 |
| 78 | assert isinstance(m.by_hour, dict) |