Spaces:

SolutionChallenge
/

solution_challenge_backend

Running

solution_challenge_backend / backend /face_live_search.py

github-actions

Deploy to Hugging Face

c794b6b 5 months ago

8.22 kB

"""
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."
        ),
    }

	"""
	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."
	),
	}