face_module/camera_runtime.py

from __future__ import annotations

import logging
import sys
import threading
import time
from typing import Any, Callable

import numpy as np

from config import (
    FACE_ANALYSIS_INTERVAL_SECONDS,
    FACE_BOX_TRACK_INTERVAL_SECONDS,
    FACE_CAMERA_BACKEND,
    FACE_CAMERA_INDEX,
    FACE_CAMERA_POLL_INTERVAL_SECONDS,
    FACE_CAMERA_PREVIEW_FPS,
    FACE_CAMERA_PREVIEW_JPEG_QUALITY,
    FACE_CAMERA_PREVIEW_WIDTH,
    FACE_CAMERA_SCAN_INDICES,
    FACE_DETECT_INTERVAL,
)

LOGGER = logging.getLogger(__name__)


def probe_camera_access(
    *,
    camera_index: int = FACE_CAMERA_INDEX,
    camera_backend: str | int | None = FACE_CAMERA_BACKEND,
    candidate_camera_indices: tuple[int, ...] = (),
) -> dict[str, object]:
    """Try to open and read one frame without loading face analysis models."""
    try:
        import cv2 as cv2_module
    except Exception as exc:
        return {
            "ok": False,
            "error": f"OpenCV (cv2) 未安装，无法启动本地摄像头：{exc}",
            "camera_index": None,
            "camera_backend": None,
            "frame_shape": None,
            "attempts": [],
        }

    runtime = FaceCameraRuntime(
        analyzer_factory=lambda: object(),
        camera_index=camera_index,
        camera_backend=camera_backend,
        candidate_camera_indices=candidate_camera_indices,
        cv2_module=cv2_module,
    )
    capture, active_index, active_backend, attempts, first_frame = runtime._open_capture(cv2_module)
    frame_shape = None
    if capture is not None:
        try:
            frame_shape = getattr(first_frame, "shape", None)
            if frame_shape is None:
                width = int(capture.get(cv2_module.CAP_PROP_FRAME_WIDTH))
                height = int(capture.get(cv2_module.CAP_PROP_FRAME_HEIGHT))
                frame_shape = (height, width, 3) if width and height else None
        except Exception:
            frame_shape = None
        runtime._release_capture_object(capture)

    return {
        "ok": capture is not None,
        "error": None if capture is not None else "无法打开并读取摄像头画面。",
        "camera_index": active_index,
        "camera_backend": active_backend,
        "frame_shape": frame_shape,
        "attempts": attempts,
    }


