"""Face ID Registration Gradio tab for Eve SDK applications. Provides a self-contained Gradio tab for registering and unregistering faces via the Eve SDK. Registration media is stored in per-session tmp directories (``tmp//``) so that each browser session is isolated and files are cleaned up when the session disconnects. Usage:: face_id_tab = FaceIdTab(eve, max_users=2) with gr.Blocks() as demo: session_registry = gr.State(value={}) with gr.Tabs(): face_id_tab.build() face_id_tab.wire(session_registry, concurrency_id="eve_sdk") """ import math import os import shutil import uuid from dataclasses import dataclass from pathlib import Path import cv2 import gradio as gr import numpy as np from eve_messages import CalibrationResultMsg from eve_worker_pool import EveWorkerPool, log_worker_activity from frame_utils import ( extract_frame_at_index, extract_frames, get_thumbnail, get_thumbnail_base64, load_media_frames, ) from log_utils import setup_logger from usage_analytics import UsageTracker from video_processing import ( VideoLimits, build_video_constraints_accordion, reencode_video, validate_video, wire_recording_limits, ) logger = setup_logger("FaceIdTab") _IMAGE_EXTENSIONS = frozenset({".png", ".jpg", ".jpeg", ".bmp", ".webp"}) _VIDEO_EXTENSIONS = frozenset({".mp4", ".avi", ".mov", ".mkv", ".webm"}) _MEDIA_EXTENSIONS = _IMAGE_EXTENSIONS | _VIDEO_EXTENSIONS @dataclass class FaceEntry: """A registered face in the session gallery. Attributes: path: Local path to the stored registration media. sdk_id: EVE SDK face ID assigned during the most recent gallery restore. ``None`` until the first video processing or live inference run. from_webcam: True when the stored media is a mirrored webcam capture. Mirrored storage keeps the thumbnail aligned with the selfie preview the user saw; frames loaded back out must be flipped to un-mirrored before being sent to the SDK for calibration so the embedding matches the (un-mirrored) frames the SDK receives during live/offline inference. """ path: str sdk_id: int | None = None from_webcam: bool = False class FaceIdTab: """Gradio tab for registering and managing Face ID users via the Eve SDK. Args: pool: Worker pool for acquiring Eve SDK workers. max_users: Maximum number of simultaneously registered users. examples_dir: Path to a directory of example images/videos to display. accept_video: Whether to show the video upload input. Defaults to False (image-only registration). video_limits: Optional constraints for uploaded registration videos. When provided, uploaded videos are validated against these limits and a "Video Constraints" accordion is shown in the UI. """ def __init__( self, pool: EveWorkerPool, max_users: int = 2, examples_dir: str = "", accept_video: bool = False, video_limits: VideoLimits | None = None, tracker: UsageTracker | None = None, ): self._pool = pool self._max_users = max_users self._accept_video = accept_video self._video_limits = video_limits self._tracker = tracker # Scan for example images (and videos, if accepted) separately self._image_examples: list[list[str]] = [] self._video_examples: list[list[str]] = [] if examples_dir: examples_path = Path(examples_dir) if examples_path.is_dir(): for p in sorted(examples_path.iterdir()): if not p.is_file(): continue suffix = p.suffix.lower() if suffix in _IMAGE_EXTENSIONS: self._image_examples.append([str(p)]) elif accept_video and suffix in _VIDEO_EXTENSIONS: self._video_examples.append([str(p)]) # Gradio components — populated by build() self._face_image_input: gr.Image self._face_video_input: gr.Video self._register_btn: gr.Button self._register_status: gr.Textbox self._slot_imgs: list[gr.Image] = [] self._remove_btns: list[gr.Button] = [] self._image_example_dataset: gr.Dataset | None = None self._video_example_dataset: gr.Dataset | None = None self._image_accordion: gr.Accordion | None = None self._video_accordion: gr.Accordion | None = None self._summaries: list[dict] = [] # each: {column, hint, imgs, height} self._example_path_state: gr.State self._select_frame_btn: gr.Button self._frame_preview: gr.Image self._video_time: gr.Number self._frame_index_state: gr.State # ------------------------------------------------------------------ # UI construction # ------------------------------------------------------------------ def build(self) -> None: """Create the Face ID Registration tab UI. Must be called inside a ``gr.Blocks`` / ``gr.Tabs`` context. """ with gr.TabItem("Face ID Registration"): gr.Markdown( "### Register Faces for Identification\n\n" "Register a user to use with Face ID in the Live Inference " "tab or the Offline Inference tab.\n\n" "> Note: When uploading a video, EVE will take the first valid frame to register the user." ) with gr.Accordion("Instructions", open=False): gr.Markdown( ( "1. Choose between registering a face from an **Image** or a " "**Video**\n" if self._accept_video else "1. Select an example image or upload your own\n" ) + "2. For an image\n" " 1. Select an example image, or upload your own\n" " 2. Press the **Register Face** button\n" "3. For a video\n" " 1. Expand the video section\n" " 2. Select an example video or upload your own\n" " 3. Press the **Register Face** button\n" "4. Go to another tab, enable **Face Identification**, and process " "a video\n\n" f"> **Platform note:** This demo supports up to {self._max_users} " "registered users. The full Eve SDK supports larger galleries and " "multi-pose calibration, but these features are limited here due to " "HuggingFace Spaces constraints." ) with gr.Row(): # --- Left column: input --- with gr.Column(scale=3): if self._accept_video: # Image section (expanded by default) with gr.Accordion( "Input from an Image", open=True ) as self._image_accordion: if self._image_examples: with gr.Accordion("Examples", open=True): self._image_example_dataset = gr.Dataset( components=[gr.Image(visible=False)], samples=self._image_examples, show_label=False, ) self._face_image_input = gr.Image( label="Upload Face Photo", sources=["upload", "webcam"], type="filepath", ) # Video section (collapsed by default) with gr.Accordion( "Input from a Video", open=False ) as self._video_accordion: if self._video_examples: with gr.Accordion("Examples", open=True): self._video_example_dataset = gr.Dataset( components=[gr.Video(visible=False)], samples=self._video_examples, show_label=False, ) if self._video_limits is not None: build_video_constraints_accordion(self._video_limits) with gr.Row(): with gr.Column(): self._face_video_input = gr.Video( label="Upload Short Video", sources=["upload", "webcam"], elem_id="face-video-input", ) with gr.Column(): self._frame_preview = gr.Image( label="Selected Frame", interactive=False, visible=False, ) self._select_frame_btn = gr.Button( "Select Current Frame", variant="secondary", size="sm", visible=False, ) else: if self._image_examples: with gr.Accordion("Image Examples", open=True): self._image_example_dataset = gr.Dataset( components=[gr.Image(visible=False)], samples=self._image_examples, show_label=False, ) self._face_image_input = gr.Image( label="Upload Face Photo", sources=["upload", "webcam"], type="filepath", ) # Hidden — needed for handler wiring but not shown self._face_video_input = gr.Video(visible=False) self._select_frame_btn = gr.Button(visible=False) self._frame_preview = gr.Image(visible=False) self._register_btn = gr.Button( "Register Face", variant="primary", interactive=False ) self._register_status = gr.Textbox(label="Status", interactive=False) self._example_path_state = gr.State(value=None) self._video_time = gr.Number(visible=False, value=0) self._frame_index_state = gr.State(value=None) # --- Right column: registered users --- with gr.Column(scale=1): gr.Markdown("#### Registered Users") for i in range(self._max_users): img = gr.Image( label=f"Slot {i + 1} — Empty", interactive=False, height=200, ) btn = gr.Button( f"Remove Slot {i + 1}", variant="stop", interactive=False, ) self._slot_imgs.append(img) self._remove_btns.append(btn) def build_summary(self, height: int = 80, scale: int = 1) -> None: """Create a summary column with minimal HTML thumbnails of registered faces. Uses ``gr.HTML`` instead of ``gr.Image`` so there are no fullscreen/download/share buttons — just a tiny thumbnail and label. The column starts hidden and appears once a face is registered. Can be called multiple times (e.g. once per tab) — each call creates an independent summary widget that is kept in sync automatically. Must be called **before** :meth:`wire`. Args: height: Max pixel height of each thumbnail image. scale: Column scale relative to siblings in the parent Row. """ imgs: list[gr.HTML] = [] column = gr.Column(scale=scale, min_width=100, visible=False) with column: gr.Markdown("**Registered Faces**") hint = gr.Markdown("_Go to the **Face ID Registration** tab to register faces._") for _ in range(self._max_users): imgs.append(gr.HTML(value="", visible=False)) self._summaries.append({"column": column, "hint": hint, "imgs": imgs, "height": height}) # ------------------------------------------------------------------ # Event wiring # ------------------------------------------------------------------ def wire( self, session_registry: gr.State, ) -> None: """Connect event handlers to the tab's UI components. Must be called inside the same ``gr.Blocks`` context as :meth:`build`. Args: session_registry: ``gr.State`` holding the per-session registry dict. """ # Validate uploaded registration videos against limits if self._video_limits is not None: wire_recording_limits( self._face_video_input, self._video_limits.max_duration_seconds, ) self._face_video_input.upload( fn=self._validate_video_upload, inputs=[self._face_video_input], outputs=[self._face_video_input], ) self._face_video_input.stop_recording( fn=self._process_webcam_recording, inputs=[self._face_video_input], outputs=[self._face_video_input], ) # Frame selector: show/hide capture button when video changes self._face_video_input.change( fn=self._on_video_change, inputs=[self._face_video_input], outputs=[self._select_frame_btn, self._frame_preview, self._frame_index_state], ) # Capture the currently displayed frame via the browser's video element self._select_frame_btn.click( fn=self._on_select_frame, inputs=[self._face_video_input, self._video_time], outputs=[self._frame_preview, self._frame_index_state], js="(video_path, _) => {" " const el = document.querySelector('#face-video-input video');" " return [video_path, el ? el.currentTime : 0];" "}", ) # Mutually exclusive accordions: expanding one collapses the other if self._image_accordion is not None and self._video_accordion is not None: self._image_accordion.expand( fn=lambda: gr.update(open=False), outputs=[self._video_accordion], ) self._video_accordion.expand( fn=lambda: gr.update(open=False), outputs=[self._image_accordion], ) # Load examples into the corresponding input on click if self._image_example_dataset is not None: self._image_example_dataset.click( fn=self._load_image_example, inputs=[self._image_example_dataset], outputs=[ self._face_image_input, self._face_video_input, self._example_path_state, ], ) if self._video_example_dataset is not None: self._video_example_dataset.click( fn=self._load_video_example, inputs=[self._video_example_dataset], outputs=[ self._face_image_input, self._face_video_input, self._example_path_state, ], ) # Enable/disable register button based on input availability for component in (self._face_image_input, self._face_video_input): component.change( fn=self._on_input_change, inputs=[ self._face_image_input, self._face_video_input, session_registry, ], outputs=self._register_btn, ) # Interleave slot images and remove buttons for outputs: # [slot1_img, remove_btn1, slot2_img, remove_btn2, ...] slot_outputs: list[gr.Component] = [] for img, btn in zip(self._slot_imgs, self._remove_btns): slot_outputs.extend([img, btn]) summary_outputs: list[gr.Component] = [] for s in self._summaries: summary_outputs.append(s["column"]) summary_outputs.append(s["hint"]) summary_outputs.extend(s["imgs"]) self._register_btn.click( fn=self.register_face, inputs=[ self._face_image_input, self._face_video_input, self._example_path_state, session_registry, self._frame_index_state, ], outputs=[ *slot_outputs, *summary_outputs, self._register_status, self._face_image_input, self._face_video_input, session_registry, self._example_path_state, self._select_frame_btn, self._frame_preview, self._frame_index_state, ], ) # Each slot gets its own remove button for slot_index in range(self._max_users): self._remove_btns[slot_index].click( fn=self._make_unregister_handler(slot_index), inputs=[session_registry], outputs=[ *slot_outputs, *summary_outputs, self._register_status, session_registry, self._register_btn, ], ) # ------------------------------------------------------------------ # Event handlers # ------------------------------------------------------------------ # Reset on success: clear preview, hide button, clear frame index. _FRAME_SELECTOR_RESET = ( gr.update(visible=False), gr.update(value=None, visible=False), None, ) def _error_return( self, registry: dict[int, FaceEntry], msg: str, frame_index: int | None, ) -> tuple: """Build a register_face return tuple for an error (state unchanged).""" gr.Warning(msg) return ( *self._slot_updates(registry), *self._summary_updates(registry), msg, gr.update(), gr.update(), registry, None, gr.update(), gr.update(), frame_index, ) def register_face( self, image_path: str | None, video_path: str | None, example_fallback_path: str | None, registry: dict[int, FaceEntry], frame_index: int | None, request: gr.Request, ) -> tuple: """Handle the Register Face button click. Returns: (*slot_updates, *summary_updates, status, clear_image, clear_video, registry, example_path_state, *frame_selector_reset) """ if len(registry) >= self._max_users: return self._error_return( registry, f"Cannot register: maximum {self._max_users} users already registered.", frame_index, ) # Fallback: if the Image/Video components haven't updated yet # (race between example-click and register-click), use the # example path stored in gr.State. if image_path is None and video_path is None and example_fallback_path is not None: ext = os.path.splitext(example_fallback_path)[1].lower() if ext in _IMAGE_EXTENSIONS: image_path = example_fallback_path elif ext in _VIDEO_EXTENSIONS: video_path = example_fallback_path if image_path is None and video_path is None: return self._error_return(registry, "Please upload an image or video.", frame_index) frame = None is_webcam = video_path is not None and _is_recording(video_path) try: if video_path is not None and frame_index is not None: frame = extract_frame_at_index(video_path, int(frame_index)) frames = [frame] else: frames = extract_frames(image_path, video_path) except Exception as exc: logger.error("Failed to extract frames for registration: %s", exc) return self._error_return(registry, f"Could not read media: {exc}", frame_index) if not frames: return self._error_return(registry, "No frames read from media.", frame_index) worker = self._pool.acquire(request.session_hash) log_worker_activity(logger, "acquired", "face-register", self._pool, worker.worker_id) try: worker.send_enable_face_id(enabled=True) result: CalibrationResultMsg = worker.send_calibrate_new_user(frames) if not result.success: return self._error_return( registry, f"Registration failed: {result.message}", frame_index ) session_tmp = _session_dir(request.session_hash) uid_hex = uuid.uuid4().hex[:8] if frame is not None: # Save the selected frame as an image so thumbnail and # restore_gallery use this exact frame, not the full video. # Webcam jpgs are stored mirrored so the thumbnail matches # the selfie-view preview the user saw. stored_path = os.path.join(session_tmp, f"face_id_{uid_hex}.jpg") display_frame = cv2.flip(frame, 1) if is_webcam else frame cv2.imwrite(stored_path, display_frame) _cleanup_recording(video_path) else: media_source = image_path if image_path is not None else video_path ext = os.path.splitext(media_source)[1] stored_path = os.path.join(session_tmp, f"face_id_{uid_hex}{ext}") shutil.copy2(media_source, stored_path) _cleanup_recording(video_path) next_key = max(registry.keys(), default=0) + 1 registry = { **registry, next_key: FaceEntry(path=stored_path, from_webcam=is_webcam), } all_frames = [_load_frames_for_sdk(entry) for entry in registry.values()] # The new entry was just added last; reuse the raw frames we # already have in memory instead of re-decoding + re-flipping # the jpg we just wrote. all_frames[-1] = frames restore_results = worker.send_restore_gallery(all_frames) for entry, r in zip(registry.values(), restore_results): entry.sdk_id = r.user_id if r.success else None except Exception as exc: logger.error("Face registration failed: %s", exc) return self._error_return(registry, f"Registration failed: {exc}", frame_index) finally: self._pool.release(worker) log_worker_activity(logger, "released", "face-register", self._pool, worker.worker_id) if self._tracker: media_type = "image" if image_path is not None else "video" self._tracker.log( request.session_hash, "face_register", media_type=media_type, slot_count=len(registry), ) return ( *self._slot_updates(registry), *self._summary_updates(registry), ( f"Successfully registered (Face ID: {registry[next_key].sdk_id})." if registry[next_key].sdk_id is not None else f"Successfully registered as User {next_key}." ), None, None, registry, None, *self._FRAME_SELECTOR_RESET, ) def _make_unregister_handler(self, slot_index: int): """Create a remove handler bound to a specific slot index.""" def handler(registry: dict[int, FaceEntry], request: gr.Request) -> tuple: user_ids = sorted(registry.keys()) if slot_index >= len(user_ids): return ( *self._slot_updates(registry), *self._summary_updates(registry), f"Slot {slot_index + 1} is empty.", registry, gr.update(), ) uid = user_ids[slot_index] removed_entry = registry[uid] if os.path.exists(removed_entry.path): os.remove(removed_entry.path) remaining = {u: entry for u, entry in registry.items() if u != uid} # Re-register surviving users on a worker worker = self._pool.acquire(request.session_hash) log_worker_activity(logger, "acquired", "face-unregister", self._pool, worker.worker_id) try: worker.send_remove_all_users() if remaining: new_registry: dict[int, FaceEntry] = {} for u, entry in remaining.items(): frames = _load_frames_for_sdk(entry) result = worker.send_calibrate_new_user(frames) if result.success: new_registry[u] = FaceEntry( path=entry.path, sdk_id=result.user_id, from_webcam=entry.from_webcam, ) else: logger.warning(f"Failed to re-register user {u}: {result.message}") if os.path.exists(entry.path): os.remove(entry.path) registry = new_registry else: registry = remaining finally: self._pool.release(worker) log_worker_activity( logger, "released", "face-unregister", self._pool, worker.worker_id ) if self._tracker: self._tracker.log( request.session_hash, "face_remove", slot_count=len(registry), ) can_register = len(registry) < self._max_users return ( *self._slot_updates(registry), *self._summary_updates(registry), f"User {uid} removed.", registry, gr.update(interactive=can_register), ) return handler # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _validate_video_upload(self, video_path: str | None) -> str | None: """Validate an uploaded registration video against limits. Returns: The video path if valid, or None if rejected. """ if not video_path or self._video_limits is None: return video_path try: validate_video(video_path, self._video_limits) return video_path except Exception as error: gr.Warning(str(error), duration=None) return None def _process_webcam_recording(self, video_path: str | None) -> str | None: """Re-encode a webcam recording to an MP4 with proper time_base. Browser MediaRecorder produces WebM with duration=Infinity, which breaks HTML5 scrubbing (``currentTime`` is stuck at 0). Re-encoding to CFR H.264 MP4 with explicit ``stream.time_base`` gives the file a known duration so the player can seek. File is stored un-mirrored in :data:`_RECORDINGS_DIR`. Gradio's player CSS-flips webcam-sourced videos during playback, so flipping the file would double-flip; frames read back out for display are mirrored at extraction time by :meth:`_extract_frame_for_display`. """ if not video_path: return video_path if self._video_limits is not None: try: validate_video(video_path, self._video_limits) except Exception as error: gr.Warning(str(error), duration=None) return None os.makedirs(_RECORDINGS_DIR, exist_ok=True) output_path = os.path.join( _RECORDINGS_DIR, f"{_RECORDING_BASENAME_PREFIX}{uuid.uuid4().hex[:8]}.mp4" ) try: reencode_video(video_path, output_path) except Exception as error: logger.error("Failed to re-encode webcam recording: %s", error) gr.Warning(f"Could not process recording: {error}") return None return output_path @staticmethod def _extract_frame_for_display(video_path: str, frame_index: int): """Extract a frame in the orientation the user sees in the preview. Webcam recording files are stored un-mirrored (see :meth:`_process_webcam_recording`); Gradio's player CSS-flips them during playback, so for the preview thumbnail to match what the user saw we mirror the raw frame here. For SDK consumption, call :func:`extract_frame_at_index` directly — the SDK processes un-mirrored frames at inference time, so passing a mirrored frame here would produce a different embedding than live/offline inference. """ frame = extract_frame_at_index(video_path, frame_index) if _is_recording(video_path): frame = cv2.flip(frame, 1) return frame def _on_input_change( self, image_path: str | None, video_path: str | None, registry: dict[int, FaceEntry], ) -> dict: has_input = image_path is not None or video_path is not None can_register = has_input and len(registry) < self._max_users return gr.update(interactive=can_register) def _on_video_change(self, video_path: str | None) -> tuple: """Auto-select frame 0 when a video is uploaded; reset when cleared.""" if not video_path: return self._FRAME_SELECTOR_RESET try: frame = self._extract_frame_for_display(video_path, 0) preview = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) except RuntimeError: return self._FRAME_SELECTOR_RESET return gr.update(visible=True), gr.update(value=preview, visible=True), 0 def _on_select_frame(self, video_path: str | None, current_time: float) -> tuple: """Capture the frame at the video's current playback position.""" if not video_path: return gr.update(), None # Guard against NaN/Infinity that HTML5