shared/face_id_tab.py

"""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/<session_hash>/``) 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 <u>Live Inference</u> "
                "tab or the <u>Offline Inference</u> 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 <video> returns when the
        # container advertises an unknown duration.
        if current_time is None or not math.isfinite(float(current_time)):
            current_time = 0.0
        cap = cv2.VideoCapture(video_path)
        fps = cap.get(cv2.CAP_PROP_FPS)
        cap.release()
        if fps <= 0 or fps > 240 or not math.isfinite(fps):
            fps = 30.0
        frame_index = max(0, int(current_time * fps))
        try:
            frame = self._extract_frame_for_display(video_path, frame_index)
            preview = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        except RuntimeError:
            return gr.update(), None
        return gr.update(value=preview, visible=True), frame_index

    @staticmethod
    def _load_image_example(sample: list[str]) -> tuple[str, None, str]:
        """Load an image example into the photo input, clearing the video input."""
        return sample[0], None, sample[0]

    @staticmethod
    def _load_video_example(sample: list[str]) -> tuple[None, str, str]:
        """Load a video example into the video input, clearing the image input."""
        return None, sample[0], sample[0]

    @staticmethod
    def _face_label(slot: int, uid: int, entry: FaceEntry) -> str:
        """Build the display label for a registered face slot."""
        if entry.sdk_id is not None:
            return f"Face ID: {entry.sdk_id}"
        return f"Slot {slot} — User {uid}"

    def _slot_updates(self, registry: dict[int, FaceEntry]) -> tuple:
        """Build interleaved (image_update, button_update) tuples for all slots."""
        updates: list[dict] = []
        user_ids = sorted(registry.keys())
        for i in range(self._max_users):
            if i < len(user_ids):
                uid = user_ids[i]
                entry = registry[uid]
                thumb = get_thumbnail(entry.path)
                label = self._face_label(i + 1, uid, entry)
                updates.append(gr.update(value=thumb, label=label))
                updates.append(gr.update(interactive=True))
            else:
                updates.append(gr.update(value=None, label=f"Slot {i + 1} — Empty"))
                updates.append(gr.update(interactive=False))
        return tuple(updates)

    def _summary_updates_single(self, registry: dict[int, FaceEntry], summary: dict) -> list:
        """Build updates for one summary group (column + hint + images)."""
        updates: list = []
        user_ids = sorted(registry.keys())
        h = summary["height"]

        updates.append(gr.update(visible=bool(registry)))

        if registry:
            hint = "_Go to the **Face ID Registration** tab to manage registered faces._"
        else:
            hint = "_Go to the **Face ID Registration** tab to register faces._"
        updates.append(gr.update(value=hint))

        for i in range(self._max_users):
            if i < len(user_ids):
                uid = user_ids[i]
                entry = registry[uid]
                label = self._face_label(i + 1, uid, entry)
                b64 = get_thumbnail_base64(entry.path)
                if b64:
                    html = (
                        f'<div style="display:flex;flex-direction:column;'
                        f'align-items:center;gap:2px">'
                        f'<img src="data:image/jpeg;base64,{b64}" '
                        f'style="max-height:{h}px;border-radius:4px" />'
                        f'<span style="font-size:11px;color:#555">{label}</span>'
                        f"</div>"
                    )
                else:
                    html = (
                        f'<div style="text-align:center;font-size:11px;'
                        f'color:#555">{label}</div>'
                    )
                updates.append(gr.update(value=html, visible=True))
            else:
                updates.append(gr.update(value="", visible=False))
        return updates

    def _summary_updates(self, registry: dict[int, FaceEntry]) -> tuple:
        """Build updates for all summary groups."""
        if not self._summaries:
            return ()
        updates: list = []
        for s in self._summaries:
            updates.extend(self._summary_updates_single(registry, s))
        return tuple(updates)

    # ------------------------------------------------------------------
    # Public API for external wiring (e.g. post-inference refresh)
    # ------------------------------------------------------------------

    @property
    def all_slot_components(self) -> list[gr.Component]:
        """All updatable slot components (tab images/buttons + all summaries)."""
        components: list[gr.Component] = []
        for img, btn in zip(self._slot_imgs, self._remove_btns):
            components.extend([img, btn])
        for s in self._summaries:
            components.append(s["column"])
            components.append(s["hint"])
            components.extend(s["imgs"])
        return components

    @property
    def summary_components(self) -> list[gr.Component]:
        """All summary components across every summary group."""
        components: list[gr.Component] = []
        for s in self._summaries:
            components.append(s["column"])
            components.append(s["hint"])
            components.extend(s["imgs"])
        return components

    def refresh_summary(self, registry: dict[int, FaceEntry]) -> tuple:
        """Refresh only the summary from the registry.

        Intended as a Gradio handler for tab-select events to ensure
        the summary stays in sync when switching tabs.
        """
        return self._summary_updates(registry)

    def refresh_all(self, registry: dict[int, FaceEntry]) -> tuple:
        """Refresh tab slot displays and summary thumbnails from the registry.

        Intended as a Gradio handler chained after operations that modify
        the registry outside the Face ID tab (e.g. video inference).
        """
        return (*self._slot_updates(registry), *self._summary_updates(registry))


# ------------------------------------------------------------------
# Module-level helpers (no instance state needed)
# ------------------------------------------------------------------


# Project-root-anchored: EveWrapper leaves the process cwd at the EVE bin
# directory, and the main process and worker subprocesses no longer share a
# cwd, so cwd-relative paths cannot round-trip between them.
_PROJECT_TMP = Path(__file__).resolve().parents[2] / "tmp"


def _session_dir(session_hash: str) -> str:
    """Return the per-session tmp directory, creating it if needed."""
    path = _PROJECT_TMP / session_hash
    path.mkdir(parents=True, exist_ok=True)
    return str(path)


_RECORDINGS_DIR = str(_PROJECT_TMP / "face_id_recordings")
# Distinctive basename marker so we can identify our re-encoded webcam
# recordings even after Gradio copies them through its served-files cache
# (which can change the parent directory but preserves the basename).
_RECORDING_BASENAME_PREFIX = "facecam_rec_"


def _is_recording(video_path: str | None) -> bool:
    """True when ``video_path`` is one of our re-encoded webcam recordings.

    Uses a basename prefix because Gradio may serve the file from its own
    files cache, in which case the parent directory no longer matches
    :data:`_RECORDINGS_DIR`.
    """
    if not video_path:
        return False
    return os.path.basename(video_path).startswith(_RECORDING_BASENAME_PREFIX)


def _cleanup_recording(video_path: str | None) -> None:
    """Delete a webcam-recording file once it's been consumed by registration.

    Only removes files inside :data:`_RECORDINGS_DIR` (not Gradio's cache
    copies, which Gradio manages itself).
    """
    if not _is_recording(video_path):
        return
    abs_recordings = os.path.abspath(_RECORDINGS_DIR)
    try:
        if os.path.commonpath([os.path.abspath(video_path), abs_recordings]) != abs_recordings:
            return
    except ValueError:
        return
    try:
        os.remove(video_path)
    except OSError:
        pass


def _load_frames_for_sdk(entry: FaceEntry) -> list[np.ndarray]:
    """Load a FaceEntry's stored frames in the orientation the SDK expects.

    Webcam entries are stored mirrored (to match the selfie preview); the
    SDK processes un-mirrored frames, so we flip those back here.
    """
    frames = load_media_frames(entry.path)
    if entry.from_webcam:
        frames = [cv2.flip(f, 1) for f in frames]
    return frames