class FaceCameraRuntime:
    """Background camera capture and face analysis runtime.

    The analyzer is created once in start() so the factory (which hits
    Streamlit's cache) is only called on startup, not every Nth frame.
    """

    def __init__(
        self,
        analyzer_factory: Callable[[], Any],
        *,
        detect_interval: int = FACE_DETECT_INTERVAL,
        camera_index: int = FACE_CAMERA_INDEX,
        camera_backend: str | int | None = FACE_CAMERA_BACKEND,
        candidate_camera_indices: tuple[int, ...] = FACE_CAMERA_SCAN_INDICES,
        poll_interval_seconds: float = FACE_CAMERA_POLL_INTERVAL_SECONDS,
        analysis_interval_seconds: float = FACE_ANALYSIS_INTERVAL_SECONDS,
        box_track_interval_seconds: float = FACE_BOX_TRACK_INTERVAL_SECONDS,
        preview_fps: float = FACE_CAMERA_PREVIEW_FPS,
        preview_width: int = FACE_CAMERA_PREVIEW_WIDTH,
        preview_jpeg_quality: int = FACE_CAMERA_PREVIEW_JPEG_QUALITY,
        camera_factory: Callable[[int], Any] | None = None,
        cv2_module: Any | None = None,
    ) -> None:
        self._analyzer_factory = analyzer_factory
        self._detect_interval = max(1, int(detect_interval))
        self._camera_index = int(camera_index)
        self._camera_backend = camera_backend
        self._candidate_camera_indices = tuple(int(index) for index in candidate_camera_indices)
        self._poll_interval_seconds = max(0.01, float(poll_interval_seconds))
        self._analysis_interval_seconds = max(0.0, float(analysis_interval_seconds))
        self._box_track_interval_seconds = max(0.1, float(box_track_interval_seconds))
        self._preview_interval_seconds = 1.0 / max(1.0, float(preview_fps))
        self._preview_width = max(0, int(preview_width))
        self._preview_jpeg_quality = min(95, max(30, int(preview_jpeg_quality)))
        self._camera_factory = camera_factory
        self._cv2 = cv2_module

        self._analyzer: Any = None
        self._thread: threading.Thread | None = None
        self._stop_event = threading.Event()
        self._capture = None
        self._lock = threading.Lock()
        self._analysis_lock = threading.Lock()
        self._analysis_thread: threading.Thread | None = None
        self._analysis_in_progress = False
        self._run_id = 0
        self._running = False
        self._latest_frame: np.ndarray | None = None
        self._latest_result: dict[str, object] | None = None
        self._latest_result_at: float | None = None
        self._latest_face_box: tuple[int, int, int, int] | None = None
        self._latest_face_box_at: float | None = None
        self._latest_error: str | None = None
        self._latest_frame_at: float | None = None
        self._latest_preview_jpeg: bytes | None = None
        self._latest_preview_at: float | None = None
        self._frame_count = 0
        self._last_analysis_scheduled_at = 0.0
        self._last_preview_encoded_at = 0.0
        self._last_box_tracked_at = 0.0
        self._preview_encode_status = "idle"
        self._haar_cascade = None
        self._open_attempts: list[str] = []
        self._active_camera_index: int | None = None
        self._active_camera_backend: str | None = None

    def start(self) -> bool:
        if self._running:
            return True

        cv2_module = self._cv2
        if cv2_module is None:
            try:
                import cv2 as _cv2
                cv2_module = _cv2
            except Exception as exc:
                self._set_error(f"OpenCV (cv2) 未安装，无法启动本地摄像头：{exc}")
                return False
            self._cv2 = cv2_module

        capture, camera_index, camera_backend, attempts, first_frame = self._open_capture(cv2_module)
        self._open_attempts = attempts
        if capture is None:
            details = "；".join(attempts) if attempts else "没有可用的摄像头打开尝试"
            self._set_error(
                "无法打开摄像头。"
                f"已尝试：{details}。"
                "请检查系统摄像头权限、是否被其他程序占用，或切换摄像头索引/后端。"
            )
            return False

        try:
            analyzer = self._analyzer_factory()
        except Exception as exc:
            LOGGER.exception("Face analyzer initialization failed.")
            self._release_capture_object(capture)
            self._set_error(f"人脸分析模型初始化失败，摄像头已释放：{exc}")
            return False

        self._analyzer = analyzer
        self._capture = capture
        self._active_camera_index = camera_index
        self._active_camera_backend = camera_backend
        self._stop_event.clear()
        with self._lock:
            self._running = True
            self._run_id += 1
            run_id = self._run_id
            self._latest_frame = None
            self._latest_result = None
            self._latest_result_at = None
            self._latest_face_box = None
            self._latest_face_box_at = None
            self._latest_error = None
            self._latest_frame_at = None
            self._latest_preview_jpeg = None
            self._latest_preview_at = None
            self._frame_count = 0
            self._last_analysis_scheduled_at = 0.0
            self._last_preview_encoded_at = 0.0
            self._last_box_tracked_at = 0.0
            self._preview_encode_status = "idle"
        if first_frame is not None:
            try:
                first_frame_rgb = cv2_module.cvtColor(first_frame, cv2_module.COLOR_BGR2RGB)
            except Exception:
                first_frame_rgb = None
            if first_frame_rgb is not None:
                with self._lock:
                    self._latest_frame = first_frame_rgb
                    self._latest_frame_at = time.time()
                    self._frame_count = 1
                self._update_preview_jpeg(first_frame_rgb, cv2_module, force=True)
        self._thread = threading.Thread(target=self._loop, args=(run_id,), name="face-camera-runtime", daemon=True)
        self._thread.start()
        return True

    def stop(self) -> None:
        self._stop_event.set()
        thread = self._thread
        if thread is not None and thread.is_alive():
            thread.join(timeout=1.5)
        self._thread = None
        with self._lock:
            self._running = False
            self._run_id += 1
        self._release_capture()

    def is_running(self) -> bool:
        return self._running

    def snapshot(self) -> dict[str, object]:
        with self._lock:
            frame = self._latest_frame
            result = self._latest_result
            error = self._latest_error
            frame_count = self._frame_count
            latest_frame_at = self._latest_frame_at
            preview_jpeg = self._latest_preview_jpeg
            preview_at = self._latest_preview_at
            result_at = self._latest_result_at
            face_box = self._latest_face_box
            face_box_at = self._latest_face_box_at
            analysis_in_progress = self._analysis_in_progress
            preview_encode_status = self._preview_encode_status
        return {
            "frame": frame,
            "result": result,
            "error": error,
            "running": self._running,
            "preview_jpeg": preview_jpeg,
            "preview_at": preview_at,
            "preview_encode_status": preview_encode_status,
            "result_at": result_at,
            "face_box": face_box,
            "face_box_at": face_box_at,
            "camera_index": self._active_camera_index,
            "camera_backend": self._active_camera_backend,
            "frame_count": frame_count,
            "latest_frame_at": latest_frame_at,
            "analysis_in_progress": analysis_in_progress,
            "open_attempts": list(self._open_attempts),
        }

    def _open_capture(self, cv2_module: Any) -> tuple[Any | None, int | None, str | None, list[str], Any | None]:
        if self._camera_factory:
            capture = self._camera_factory(self._camera_index)
            ok, reason, first_frame = self._validate_capture(capture)
            attempts = [f"index={self._camera_index} backend=custom: {reason}"]
            if ok:
                return capture, self._camera_index, "custom", attempts, first_frame
            self._release_capture_object(capture)
            return None, None, None, attempts, None

        attempts: list[str] = []
        for camera_index in self._camera_indices():
            for backend_name, backend_id in self._backend_candidates(cv2_module):
                try:
                    if backend_id is None:
                        capture = cv2_module.VideoCapture(camera_index)
                    else:
                        capture = cv2_module.VideoCapture(camera_index, backend_id)
                except Exception as exc:
                    attempts.append(f"index={camera_index} backend={backend_name}: 异常 {exc}")
                    continue

                ok, reason, first_frame = self._validate_capture(capture)
                attempts.append(f"index={camera_index} backend={backend_name}: {reason}")
                if ok:
                    return capture, camera_index, backend_name, attempts, first_frame
                self._release_capture_object(capture)

        return None, None, None, attempts, None

    def _camera_indices(self) -> tuple[int, ...]:
        ordered = [self._camera_index, *self._candidate_camera_indices]
        seen: set[int] = set()
        indices: list[int] = []
        for index in ordered:
            if index not in seen:
                seen.add(index)
                indices.append(index)
        return tuple(indices)

    def _backend_candidates(self, cv2_module: Any) -> tuple[tuple[str, int | None], ...]:
        requested = self._camera_backend
        if isinstance(requested, int):
            return ((str(requested), requested),)

        requested_name = str(requested or "auto").strip().lower()
        if requested_name != "auto":
            backend = self._backend_from_name(cv2_module, requested_name)
            return (backend,) if backend else (("Default", None),)

        if sys.platform.startswith("win"):
            names = ("dshow", "any", "msmf")
        elif sys.platform == "darwin":
            names = ("avfoundation", "any")
        else:
            names = ("v4l2", "any")

        candidates = [
            backend
            for name in names
            if (backend := self._backend_from_name(cv2_module, name)) is not None
        ]
        return tuple(dict.fromkeys(candidates)) or (("Default", None),)

    @staticmethod
    def _backend_from_name(cv2_module: Any, name: str) -> tuple[str, int | None] | None:
        aliases = {
            "any": ("Default", None),
            "default": ("Default", None),
            "opencv": ("Default", None),
            "dshow": ("DirectShow", "CAP_DSHOW"),
            "directshow": ("DirectShow", "CAP_DSHOW"),
            "msmf": ("Media Foundation", "CAP_MSMF"),
            "mediafoundation": ("Media Foundation", "CAP_MSMF"),
            "v4l2": ("V4L2", "CAP_V4L2"),
            "avfoundation": ("AVFoundation", "CAP_AVFOUNDATION"),
        }
        label_and_attr = aliases.get(name)
        if label_and_attr is None:
            return None
        label, attr = label_and_attr
        if attr is None:
            return label, None
        backend_id = getattr(cv2_module, attr, None)
        if backend_id is None:
            return None
        return label, int(backend_id)

    @staticmethod
    def _validate_capture(capture: Any | None) -> tuple[bool, str, Any | None]:
        if capture is None:
            return False, "创建失败", None
        try:
            if not capture.isOpened():
                return False, "未打开", None
        except Exception as exc:
            return False, f"状态检查异常 {exc}", None

        for _ in range(3):
            try:
                ret, frame = capture.read()
            except Exception as exc:
                return False, f"读取异常 {exc}", None
            if ret and frame is not None:
                return True, "已打开并读取到画面", frame
            time.sleep(0.05)
        return False, "已打开但无法读取画面", None

    def _loop(self, run_id: int) -> None:
        cv2_module = self._cv2
        capture = self._capture
        frame_count = 0

        try:
            while not self._stop_event.is_set() and self._is_current_run(run_id):
                if capture is None or cv2_module is None:
                    self._set_error("摄像头运行时未正确初始化。")
                    break

                ret, frame = capture.read()
                if not ret:
                    self._set_error("无法读取摄像头画面。")
                    break

                frame_rgb = cv2_module.cvtColor(frame, cv2_module.COLOR_BGR2RGB)
                self._track_face_box(frame_rgb, cv2_module)

                with self._lock:
                    self._latest_frame = frame_rgb
                    self._latest_frame_at = time.time()
                    self._frame_count += 1
                self._update_preview_jpeg(frame_rgb, cv2_module)

                now = time.monotonic()
                can_analyze = now - self._last_analysis_scheduled_at >= self._analysis_interval_seconds
                if can_analyze and frame_count % self._detect_interval == 0:
                    self._last_analysis_scheduled_at = now
                    self._schedule_analysis(frame_rgb.copy(), run_id)

                frame_count += 1
                time.sleep(self._poll_interval_seconds)
        finally:
            if self._is_current_run(run_id):
                with self._lock:
                    self._running = False
                self._release_capture()

    def _schedule_analysis(self, frame_rgb: np.ndarray, run_id: int) -> None:
        with self._analysis_lock:
            if self._analysis_in_progress or not self._is_current_run(run_id):
                return
            self._analysis_in_progress = True
            self._analysis_thread = threading.Thread(
                target=self._analyze_frame,
                args=(frame_rgb, run_id),
                name="face-camera-analysis",
                daemon=True,
            )
            self._analysis_thread.start()

    def _analyze_frame(self, frame_rgb: np.ndarray, run_id: int) -> None:
        try:
            detected_result = self._analyzer.analyze_image(frame_rgb)
        except Exception as exc:
            LOGGER.exception("Face frame analysis failed.")
            if self._is_current_run(run_id):
                self._set_error(f"人脸分析失败：{exc}")
        else:
            if self._is_current_run(run_id):
                with self._lock:
                    self._latest_error = None
                    if isinstance(detected_result, dict):
                        self._latest_result = detected_result
                        self._latest_result_at = time.time()
                        box = _normalize_box(detected_result.get("face_box"))
                        if box is not None:
                            self._latest_face_box = box
                            self._latest_face_box_at = time.time()
        finally:
            with self._analysis_lock:
                self._analysis_in_progress = False

    def _track_face_box(self, frame_rgb: np.ndarray, cv2_module: Any) -> None:
        now = time.monotonic()
        if now - self._last_box_tracked_at < self._box_track_interval_seconds:
            return
        self._last_box_tracked_at = now

        box = _detect_largest_face_box(frame_rgb, cv2_module, self)
        with self._lock:
            self._latest_face_box = box
            self._latest_face_box_at = time.time()

    def _update_preview_jpeg(self, frame_rgb: np.ndarray, cv2_module: Any, *, force: bool = False) -> None:
        now = time.monotonic()
        if not force and now - self._last_preview_encoded_at < self._preview_interval_seconds:
            return
        self._last_preview_encoded_at = now

        if not hasattr(cv2_module, "imencode"):
            return

        with self._lock:
            face_box = self._latest_face_box
            result = dict(self._latest_result) if isinstance(self._latest_result, dict) else None
            result_at = self._latest_result_at
            analysis_in_progress = self._analysis_in_progress

        preview_frame = frame_rgb
        try:
            if (
                self._preview_width
                and getattr(frame_rgb, "ndim", 0) == 3
                and frame_rgb.shape[1] > self._preview_width
                and hasattr(cv2_module, "resize")
            ):
                target_height = max(1, int(frame_rgb.shape[0] * self._preview_width / frame_rgb.shape[1]))
                interpolation = getattr(cv2_module, "INTER_AREA", 3)
                preview_frame = cv2_module.resize(frame_rgb, (self._preview_width, target_height), interpolation=interpolation)

            preview_frame = _draw_preview_overlay(
                preview_frame,
                cv2_module,
                face_box=face_box,
                source_shape=frame_rgb.shape,
                result=result,
                result_at=result_at,
                analysis_in_progress=analysis_in_progress,
            )

            encode_frame = preview_frame
            if hasattr(cv2_module, "COLOR_RGB2BGR"):
                encode_frame = cv2_module.cvtColor(preview_frame, cv2_module.COLOR_RGB2BGR)

            params: list[int] = []
            quality_prop = getattr(cv2_module, "IMWRITE_JPEG_QUALITY", None)
            if quality_prop is not None:
                params = [int(quality_prop), self._preview_jpeg_quality]
            ok, encoded = cv2_module.imencode(".jpg", encode_frame, params)
        except Exception:
            LOGGER.warning("Preview frame encoding failed.", exc_info=True)
            with self._lock:
                self._preview_encode_status = "error"
            return

        if not ok:
            with self._lock:
                self._preview_encode_status = "failed"
            return
        with self._lock:
            self._latest_preview_jpeg = encoded.tobytes()
            self._latest_preview_at = time.time()
            self._preview_encode_status = "ok"

    def _is_current_run(self, run_id: int) -> bool:
        return self._running and run_id == self._run_id

    def _release_capture(self) -> None:
        capture = self._capture
        self._capture = None
        self._release_capture_object(capture)
        self._active_camera_index = None
        self._active_camera_backend = None

    def _set_error(self, message: str) -> None:
        with self._lock:
            self._latest_error = message

    def _clear_error(self) -> None:
        with self._lock:
            self._latest_error = None

    @staticmethod
    def _release_capture_object(capture: Any | None) -> None:
        if capture is None:
            return
        try:
            capture.release()
        except Exception:
            LOGGER.warning("Release camera failed.", exc_info=True)


