Advanced AI, Command Center UI redesign, ALPR/audio, and engagement README

backend/agents/consolidated/detector.py

@@ -165,27 +165,52 @@ async def _pipeline_lpr(self, camera_id: str, cam: dict, objects: list[dict]) ->
         if not vehicles:
             return None
         cam_name = cam.get("name", camera_id)
-        result = await self.execute_tool_loop(
-            prompt=(
-                f"Analyze current frame from camera {camera_id} ({cam_name}). "
-                f"{len(vehicles)} vehicle(s) detected. Use analyze_frame_with_gemini to "
-                f"read all visible license plates. For each plate report: plate_text, "
-                f"vehicle_type, vehicle_color, confidence. Then use store_observation to "
-                f"record each reading with category='plate_read'."
-            ),
-            context_data={"camera_id": camera_id, "vehicle_count": len(vehicles), "task": "lpr"},
-        )
-        resp = result.get("response", "")
+
+        # Real local ALPR (EasyOCR) — fast, deterministic plate reads instead of
+        # the slow/inconsistent vision-model path. Each vehicle box is OCR'd and
+        # checked against active vehicle BOLOs.
+        from backend.services.alpr_service import alpr_service
+        if not alpr_service.available():
+            return None
+        from backend.services.video_capture import capture_manager
+        stream = capture_manager.get_stream(camera_id)
+        latest = stream.get_latest_frame() if stream else None
+        if latest is None:
+            return None
+        _, frame = latest
+
+        reads: list[dict] = []
+        bolo_hits: list[dict] = []
+        for v in vehicles:
+            bbox = v.get("bbox")
+            if not bbox:
+                continue
+            r = await alpr_service.read_and_match(frame, bbox)
+            if r.get("plate"):
+                reads.append({"plate": r["plate"], "confidence": r.get("confidence"), "track_id": v.get("track_id")})
+                if r.get("matches"):
+                    bolo_hits.append({"plate": r["plate"], "matches": r["matches"], "track_id": v.get("track_id")})
+        if not reads:
+            return None
+
         await self.send_message(CH_PERCEPTIONS, {
             "type": "plate_read", "camera_id": camera_id, "camera_name": cam_name,
-            "vehicle_count": len(vehicles), "analysis": resp[:500],
+            "plates": reads, "bolo_hits": bolo_hits,
             "timestamp": datetime.now(timezone.utc).isoformat(),
         })
+        for hit in bolo_hits:
+            await self.send_message(CH_PERCEPTIONS, {
+                "type": "bolo_vehicle_match", "camera_id": camera_id, "camera_name": cam_name,
+                "plate": hit["plate"], "severity": "high",
+                "reason": "; ".join(m.get("reason", "BOLO vehicle") for m in hit["matches"]),
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+            })
         await self.log_action("lpr_scan", {
             "camera_id": camera_id, "vehicles_detected": len(vehicles),
-            "decision": f"LPR scan {cam_name}: {len(vehicles)} vehicle(s)",
+            "decision": f"ALPR {cam_name}: read {len(reads)} plate(s), {len(bolo_hits)} BOLO hit(s)",
         })
-        return {"vehicles_scanned": len(vehicles), "response": resp[:300]}
+        return {"vehicles_scanned": len(vehicles), "plates_read": len(reads),
+                "bolo_hits": len(bolo_hits), "plates": [r["plate"] for r in reads]}
 
     # ── Pipeline: PPE Compliance ──────────────────────────────────
 

backend/agents/monitoring_agent.py

@@ -46,6 +46,8 @@
 
 # AI vision is expensive; only invoke it every N frames per camera.
 _AI_EVERY_N_FRAMES = 15
+# Refresh learned adaptive thresholds from the baseline service this often.
+_THRESHOLD_REFRESH_EVERY_N = 300
 # Minimum seconds between full pipeline runs for a single camera.
 _MIN_INTERVAL_SECONDS = 1.0
 
