"""Traffic analytics engine. Responsibilities: - Vehicle counting when tracks cross named counting lines - Speed estimation from pixel displacement × calibration factor - Per-class breakdown - Anomaly detection (count spikes, stopped vehicles) - Confidence drift monitoring """ from __future__ import annotations import time from collections import defaultdict, deque from dataclasses import dataclass, field from typing import Any import numpy as np from core.tracker import Track # ── Data structures ─────────────────────────────────────────────────────────── @dataclass class CountingLineSpec: """Specification for a named counting line.""" name: str p1: tuple[int, int] p2: tuple[int, int] @dataclass class SpeedSample: track_id: int speed_kmh: float class_id: int timestamp: float @dataclass class FrameMetrics: """Analytics snapshot for one frame — sent over WebSocket.""" timestamp: float frame_index: int # Counts total_count: int # vehicles counted crossing all lines (summed) count_per_class: dict[str, int] # cumulative per class (from first line) count_per_line: dict[str, int] # cumulative count per named line vehicles_in_frame: int # currently visible tracks # Speed avg_speed_kmh: float # rolling average over last N samples speed_samples: list[float] # recent speed readings (last 10) # Alerts alerts: list[str] # Raw tracks for frontend overlay tracks: list[dict[str, Any]] @dataclass class Alert: message: str severity: str # "info" | "warning" | "critical" timestamp: float = field(default_factory=time.time) # ── Counting line ───────────────────────────────────────────────────────────── class CountingLine: """A line segment that counts vehicles crossing it. The line is defined by two points (x1,y1)→(x2,y2) in pixel coords. A crossing is detected when a track's centre moves from one side to the other. """ def __init__(self, p1: tuple[int, int], p2: tuple[int, int]) -> None: self.p1 = np.array(p1, dtype=float) self.p2 = np.array(p2, dtype=float) self._prev_sides: dict[int, float] = {} self.total_count: int = 0 self.count_per_class: dict[str, int] = defaultdict(int) def update(self, tracks: list[Track]) -> list[Track]: """Check each track for a crossing. Returns newly-crossed tracks.""" crossed = [] current_ids = {t.track_id for t in tracks} for track in tracks: cx, cy = track.center side = self._side(cx, cy) prev = self._prev_sides.get(track.track_id) if prev is not None and prev != 0 and side != 0 and prev != side: self.total_count += 1 self.count_per_class[track.class_name] += 1 crossed.append(track) if side != 0: self._prev_sides[track.track_id] = side # Clean up lost tracks for tid in list(self._prev_sides): if tid not in current_ids: del self._prev_sides[tid] return crossed def _side(self, px: float, py: float) -> float: dx = self.p2[0] - self.p1[0] dy = self.p2[1] - self.p1[1] return float(np.sign(dx * (py - self.p1[1]) - dy * (px - self.p1[0]))) # ── Speed estimator ────────────────────────────────────────────────────────── class SpeedEstimator: """Estimates speed from pixel displacement between frames.""" def __init__( self, pixels_per_meter: float = 10.0, fps: float = 25.0, smoothing_window: int = 5, ) -> None: self.pixels_per_meter = pixels_per_meter self.fps = fps self.smoothing_window = smoothing_window self._history: dict[int, deque[tuple[float, float]]] = defaultdict( lambda: deque(maxlen=smoothing_window) ) def update(self, tracks: list[Track]) -> list[SpeedSample]: samples = [] current_ids = {t.track_id for t in tracks} for track in tracks: hist = self._history[track.track_id] hist.append(track.center) if len(hist) >= 2: positions = list(hist) total_px = sum( np.linalg.norm(np.array(positions[i + 1]) - np.array(positions[i])) for i in range(len(positions) - 1) ) px_per_frame = total_px / (len(positions) - 1) speed_kmh = (px_per_frame / self.pixels_per_meter) * self.fps * 3.6 samples.append( SpeedSample( track_id=track.track_id, speed_kmh=round(speed_kmh, 1), class_id=track.class_id, timestamp=time.time(), ) ) for tid in list(self._history): if tid not in current_ids: del self._history[tid] return samples # ── Anomaly detector ────────────────────────────────────────────────────────── class AnomalyDetector: """Threshold-based anomaly detection. - Count spike: current vehicle count > mean + 2σ of rolling window - Stopped vehicle: track barely moves for N consecutive frames """ SPIKE_WINDOW = 60 SPIKE_Z_THRESHOLD = 2.0 STOPPED_MIN_FRAMES = 30 STOPPED_PX_THRESHOLD = 5.0 def __init__(self) -> None: self._count_history: deque[int] = deque(maxlen=self.SPIKE_WINDOW) self._stationary_frames: dict[int, int] = defaultdict(int) self._last_centers: dict[int, tuple[float, float]] = {} def update(self, tracks: list[Track]) -> list[Alert]: alerts: list[Alert] = [] current_count = len(tracks) current_ids = {t.track_id for t in tracks} # Count spike self._count_history.append(current_count) if len(self._count_history) == self.SPIKE_WINDOW: arr = np.array(self._count_history) mean, std = arr.mean(), arr.std() if std > 0 and current_count > mean + self.SPIKE_Z_THRESHOLD * std: alerts.append(Alert( message=( f"Traffic surge: {current_count} vehicles " f"(mean={mean:.1f}, +{self.SPIKE_Z_THRESHOLD}σ)" ), severity="warning", )) # Stopped vehicles for track in tracks: cx, cy = track.center last = self._last_centers.get(track.track_id) if last is not None: dist = np.linalg.norm(np.array([cx, cy]) - np.array(last)) if dist < self.STOPPED_PX_THRESHOLD: self._stationary_frames[track.track_id] += 1 else: self._stationary_frames[track.track_id] = 0 self._last_centers[track.track_id] = (cx, cy) if self._stationary_frames[track.track_id] == self.STOPPED_MIN_FRAMES: alerts.append(Alert( message=f"Stopped vehicle: ID {track.track_id} ({track.class_name})", severity="info", )) for tid in list(self._stationary_frames): if tid not in current_ids: del self._stationary_frames[tid] self._last_centers.pop(tid, None) return alerts # ── Drift monitor ───────────────────────────────────────────────────────────── class DriftMonitor: """Monitors detection confidence for model drift. Establishes a baseline mean confidence over the first BASELINE_FRAMES frames, then alerts when the rolling average drops more than THRESHOLD below baseline. Useful for catching model degradation in production. """ BASELINE_FRAMES = 100 # frames to collect before locking in baseline WINDOW = 30 # rolling window for current mean THRESHOLD = 0.15 # relative drop fraction that triggers an alert ALERT_COOLDOWN = 150 # minimum frames between repeated drift alerts def __init__(self) -> None: self._baseline_samples: list[float] = [] self._baseline: float | None = None self._window: deque[float] = deque(maxlen=self.WINDOW) self._frames_since_alert: int = self.ALERT_COOLDOWN def update(self, tracks: list[Track]) -> list[Alert]: if not tracks: return [] avg_conf = sum(t.confidence for t in tracks) / len(tracks) if self._baseline is None: self._baseline_samples.append(avg_conf) if len(self._baseline_samples) >= self.BASELINE_FRAMES: self._baseline = sum(self._baseline_samples) / len(self._baseline_samples) print(f"[DriftMonitor] Baseline confidence locked: {self._baseline:.3f}") return [] self._window.append(avg_conf) self._frames_since_alert += 1 if len(self._window) < self.WINDOW: return [] rolling_mean = sum(self._window) / len(self._window) threshold = self._baseline * (1.0 - self.THRESHOLD) if rolling_mean < threshold and self._frames_since_alert >= self.ALERT_COOLDOWN: drop_pct = (self._baseline - rolling_mean) / self._baseline * 100 self._frames_since_alert = 0 return [Alert( message=( f"Confidence drift: rolling avg {rolling_mean:.2f} " f"vs baseline {self._baseline:.2f} ({drop_pct:.0f}% drop)" ), severity="warning", )] return [] # ── Main analytics engine ──────────────────────────────────────────────────── class AnalyticsEngine: """Orchestrates counting, speed, anomaly detection, and drift monitoring. Usage:: engine = AnalyticsEngine( counting_lines=[ CountingLineSpec("north", (640, 0), (640, 360)), CountingLineSpec("south", (640, 360), (640, 720)), ] ) metrics = engine.update(tracks) """ SPEED_HISTORY_LEN = 50 def __init__( self, counting_lines: list[CountingLineSpec] | None = None, pixels_per_meter: float = 10.0, fps: float = 25.0, ) -> None: # Build named counting lines self._lines: list[tuple[str, CountingLine]] = [] for spec in (counting_lines or []): self._lines.append((spec.name, CountingLine(spec.p1, spec.p2))) self.speed_estimator = SpeedEstimator(pixels_per_meter=pixels_per_meter, fps=fps) self.anomaly_detector = AnomalyDetector() self.drift_monitor = DriftMonitor() self._speed_history: deque[float] = deque(maxlen=self.SPEED_HISTORY_LEN) self._recent_alerts: deque[Alert] = deque(maxlen=20) self._frame_index: int = 0 def update(self, tracks: list[Track]) -> FrameMetrics: """Process one frame's tracks and return an analytics snapshot.""" # ── Multi-line counting ─────────────────────────────────────────────── count_per_line: dict[str, int] = {} total_count = 0 count_per_class: dict[str, int] = {} for name, line in self._lines: line.update(tracks) count_per_line[name] = line.total_count total_count += line.total_count # Use the first line's per-class breakdown for backward compat if self._lines: count_per_class = dict(self._lines[0][1].count_per_class) # ── Speed ───────────────────────────────────────────────────────────── speed_samples = self.speed_estimator.update(tracks) for s in speed_samples: if s.speed_kmh < 200: self._speed_history.append(s.speed_kmh) avg_speed = float(np.mean(list(self._speed_history))) if self._speed_history else 0.0 recent_speeds = [s.speed_kmh for s in speed_samples[-10:]] # ── Anomaly + drift ─────────────────────────────────────────────────── new_alerts: list[Alert] = [] new_alerts.extend(self.anomaly_detector.update(tracks)) new_alerts.extend(self.drift_monitor.update(tracks)) self._recent_alerts.extend(new_alerts) alert_strings = [ f"[{a.severity.upper()}] {a.message}" for a in list(self._recent_alerts)[-5:] ] # ── Serialise tracks ────────────────────────────────────────────────── speed_map = {s.track_id: s.speed_kmh for s in speed_samples} tracks_data = [ { "id": t.track_id, "bbox": list(t.bbox), "class_id": t.class_id, "class_name": t.class_name, "confidence": round(t.confidence, 3), "speed_kmh": speed_map.get(t.track_id), } for t in tracks ] self._frame_index += 1 return FrameMetrics( timestamp=time.time(), frame_index=self._frame_index, total_count=total_count, count_per_class=count_per_class, count_per_line=count_per_line, vehicles_in_frame=len(tracks), avg_speed_kmh=round(avg_speed, 1), speed_samples=recent_speeds, alerts=alert_strings, tracks=tracks_data, ) @property def cumulative_counts(self) -> dict[str, int]: if self._lines: return dict(self._lines[0][1].count_per_class) return {} @property def line_specs(self) -> list[CountingLineSpec]: """Return the counting line specs for serialisation.""" out = [] for name, line in self._lines: out.append(CountingLineSpec( name=name, p1=(int(line.p1[0]), int(line.p1[1])), p2=(int(line.p2[0]), int(line.p2[1])), )) return out