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 "--"