Spaces:
Sleeping
Sleeping
File size: 3,975 Bytes
9b76f08 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """Browser-based live demo: face detection + tracking + per-session persistent ID.
Deploy target: Hugging Face Spaces (Gradio SDK, Python 3.10, free CPU Basic).
This is a demo of the pipeline in ../app, not the NUC/production app itself -
no RTSP, no SQLite, nothing saved. Each visitor gets their own tracker and
FAISS gallery for the lifetime of their browser tab (held in gr.State);
nothing is stored server-side or shared between visitors.
"""
import os
from typing import Dict, List, Optional, Tuple
import numpy as np
import gradio as gr
import supervision as sv
from face_engine import FaceEngine
from identity_store import IdentityStore
DET_SIZE = int(os.environ.get("DET_SIZE", "320"))
MODEL_NAME = os.environ.get("MODEL_NAME", "buffalo_s")
ONNX_THREADS = int(os.environ.get("ONNX_THREADS", "2"))
MATCH_THRESHOLD = float(os.environ.get("MATCH_THRESHOLD", "0.38"))
box_annotator = sv.BoxAnnotator()
label_annotator = sv.LabelAnnotator()
# One shared FaceEngine process-wide (loading the ONNX models per visitor
# would be far too slow); it holds no per-visitor state, so this is safe.
face_engine = FaceEngine(
det_size=(DET_SIZE, DET_SIZE),
ctx_id=-1,
model_name=MODEL_NAME,
num_threads=ONNX_THREADS,
)
SessionState = Tuple[sv.ByteTrack, IdentityStore, Dict[int, int]]
def build_detections(faces: list) -> sv.Detections:
if not faces:
return sv.Detections.empty()
xyxy = np.array([face.bbox for face in faces], dtype=np.float32)
confidence = np.array([face.det_score for face in faces], dtype=np.float32)
embeddings = np.array([face.normed_embedding for face in faces], dtype=np.float32)
class_id = np.zeros(len(faces), dtype=int)
return sv.Detections(
xyxy=xyxy,
confidence=confidence,
class_id=class_id,
data={"embedding": embeddings},
)
def process_frame(
frame_rgb: Optional[np.ndarray], state: Optional[SessionState]
) -> Tuple[Optional[np.ndarray], SessionState]:
if state is None:
state = (sv.ByteTrack(), IdentityStore(embedding_dim=512, match_threshold=MATCH_THRESHOLD), {})
tracker, identity_store, track_to_person = state
if frame_rgb is None:
return None, state
# gradio's numpy Image type is RGB; our pipeline (InsightFace, supervision) expects BGR.
frame_bgr = frame_rgb[:, :, ::-1]
faces = face_engine.detect(frame_bgr)
detections = build_detections(faces)
detections = tracker.update_with_detections(detections)
labels: List[str] = []
embeddings = detections.data.get("embedding") if detections.data else None
for i in range(len(detections)):
track_id = int(detections.tracker_id[i])
if embeddings is not None:
person_id = identity_store.match_or_register(embeddings[i])
track_to_person[track_id] = person_id
person_id = track_to_person.get(track_id, track_id)
labels.append(f"ID {person_id}")
annotated_bgr = box_annotator.annotate(scene=frame_bgr.copy(), detections=detections)
annotated_bgr = label_annotator.annotate(scene=annotated_bgr, detections=detections, labels=labels)
return annotated_bgr[:, :, ::-1], (tracker, identity_store, track_to_person)
with gr.Blocks(title="Face Tracker Demo") as demo:
gr.Markdown(
"# Real-time multi-person face identification & tracking\n"
"Allow webcam access below. Each detected face gets a box and a "
"persistent `ID n` for as long as this tab stays open. Nothing is "
"recorded, stored, or sent anywhere beyond this page - it all resets "
"when you reload."
)
state = gr.State(None)
with gr.Row():
webcam = gr.Image(sources=["webcam"], streaming=True, label="Your webcam", mirror_webcam=True)
output = gr.Image(label="Detected + tracked", streaming=True)
webcam.stream(fn=process_frame, inputs=[webcam, state], outputs=[output, state])
if __name__ == "__main__":
demo.queue().launch()
|