def _normalize_box(value: object) -> tuple[int, int, int, int] | None:
    if not isinstance(value, (tuple, list)) or len(value) != 4:
        return None
    try:
        x, y, width, height = (int(item) for item in value)
    except (TypeError, ValueError):
        return None
    if width <= 0 or height <= 0:
        return None
    return x, y, width, height


def _detect_largest_face_box(
    frame_rgb: np.ndarray,
    cv2_module: Any,
    runtime: FaceCameraRuntime,
) -> tuple[int, int, int, int] | None:
    if not all(hasattr(cv2_module, attr) for attr in ("CascadeClassifier", "cvtColor")):
        return None
    try:
        detector = runtime._haar_cascade
        if detector is None:
            cascade_path = None
            data = getattr(cv2_module, "data", None)
            haarcascades = getattr(data, "haarcascades", None)
            if haarcascades:
                from pathlib import Path

                for name in (
                    "haarcascade_frontalface_default.xml",
                    "haarcascade_frontalface_alt.xml",
                    "haarcascade_frontalface_alt2.xml",
                ):
                    candidate = Path(haarcascades) / name
                    if candidate.exists():
                        cascade_path = str(candidate)
                        break
            if cascade_path is None:
                return None
            detector = cv2_module.CascadeClassifier(cascade_path)
            if hasattr(detector, "empty") and detector.empty():
                return None
            runtime._haar_cascade = detector

        gray = cv2_module.cvtColor(frame_rgb, cv2_module.COLOR_RGB2GRAY)
        faces = detector.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=4, minSize=(48, 48))
        if len(faces) == 0:
            return None
        x, y, width, height = max(faces, key=lambda box: int(box[2]) * int(box[3]))
        return int(x), int(y), int(width), int(height)
    except Exception:
        LOGGER.debug("Lightweight preview face tracking failed.", exc_info=True)
        return None