@@ -101,6 +103,19 @@ async def process_frame(
         counter = self._frame_counters.get(camera_id, 0) + 1
         self._frame_counters[camera_id] = counter
 
+        # Refresh learned adaptive thresholds + active BOLOs into sync caches (throttled).
+        if counter % _THRESHOLD_REFRESH_EVERY_N == 1:
+            try:
+                from backend.services.adaptive_thresholds import adaptive_thresholds
+                zid = zone_info.get("id") if zone_info else None
+                async with async_session() as _db:
+                    await adaptive_thresholds.refresh(_db, camera_id, zid)
+                    if getattr(settings, "BOLO_REALTIME_ENABLED", True):
+                        from backend.services.bolo_matcher import bolo_matcher
+                        await bolo_matcher.refresh(_db)
+            except Exception as exc:  # noqa: BLE001
+                logger.debug("threshold/bolo refresh failed for %s: %s", camera_id, exc)
+
         if settings.VISION_VERIFIED_DETECTION:
             # Verified-vision path: structured scene intelligence + an adversarial
             # verifier. Hallucination-resistant — only threats a skeptic confirms
@@ -145,6 +160,25 @@ async def process_frame(
             except Exception as exc:  # noqa: BLE001
                 logger.debug("pose behaviours failed for %s: %s", camera_id, exc)
 
+        # ── 3a-quater. Real-time BOLO appearance matching ────────
+        # Embed each new person track once and match against active person BOLOs.
+        if getattr(settings, "BOLO_REALTIME_ENABLED", True):
+            try:
+                from backend.services.bolo_matcher import bolo_matcher
+                if bolo_matcher.has_active():
+                    threats.extend(bolo_matcher.scan_frame(frame, detections, camera_id))
+            except Exception as exc:  # noqa: BLE001
+                logger.debug("BOLO scan failed for %s: %s", camera_id, exc)
+
+        # ── 3a-quinquies. Threat-escalation chains ───────────────
+        # Detect escalating behaviour SEQUENCES on a single entity (e.g.
+        # loitering → running → fall) across the track-bearing threats above.
+        try:
+            from backend.services.escalation_tracker import escalation_tracker
+            threats.extend(escalation_tracker.observe(camera_id, threats, timestamp_epoch))
+        except Exception as exc:  # noqa: BLE001
+            logger.debug("escalation tracking failed for %s: %s", camera_id, exc)
+
         # ── 3a-ter. SAM2 mask enrichment for FLAGGED objects ──────
         # Only objects referenced by a threat get a pixel-precise SAM2 mask
         # (occlusion-robust extent), attached to the threat + its detection so it

backend/api/bolo.py

@@ -200,6 +200,41 @@ async def deactivate_bolo(
         raise HTTPException(status_code=500, detail="Failed to deactivate BOLO entry")
 
 
+class EnrollAppearanceRequest(BaseModel):
+    image_base64: str = Field(..., description="Reference image (base64 JPEG/PNG)")
+    bbox: Optional[List[float]] = Field(None, description="[x1,y1,x2,y2]; whole image if omitted")
+
+
+@router.post("/{bolo_id}/enroll-appearance")
+async def enroll_appearance(bolo_id: uuid.UUID, body: EnrollAppearanceRequest, _user=Depends(get_current_user)):
+    """Enroll a person BOLO with a CLIP appearance embedding from a reference
+    image so the real-time matcher can flag this person across cameras."""
+    import base64
+    import cv2
+    import numpy as np
+    from backend.services.appearance_embedder import appearance_embedding
+    from backend.services.bolo_service import bolo_service
+
+    try:
+        raw = base64.b64decode(body.image_base64.split(",", 1)[-1])
+        frame = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
+    except Exception:
+        raise HTTPException(status_code=400, detail="Invalid image_base64")
+    if frame is None:
+        raise HTTPException(status_code=400, detail="Could not decode image")
+
+    h, w = frame.shape[:2]
+    bbox = body.bbox or [0, 0, w, h]
+    emb = appearance_embedding(frame, bbox)
+    if not emb:
+        raise HTTPException(status_code=422, detail="Could not compute appearance embedding")
+
+    ok = await bolo_service.enroll_appearance(bolo_id, emb)
+    if not ok:
+        raise HTTPException(status_code=404, detail="BOLO not found")
+    return {"enrolled": True, "bolo_id": str(bolo_id), "embedding_dim": len(emb)}
+
+
 @router.get("/{bolo_id}/sightings", response_model=List[dict])
 async def get_bolo_sightings(
     bolo_id: uuid.UUID,

backend/api/compliance_dashboard.py

@@ -53,6 +53,20 @@ async def compliance_history(
         raise HTTPException(400, str(e))
 
 
+@router.get("/forecast")
+async def compliance_forecast(
+    framework: str = "gdpr",
+    target: float = Query(0.8, ge=0.0, le=1.0),
+    db: AsyncSession = Depends(get_db),
+):
+    """Project the compliance score from its recent trend and estimate when it
+    will breach the target — enabling proactive remediation."""
+    try:
+        return await compliance_dashboard_service.forecast_compliance(db, framework=framework, target=target)
+    except Exception as e:
+        raise HTTPException(400, str(e))
+
+
 @router.get("/issues")
 async def compliance_issues(
     severity: str = None,

backend/api/forensics.py

@@ -485,6 +485,40 @@ async def subject_search(
     }
 
 
+class ImageSearchRequest(BaseModel):
+    image_base64: str = Field(..., description="Query image (base64 JPEG/PNG)")
+    bbox: Optional[List[float]] = Field(None, description="[x1,y1,x2,y2] crop; whole image if omitted")
+    camera_id: Optional[str] = None
+    max_results: int = Field(20, ge=1, le=100)
+
+
+@router.post("/search-by-image")
+async def search_by_image(
+    body: ImageSearchRequest,
+    _user=Depends(require_role(UserRole.ANALYST)),
+):
+    """Image-based "looks-like" forensic search: upload an image (or a crop) and
+    find the most visually similar people/objects seen across all cameras, using
+    CLIP embeddings against the object-crop vector store.
+    """
+    import base64
+    import cv2
+    import numpy as np
+    from backend.services.forensic_search_service import forensic_search_service
+
+    try:
+        raw = base64.b64decode(body.image_base64.split(",", 1)[-1])
+        frame = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
+    except Exception:
+        raise HTTPException(status_code=400, detail="Invalid image_base64")
+    if frame is None:
+        raise HTTPException(status_code=400, detail="Could not decode image")
+
+    return await forensic_search_service.search_objects_by_image(
+        frame, bbox=body.bbox, camera_id=body.camera_id, top_k=body.max_results,
+    )
+
+
 @router.post("/movement-trail")
 async def movement_trail(
     body: MovementTrailRequest,

backend/api/intelligence.py

@@ -160,6 +160,50 @@ async def deep_analyze(body: SceneAnalyzeRequest):
     }
 
 
+class AnalyzeAudioRequest(BaseModel):
+    audio_base64: str = Field(..., description="WAV audio (base64)")
+
+
+@router.post("/analyze-audio")
+async def analyze_audio(body: AnalyzeAudioRequest):
+    """Detect security-relevant sound events (gunshot/glass/scream/alarm) in a WAV
+    clip via the local audio DSP engine."""
+    import base64
+    from backend.services.audio_detection_service import audio_detection_service
+    try:
+        raw = base64.b64decode(body.audio_base64.split(",", 1)[-1])
+    except Exception:
+        raise HTTPException(status_code=400, detail="Invalid audio_base64")
+    events = audio_detection_service.analyze_bytes(raw)
+    return {"event_count": len(events), "events": events}
+
+
+class ReadPlateRequest(BaseModel):
+    image_base64: str = Field(..., description="Image containing a plate (base64)")
+    bbox: Optional[List[float]] = Field(None, description="Optional vehicle/plate crop [x1,y1,x2,y2]")
+
+
+@router.post("/read-plate")
+async def read_plate(body: ReadPlateRequest):
+    """Read a license plate via the local ALPR (EasyOCR) engine and check it
+    against active vehicle BOLOs."""
+    import base64
+    import cv2
+    import numpy as np
+    from backend.services.alpr_service import alpr_service
+
+    try:
+        raw = base64.b64decode(body.image_base64.split(",", 1)[-1])
+        frame = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
+    except Exception:
+        raise HTTPException(status_code=400, detail="Invalid image_base64")
+    if frame is None:
+        raise HTTPException(status_code=400, detail="Could not decode image")
+    if not alpr_service.available():
+        raise HTTPException(status_code=503, detail="ALPR engine unavailable")
+    return await alpr_service.read_and_match(frame, body.bbox)
+
+
 class SegmentRequest(BaseModel):
     camera_id: Optional[str] = None
     image_base64: Optional[str] = None

backend/config.py

@@ -125,6 +125,18 @@ class Settings(BaseSettings):
     # this many objects per frame to bound cost.
     SAM2_MAX_OBJECTS: int = 5
 
+    # ── Real-time BOLO appearance matching ─────────────────
+    BOLO_REALTIME_ENABLED: bool = True
+    BOLO_MATCH_THRESHOLD: float = 0.82
+
+    # ── ALPR (license-plate OCR via EasyOCR) ───────────────
+    ALPR_ENABLED: bool = True
+    ALPR_GPU: bool = False          # CPU by default to avoid GPU contention with Ollama
+    ALPR_MIN_CONFIDENCE: float = 0.4
+
+    # ── Audio event detection (DSP; pluggable deep model) ──
+    AUDIO_DETECTION_ENABLED: bool = True
+
     # ── Qdrant ────────────────────────────────────────────────
     QDRANT_HOST: str = "localhost"
     QDRANT_PORT: int = 6333

backend/services/adaptive_thresholds.py

@@ -0,0 +1,58 @@
+"""Adaptive thresholds cache — wires learned baselines into the hot loop.
+
+`baseline_learning_service.get_adaptive_thresholds` returns per-camera/zone
+thresholds (mean + 2*std of learned-normal), but it is async and hits the DB —
+too expensive to call per frame from the synchronous detection loop. This cache
+bridges the gap: an async `refresh()` (called on a throttle from the monitoring
+agent, which already holds a DB session) populates an in-memory table, and a
+fast sync `get()` lets hot-loop code (e.g. temporal_behaviour) read a learned
+threshold, transparently falling back to the caller's hard-coded default when no
+fresh baseline exists.
+"""
+from __future__ import annotations
+
+import logging
+import time
+import uuid
+from typing import Any, Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+_TTL_SECONDS = 900.0  # a cached threshold older than this is treated as absent
+
+
+class AdaptiveThresholds:
+    def __init__(self) -> None:
+        # camera_id(str) -> {metric: value, "_t": fetched_at}
+        self._cache: Dict[str, Dict[str, Any]] = {}
+
+    def get(self, camera_id: str, key: str, default: float) -> float:
+        """Return a fresh learned threshold for the camera, else `default`."""
+        entry = self._cache.get(str(camera_id))
+        if not entry:
+            return default
+        if time.time() - entry.get("_t", 0.0) > _TTL_SECONDS:
+            return default
+        val = entry.get(key)
+        return float(val) if isinstance(val, (int, float)) else default
+
+    def source(self, camera_id: str) -> str:
+        entry = self._cache.get(str(camera_id))
+        if not entry or time.time() - entry.get("_t", 0.0) > _TTL_SECONDS:
+            return "default"
+        return entry.get("source", "default")
+
+    async def refresh(self, db, camera_id: str, zone_id: Optional[str] = None) -> None:
+        """Pull the latest learned thresholds for a camera into the cache."""
+        try:
+            from backend.services.baseline_learning_service import baseline_learning_service
+            cam_uuid = uuid.UUID(str(camera_id))
+            zone_uuid = uuid.UUID(str(zone_id)) if zone_id else None
+            thr = await baseline_learning_service.get_adaptive_thresholds(db, cam_uuid, zone_uuid)
+            thr["_t"] = time.time()
+            self._cache[str(camera_id)] = thr
+        except Exception as exc:  # noqa: BLE001 — never let threshold refresh break the loop
+            logger.debug("adaptive_thresholds.refresh failed for %s: %s", camera_id, exc)
+
+
+adaptive_thresholds = AdaptiveThresholds()