""" face_live_search.py ──────────────────── Search for a query face (from an uploaded image) across every active live camera feed by comparing embedding similarity — without re-running detection on the live frame. The live camera results stored in `vision_engine.face_results` include the `embedding` field when produced by the local (non-cloud) vision engine. In cloud/browser mode we fall back to comparing the uploaded query frame against the latest raw frame for each camera via `match_frame`. """ from __future__ import annotations import logging import os from typing import Any import cv2 import numpy as np logger = logging.getLogger(__name__) DEFAULT_THRESHOLD = float(os.environ.get("FACE_MATCH_THRESHOLD", "0.22")) def _cosine(a: np.ndarray, b: np.ndarray) -> float: na, nb = np.linalg.norm(a), np.linalg.norm(b) if na == 0 or nb == 0: return 0.0 return float(np.dot(a, b) / (na * nb)) def _is_known_identity(name: str | None) -> bool: if not name: return False lowered = str(name).strip().lower() return lowered not in ("unknown", "unidentified", "none", "") and not lowered.startswith("unknown_") def search_query_in_live_feeds( query_frame: np.ndarray, face_engine, vision_engine, threshold: float | None = None, ) -> dict[str, Any]: """ Compare the face in `query_frame` against every live camera feed. Strategy -------- 1. Extract the embedding from the query image (the uploaded photo). 2. For each camera that has active face results: a. If the stored face result includes an `embedding`, compare directly. b. Otherwise fall back to `match_frame` on the latest raw frame (if the engine exposes `latest_raw_frames`). 3. Return the best matching camera, confidence, and which name was assigned to the matched face in the live stream (so the UI can show the correct unknown_N or known name). """ try: from Face_Recognition.face_matcher import _cosine as _c, DEFAULT_THRESHOLD except ImportError: from face_matcher import _cosine as _c, DEFAULT_THRESHOLD # type: ignore[no-redef] thresh = threshold if threshold is not None else DEFAULT_THRESHOLD if face_engine is None or getattr(face_engine, "app", None) is None: return { "found": False, "reason": "Face recognition engine unavailable.", "cameras_searched": 0, } # ── Step 1: Extract query embedding ────────────────────────────────────── query_face = None with face_engine.lock: faces = face_engine.app.get(query_frame) if faces: query_face = max(faces, key=lambda f: (f.bbox[2] - f.bbox[0]) * (f.bbox[3] - f.bbox[1])) if query_face is None: return {"found": False, "reason": "No face detected in the uploaded image."} query_emb = query_face.embedding # ── Step 2: Search every live camera ───────────────────────────────────── live_face_results: dict = {} try: live_face_results = vision_engine.face_results or {} except Exception: pass best_cam = None best_score = 0.0 best_name = None best_face_hit: dict[str, Any] | None = None best_raw_frame = None cameras_searched = 0 for cam_id, face_list in live_face_results.items(): if not face_list: continue cameras_searched += 1 raw_frame = None try: raw_frame = getattr(vision_engine, "latest_raw_frames", {}).get(cam_id) except Exception: pass for face_hit in face_list: hit_name = face_hit.get("name", "Unknown") emb = face_hit.get("embedding") # Prefer stored embedding comparisons (fast + consistent) if emb is not None: if raw_frame is None: # Still fine: we only use embedding similarity. pass emb = np.asarray(emb, dtype=np.float32) score = _cosine(query_emb, emb) if score > best_score: best_score = score best_cam = cam_id best_name = hit_name best_face_hit = face_hit best_raw_frame = raw_frame continue # Fallback: re-run InsightFace on the latest raw frame for this camera. # Do NOT discard candidates based on hit_name here; the best similarity # is what matters, and hit_name may be "Unknown" even when the face # is actually enrolled. if raw_frame is None: continue try: with face_engine.lock: live_faces = face_engine.app.get(raw_frame) if not live_faces: continue # Score against every detected face in the raw frame. for lf in live_faces: s = _cosine(query_emb, lf.embedding) if s <= best_score: continue best_score = s best_cam = cam_id # Use the stored name if available; otherwise let the final # "found" logic decide. if _is_known_identity(hit_name): best_name = hit_name else: best_name = lf.get('name') if isinstance(lf, dict) else (hit_name or "Unknown") best_face_hit = face_hit best_raw_frame = raw_frame except Exception as exc: logger.debug("fallback frame re-check failed for %s: %s", cam_id, exc) # Always generate evidence from the LIVE raw frame so the UI shows the # detected person, not the uploaded query image. evidence_image = None if best_raw_frame is not None and best_face_hit: try: x1, y1, x2, y2 = [int(v) for v in (best_face_hit.get("bbox") or [0, 0, 0, 0])] h, w = best_raw_frame.shape[:2] x1, y1 = max(0, x1), max(0, y1) x2, y2 = min(w, x2), min(h, y2) # Use the exact full frame instead of a zoomed/cropped picture frame_to_encode = best_raw_frame.copy() if x2 > x1 and y2 > y1: # Draw a green bounding box around the detected face on the full frame cv2.rectangle(frame_to_encode, (x1, y1), (x2, y2), (0, 255, 0), 2) ok, buf = cv2.imencode(".jpg", frame_to_encode, [int(cv2.IMWRITE_JPEG_QUALITY), 70]) if ok: import base64 evidence_image = f"data:image/jpeg;base64,{base64.b64encode(buf).decode()}" except Exception as exc: logger.debug("live evidence frame encoding failed for %s: %s", best_cam, exc) # Fallback to stored thumbnail only if crop failed. if evidence_image is None and best_face_hit: evidence_image = best_face_hit.get("thumbnail") if best_cam and best_score >= thresh: return { "found": True, "name": best_name, "cam_id": best_cam, "confidence": round(best_score, 3), "cameras_searched": cameras_searched, "search_mode": "live", "location": f"Live Camera: {best_cam}", "details": f"Detected on active live feed {best_cam}", "match_image": evidence_image, "reason": f"Query face matched live camera feed '{best_cam}' with confidence {best_score:.1%}.", } return { "found": False, "best_score": round(best_score, 3), "cameras_searched": cameras_searched, "search_mode": "live", "reason": ( "The uploaded face was not detected on any active live camera feed." if cameras_searched > 0 else "No active camera feeds with recognised faces available." ), }