def _draw_preview_overlay(
    frame_rgb: np.ndarray,
    cv2_module: Any,
    *,
    face_box: tuple[int, int, int, int] | None,
    source_shape: tuple[int, ...],
    result: dict[str, object] | None,
    result_at: float | None,
    analysis_in_progress: bool,
) -> np.ndarray:
    output = frame_rgb.copy()
    if not all(hasattr(cv2_module, attr) for attr in ("rectangle", "putText")):
        return output

    scale_x = output.shape[1] / max(1, int(source_shape[1]))
    scale_y = output.shape[0] / max(1, int(source_shape[0]))
    if face_box is not None:
        x, y, width, height = face_box
        x0 = int(x * scale_x)
        y0 = int(y * scale_y)
        x1 = int((x + width) * scale_x)
        y1 = int((y + height) * scale_y)
        cv2_module.rectangle(output, (x0, y0), (x1, y1), (60, 220, 90), 2)

    lines = _overlay_lines(result, result_at, analysis_in_progress)
    font = getattr(cv2_module, "FONT_HERSHEY_SIMPLEX", 0)
    line_height = 22
    box_width = min(output.shape[1] - 16, 310)
    box_height = max(48, 12 + len(lines) * line_height)
    cv2_module.rectangle(output, (8, 8), (8 + box_width, 8 + box_height), (8, 18, 34), -1)
    cv2_module.rectangle(output, (8, 8), (8 + box_width, 8 + box_height), (80, 170, 255), 1)
    for index, line in enumerate(lines):
        y = 30 + index * line_height
        cv2_module.putText(output, line, (18, y), font, 0.52, (245, 248, 255), 1, getattr(cv2_module, "LINE_AA", 16))
    return output


def _overlay_lines(
    result: dict[str, object] | None,
    result_at: float | None,
    analysis_in_progress: bool,
) -> list[str]:
    if result is None:
        state = "Analyzing..." if analysis_in_progress else "Waiting for face analysis"
        return [state, "Emotion: --  Conf: --"]

    if not result.get("available"):
        message = str(result.get("error_code") or "no_face")
        return ["No face detected", f"Status: {message}"]

    updated = "--"
    if result_at is not None:
        age = max(0.0, time.time() - float(result_at))
        updated = f"{age:.0f}s ago"
    emotion = str(result.get("emotion") or "--")
    confidence = _format_overlay_float(result.get("confidence"))
    quality = _format_overlay_float(result.get("quality"))
    valence = _format_overlay_float(result.get("valence"))
    arousal = _format_overlay_float(result.get("arousal"))
    return [
        f"Emotion: {emotion}  Conf: {confidence}",
        f"Quality: {quality}  Updated: {updated}",
        f"V: {valence}  A: {arousal}",
    ]


def _format_overlay_float(value: object) -> str:
    try:
        return f"{float(value):.2f}"
    except (TypeError, ValueError):
        return "--"