diff --git a/.env.example b/.env.example index ac4b679..d4e484c 100644 --- a/.env.example +++ b/.env.example @@ -175,3 +175,15 @@ FEATURE_FACE_VERIFICATION=True FEATURE_LIVENESS_DETECTION=True FEATURE_BATCH_PROCESSING=True FEATURE_QUALITY_ASSESSMENT=True + +# ============================================================================ +# MediaPipe Face Landmarker (Tasks API — ported 2026-05-12) +# ---------------------------------------------------------------------------- +# The runtime container bakes face_landmarker.task at /app/models/. Override +# the path in dev to point at a local copy, or in prod to point at an +# operator-rotated asset. SHA256 is enforced when set. +# ============================================================================ +FACE_LANDMARKER_MODEL_PATH=/app/models/face_landmarker.task +# float16/latest as of 2026-05-12; verify with: +# sha256sum /app/models/face_landmarker.task +FACE_LANDMARKER_MODEL_SHA256=64184e229b263107bc2b804c6625db1341ff2bb731874b0bcc2fe6544e0bc9ff diff --git a/Dockerfile b/Dockerfile index e148d07..3194226 100644 --- a/Dockerfile +++ b/Dockerfile @@ -58,6 +58,23 @@ RUN python -c "import cv2; print('OpenCV version:', cv2.__version__)" && \ python -c "import numpy; print('NumPy version:', numpy.__version__)" && \ python -c "import tensorflow; print('TensorFlow version:', tensorflow.__version__)" +# Bake MediaPipe face_landmarker.task into the image (2026-05-12). +# The new Tasks API (mp.tasks.vision.FaceLandmarker) requires a .task model +# asset; without it the gaze tracker, active-liveness detector, quality +# assessor and landmark detector all fail-soft with "model missing". +# Pinning + SHA-verifying at build time gives us a reproducible image and +# closes the supply-chain check the loader expects at runtime. +# Companion env vars in .env.example: +# FACE_LANDMARKER_MODEL_PATH=/app/models/face_landmarker.task +# FACE_LANDMARKER_MODEL_SHA256=64184e229b263107bc2b804c6625db1341ff2bb731874b0bcc2fe6544e0bc9ff +ARG FACE_LANDMARKER_SHA256=64184e229b263107bc2b804c6625db1341ff2bb731874b0bcc2fe6544e0bc9ff +RUN set -eux; \ + mkdir -p /app/models; \ + curl -fsSL -o /app/models/face_landmarker.task \ + "https://storage.googleapis.com/mediapipe-models/face_landmarker/face_landmarker/float16/latest/face_landmarker.task"; \ + echo "${FACE_LANDMARKER_SHA256} /app/models/face_landmarker.task" | sha256sum -c -; \ + chmod 0644 /app/models/face_landmarker.task + # Copy application code COPY . . diff --git a/app/core/config.py b/app/core/config.py index 41c19e5..e8787b0 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -693,11 +693,24 @@ def get_api_key_config(self) -> dict: ".env.prod when shipping; empty = skip verification (dev only)." ), ) + FACE_LANDMARKER_MODEL_PATH: str = Field( + default=str(_REPO_ROOT / "models" / "face_landmarker.task"), + description=( + "Filesystem path to the MediaPipe face_landmarker.task asset. " + "Used by server-side facial-landmark consumers (gaze tracker, " + "active liveness, quality assessor) after the 2026-05-12 port " + "from mp.solutions to mp.tasks.vision. The runtime container " + "bakes this file at /app/models/face_landmarker.task; override " + "via the env var to point at an operator-rotated asset." + ), + ) FACE_LANDMARKER_MODEL_SHA256: str = Field( default="", description=( "Expected SHA256 hex digest for face_landmarker.task. Set in " - ".env.prod; empty = skip verification with a log warning." + ".env.prod; empty = skip verification with a log warning. " + "Live asset SHA256 (float16/latest, 2026-05-12): " + "64184e229b263107bc2b804c6625db1341ff2bb731874b0bcc2fe6544e0bc9ff" ), ) diff --git a/app/infrastructure/ml/landmarks/face_landmarker_loader.py b/app/infrastructure/ml/landmarks/face_landmarker_loader.py new file mode 100644 index 0000000..4da5807 --- /dev/null +++ b/app/infrastructure/ml/landmarks/face_landmarker_loader.py @@ -0,0 +1,165 @@ +"""Shared MediaPipe Face Landmarker (Tasks API) loader. + +Centralises the `.task` model resolution + SHA256 integrity check used by +every server-side consumer of facial landmarks. Replaces the deprecated +`mediapipe.solutions.face_mesh` API (removed in mediapipe 0.10.35). + +The loader honours: +- ``FACE_LANDMARKER_MODEL_PATH`` env var (per-env override). +- A repo-relative ``./models/face_landmarker.task`` default. +- ``FACE_LANDMARKER_MODEL_SHA256`` env var for integrity verification + (warn-and-disable on mismatch; production MUST set this). + +Consumers should treat the returned object as opaque and call ``.detect()`` +or ``.detect_for_video()`` against an ``mp.Image`` instance. + +This module DOES NOT cache the loader globally; each consumer keeps a +process-local cache appropriate to its lifecycle (singleton vs. per-request). +""" + +from __future__ import annotations + +import hashlib +import logging +import os +from pathlib import Path +from typing import Any, Optional + +logger = logging.getLogger(__name__) + +# Default location: /models/face_landmarker.task. The file is gitignored +# but is baked into the runtime container by the Dockerfile (see PR +# `port: migrate from mp.solutions to mp.tasks.vision`). +_DEFAULT_MODEL_PATH = ( + Path(__file__).resolve().parent.parent.parent.parent / "models" / "face_landmarker.task" +) + + +def resolve_face_landmarker_model_path() -> Optional[Path]: + """Return the on-disk path to face_landmarker.task or None if absent. + + Resolution order: + 1. ``FACE_LANDMARKER_MODEL_PATH`` env var (if set, must exist). + 2. ``/models/face_landmarker.task``. + 3. ``./models/face_landmarker.task`` relative to cwd (dev fallback). + """ + env_path = os.getenv("FACE_LANDMARKER_MODEL_PATH", "").strip() + candidates = [] + if env_path: + candidates.append(Path(env_path)) + candidates.append(_DEFAULT_MODEL_PATH) + candidates.append(Path("models/face_landmarker.task")) + + for candidate in candidates: + if candidate.exists(): + return candidate + return None + + +def verify_model_sha256(model_path: Path) -> bool: + """Verify the SHA256 of the model file against ``FACE_LANDMARKER_MODEL_SHA256``. + + Returns True when: + - The env var is unset (verification skipped — dev only). + - The env var matches the on-disk file. + + Returns False when the env var is set and the digest mismatches. + Logs a warning either way. + """ + expected = os.getenv("FACE_LANDMARKER_MODEL_SHA256", "").strip() + if not expected: + logger.warning( + "FACE_LANDMARKER_MODEL_SHA256 not set — model integrity NOT verified. " + "Set this in production to prevent supply-chain tampering." + ) + return True + try: + actual = hashlib.sha256(model_path.read_bytes()).hexdigest() + except OSError as exc: + logger.error("Could not read face_landmarker.task for SHA256 check: %s", exc) + return False + if actual.lower() != expected.lower(): + logger.error( + "face_landmarker.task SHA256 mismatch: expected=%s actual=%s path=%s", + expected, + actual, + model_path, + ) + return False + return True + + +def create_face_landmarker( + *, + static_image_mode: bool = True, + num_faces: int = 1, + output_face_blendshapes: bool = False, + min_face_detection_confidence: float = 0.5, + min_face_presence_confidence: float = 0.5, + min_tracking_confidence: float = 0.5, +) -> Optional[Any]: + """Create a MediaPipe FaceLandmarker (Tasks API) or return None on failure. + + Args mirror the relevant subset of the legacy ``mp.solutions.face_mesh`` + constructor so call sites can be ported in-place. ``static_image_mode`` + maps to ``RunningMode.IMAGE`` (True) or ``RunningMode.VIDEO`` (False). + + Returns: + FaceLandmarker instance ready for ``.detect(mp_image)`` (IMAGE mode) or + ``.detect_for_video(mp_image, timestamp_ms)`` (VIDEO mode). Returns + ``None`` if MediaPipe is missing, the model file is absent, or the + SHA256 check fails — callers should fail-soft. + """ + try: + from mediapipe.tasks import python as mp_python + from mediapipe.tasks.python import vision as mp_vision + except ImportError as exc: + logger.error("MediaPipe Tasks API not importable: %s", exc) + return None + + model_path = resolve_face_landmarker_model_path() + if model_path is None: + logger.warning( + "face_landmarker.task not found. Set FACE_LANDMARKER_MODEL_PATH or " + "place the asset under /models/." + ) + return None + if not verify_model_sha256(model_path): + return None + + try: + running_mode = mp_vision.RunningMode.IMAGE if static_image_mode else mp_vision.RunningMode.VIDEO + base_options = mp_python.BaseOptions(model_asset_path=str(model_path)) + options = mp_vision.FaceLandmarkerOptions( + base_options=base_options, + running_mode=running_mode, + num_faces=num_faces, + min_face_detection_confidence=min_face_detection_confidence, + min_face_presence_confidence=min_face_presence_confidence, + min_tracking_confidence=min_tracking_confidence, + output_face_blendshapes=output_face_blendshapes, + ) + landmarker = mp_vision.FaceLandmarker.create_from_options(options) + logger.info( + "MediaPipe FaceLandmarker (Tasks API) initialised " + "(mode=%s, num_faces=%s, blendshapes=%s, model=%s)", + running_mode.name, + num_faces, + output_face_blendshapes, + model_path, + ) + return landmarker + except Exception: # noqa: BLE001 + logger.exception("Failed to create FaceLandmarker from Tasks API") + return None + + +def to_mp_image(image_rgb): + """Wrap an RGB numpy array in an ``mp.Image`` (SRGB format). + + Helper so call sites don't need ``import mediapipe`` just for the format + enum. The caller is responsible for BGR→RGB conversion. + """ + import mediapipe as mp + + return mp.Image(image_format=mp.ImageFormat.SRGB, data=image_rgb) diff --git a/app/infrastructure/ml/landmarks/mediapipe_landmarks.py b/app/infrastructure/ml/landmarks/mediapipe_landmarks.py index 042b5ee..be376d6 100644 --- a/app/infrastructure/ml/landmarks/mediapipe_landmarks.py +++ b/app/infrastructure/ml/landmarks/mediapipe_landmarks.py @@ -1,4 +1,10 @@ -"""MediaPipe-based facial landmark detector implementation.""" +"""MediaPipe-based facial landmark detector implementation. + +Ported 2026-05-12 from the legacy ``mediapipe.solutions.face_mesh`` API to +``mediapipe.tasks.vision.FaceLandmarker``. The ``mp.solutions`` namespace was +removed in mediapipe 0.10.35; the new Tasks API requires a ``.task`` model +asset and exposes landmarks as ``result.face_landmarks[0][i].(x|y|z)``. +""" import logging from typing import List, Optional @@ -7,17 +13,23 @@ from app.domain.entities.face_landmarks import HeadPose, Landmark, LandmarkResult from app.domain.exceptions.feature_errors import LandmarkError +from app.infrastructure.ml.landmarks.face_landmarker_loader import ( + create_face_landmarker, + to_mp_image, +) logger = logging.getLogger(__name__) class MediaPipeLandmarkDetector: - """Facial landmark detector using MediaPipe Face Mesh. + """Facial landmark detector using MediaPipe Face Landmarker (Tasks API). Detects 468 facial landmarks with optional 3D coordinates. """ - # Facial region indices for MediaPipe Face Mesh + # Facial region indices for MediaPipe Face Mesh (canonical 468-pt topology; + # indices are stable between the legacy face_mesh and the Tasks-API + # face_landmarker outputs). REGIONS = { "left_eye": [33, 133, 160, 159, 158, 144, 145, 153], "right_eye": [362, 263, 387, 386, 385, 373, 374, 380], @@ -34,21 +46,23 @@ class MediaPipeLandmarkDetector: def __init__(self) -> None: """Initialize MediaPipe landmark detector.""" - self._face_mesh = None - logger.info("MediaPipeLandmarkDetector initialized") - - def _get_face_mesh(self): - """Lazy load MediaPipe Face Mesh.""" - if self._face_mesh is None: - import mediapipe as mp + self._face_landmarker = None + logger.info("MediaPipeLandmarkDetector initialized (Tasks API)") - self._face_mesh = mp.solutions.face_mesh.FaceMesh( + def _get_face_landmarker(self): + """Lazy load MediaPipe Face Landmarker (Tasks API).""" + if self._face_landmarker is None: + self._face_landmarker = create_face_landmarker( static_image_mode=True, - max_num_faces=1, - refine_landmarks=True, - min_detection_confidence=0.5, + num_faces=1, + min_face_detection_confidence=0.5, ) - return self._face_mesh + if self._face_landmarker is None: + raise LandmarkError( + "MediaPipe FaceLandmarker unavailable — model asset missing or " + "Tasks API not importable. See logs for details." + ) + return self._face_landmarker def detect( self, image: np.ndarray, include_3d: bool = False @@ -68,19 +82,25 @@ def detect( logger.debug(f"Starting landmark detection (include_3d={include_3d})") try: - face_mesh = self._get_face_mesh() - results = face_mesh.process(image) - - if not results.multi_face_landmarks: + face_landmarker = self._get_face_landmarker() + # Tasks API expects an mp.Image wrapping an RGB ndarray. Callers + # of this method already pass RGB (see the docstring), so no + # additional colour-space conversion is needed. + mp_image = to_mp_image(image) + result = face_landmarker.detect(mp_image) + + face_landmarks_list = result.face_landmarks or [] + if not face_landmarks_list: raise LandmarkError("No face landmarks detected") - # Get first face landmarks - face_landmarks = results.multi_face_landmarks[0] + # Get first face landmarks. In the Tasks API each element is itself + # a flat list of NormalizedLandmark (no `.landmark` attribute). + face_landmarks = face_landmarks_list[0] height, width = image.shape[:2] # Extract landmarks landmarks = [] - for idx, lm in enumerate(face_landmarks.landmark): + for idx, lm in enumerate(face_landmarks): x = int(lm.x * width) y = int(lm.y * height) z = lm.z if include_3d else None diff --git a/app/infrastructure/ml/liveness/active_liveness_detector.py b/app/infrastructure/ml/liveness/active_liveness_detector.py index 6ae9090..b7fb053 100644 --- a/app/infrastructure/ml/liveness/active_liveness_detector.py +++ b/app/infrastructure/ml/liveness/active_liveness_detector.py @@ -1,7 +1,11 @@ """Active liveness detector using facial landmark analysis. -This detector uses MediaPipe Face Mesh to detect facial landmarks -and analyze facial actions (smile, blink) for liveness verification. +This detector uses MediaPipe Face Landmarker (Tasks API) to detect facial +landmarks and analyze facial actions (smile, blink) for liveness verification. + +Ported 2026-05-12 from ``mp.solutions.face_mesh`` to +``mp.tasks.vision.FaceLandmarker`` (the legacy API was removed in mediapipe +0.10.35). """ import logging @@ -12,6 +16,10 @@ from app.domain.entities.liveness_result import LivenessResult from app.domain.interfaces.liveness_detector import ILivenessDetector +from app.infrastructure.ml.landmarks.face_landmarker_loader import ( + create_face_landmarker, + to_mp_image, +) logger = logging.getLogger(__name__) @@ -63,31 +71,31 @@ def __init__( self._liveness_threshold = liveness_threshold self._min_detection_confidence = min_detection_confidence self._min_tracking_confidence = min_tracking_confidence - self._face_mesh = None + self._face_landmarker = None logger.info( - f"ActiveLivenessDetector initialized: " + f"ActiveLivenessDetector initialized (Tasks API): " f"EAR threshold={ear_threshold}, MAR threshold={mar_threshold}, " f"liveness threshold={liveness_threshold}" ) - def _get_face_mesh(self): - """Lazy initialization of MediaPipe Face Mesh.""" - if self._face_mesh is None: - try: - import mediapipe as mp - self._face_mesh = mp.solutions.face_mesh.FaceMesh( - static_image_mode=True, - max_num_faces=1, - refine_landmarks=True, - min_detection_confidence=self._min_detection_confidence, - min_tracking_confidence=self._min_tracking_confidence, + def _get_face_landmarker(self): + """Lazy initialization of MediaPipe Face Landmarker (Tasks API).""" + if self._face_landmarker is None: + self._face_landmarker = create_face_landmarker( + static_image_mode=True, + num_faces=1, + min_face_detection_confidence=self._min_detection_confidence, + min_face_presence_confidence=self._min_detection_confidence, + min_tracking_confidence=self._min_tracking_confidence, + ) + if self._face_landmarker is None: + raise RuntimeError( + "MediaPipe FaceLandmarker unavailable — model asset missing " + "or Tasks API not importable. See logs for details." ) - logger.info("MediaPipe Face Mesh initialized") - except ImportError: - logger.error("MediaPipe not installed. Run: pip install mediapipe") - raise - return self._face_mesh + logger.info("MediaPipe Face Landmarker initialized for active liveness") + return self._face_landmarker async def check_liveness(self, image: np.ndarray) -> LivenessResult: """Check if image shows a live person using facial action analysis. @@ -119,11 +127,13 @@ async def detect( # Convert BGR to RGB for MediaPipe rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) - # Get facial landmarks - face_mesh = self._get_face_mesh() - results = face_mesh.process(rgb_image) + # Get facial landmarks via Tasks API + face_landmarker = self._get_face_landmarker() + mp_image = to_mp_image(rgb_image) + result = face_landmarker.detect(mp_image) - if not results.multi_face_landmarks: + face_landmarks_list = result.face_landmarks or [] + if not face_landmarks_list: logger.warning("No face landmarks detected") return LivenessResult( is_live=False, @@ -138,7 +148,7 @@ async def detect( }, ) - landmarks = results.multi_face_landmarks[0].landmark + landmarks = face_landmarks_list[0] h, w = image.shape[:2] # Convert normalized landmarks to pixel coordinates diff --git a/app/infrastructure/ml/proctoring/mediapipe_gaze_tracker.py b/app/infrastructure/ml/proctoring/mediapipe_gaze_tracker.py index becd223..e59d6e4 100644 --- a/app/infrastructure/ml/proctoring/mediapipe_gaze_tracker.py +++ b/app/infrastructure/ml/proctoring/mediapipe_gaze_tracker.py @@ -1,10 +1,18 @@ """MediaPipe-based gaze tracker implementation for proctoring. -Uses MediaPipe Face Mesh (468 landmarks) to track gaze direction -and head pose for detecting if the user is looking at the screen. +Uses MediaPipe Face Landmarker (Tasks API) to track gaze direction and head +pose for detecting if the user is looking at the screen. + +Ported 2026-05-12 from the legacy ``mediapipe.solutions.face_mesh`` API to +``mediapipe.tasks.vision.FaceLandmarker``. The canonical ``face_landmarker.task`` +asset returns 478 landmarks (468 face + 10 iris) so the iris indices below +(468-477) remain valid without the legacy ``refine_landmarks=True`` toggle. +Frame submission uses VIDEO running-mode with monotonically increasing +``timestamp_ms`` so the Tasks API can do its own temporal smoothing. """ import logging +import time from datetime import datetime from typing import Dict, List, Optional, Tuple @@ -17,6 +25,10 @@ HeadPose, ) from app.domain.interfaces.gaze_tracker import IGazeTracker +from app.infrastructure.ml.landmarks.face_landmarker_loader import ( + create_face_landmarker, + to_mp_image, +) logger = logging.getLogger(__name__) @@ -77,34 +89,45 @@ def __init__( self._min_tracking_confidence = min_tracking_confidence self._gaze_threshold = gaze_threshold self._pitch_threshold, self._yaw_threshold = head_pose_threshold - self._face_mesh = None + self._face_landmarker = None + # Monotonic timestamp counter for VIDEO running-mode. The Tasks API + # rejects non-increasing timestamps with InvalidArgumentError, so we + # never feed it wall-clock values directly. + self._frame_timestamp_ms = 0 # Track off-screen start per session to avoid accumulating across sessions self._off_screen_start: Dict[str, Optional[datetime]] = {} logger.info( - f"MediaPipeGazeTracker initialized: " + f"MediaPipeGazeTracker initialized (Tasks API): " f"gaze_threshold={gaze_threshold}, " f"head_pose_threshold={head_pose_threshold}" ) - def _get_face_mesh(self): - """Lazy initialization of MediaPipe Face Mesh.""" - if self._face_mesh is None: - try: - import mediapipe as mp - - self._face_mesh = mp.solutions.face_mesh.FaceMesh( - static_image_mode=False, # Video mode for tracking - max_num_faces=1, - refine_landmarks=True, # Required for iris landmarks - min_detection_confidence=self._min_detection_confidence, - min_tracking_confidence=self._min_tracking_confidence, + def _get_face_landmarker(self): + """Lazy initialization of MediaPipe Face Landmarker (Tasks API).""" + if self._face_landmarker is None: + self._face_landmarker = create_face_landmarker( + static_image_mode=False, # VIDEO running mode for tracking + num_faces=1, + min_face_detection_confidence=self._min_detection_confidence, + min_face_presence_confidence=self._min_detection_confidence, + min_tracking_confidence=self._min_tracking_confidence, + ) + if self._face_landmarker is None: + raise RuntimeError( + "MediaPipe FaceLandmarker unavailable for gaze tracking — " + "asset missing or Tasks API not importable." ) - logger.info("MediaPipe Face Mesh initialized for gaze tracking") - except ImportError: - logger.error("MediaPipe not installed. Run: pip install mediapipe") - raise - return self._face_mesh + logger.info("MediaPipe Face Landmarker initialized for gaze tracking") + return self._face_landmarker + + def _next_video_timestamp_ms(self) -> int: + """Return a strictly monotonic timestamp for VIDEO running-mode.""" + # Use a monotonic counter rather than wall-clock so repeated calls in + # the same millisecond still get unique, ordered timestamps. + next_ts = max(self._frame_timestamp_ms + 1, int(time.monotonic() * 1000)) + self._frame_timestamp_ms = next_ts + return next_ts async def analyze( self, @@ -126,11 +149,15 @@ async def analyze( rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) h, w = image.shape[:2] - # Get facial landmarks - face_mesh = self._get_face_mesh() - results = face_mesh.process(rgb_image) + # Get facial landmarks via Tasks API + face_landmarker = self._get_face_landmarker() + mp_image = to_mp_image(rgb_image) + result = face_landmarker.detect_for_video( + mp_image, self._next_video_timestamp_ms() + ) - if not results.multi_face_landmarks: + face_landmarks_list = result.face_landmarks or [] + if not face_landmarks_list: logger.debug("No face detected for gaze tracking") duration = self._get_off_screen_duration(timestamp, is_off_screen=True, session_id=session_id) return GazeAnalysisResult( @@ -143,7 +170,7 @@ async def analyze( duration_off_screen_sec=duration, ) - landmarks = results.multi_face_landmarks[0].landmark + landmarks = face_landmarks_list[0] # Convert to pixel coordinates landmark_points = [(lm.x * w, lm.y * h, lm.z) for lm in landmarks] @@ -191,13 +218,17 @@ async def get_head_pose( rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) h, w = image.shape[:2] - face_mesh = self._get_face_mesh() - results = face_mesh.process(rgb_image) + face_landmarker = self._get_face_landmarker() + mp_image = to_mp_image(rgb_image) + result = face_landmarker.detect_for_video( + mp_image, self._next_video_timestamp_ms() + ) - if not results.multi_face_landmarks: + face_landmarks_list = result.face_landmarks or [] + if not face_landmarks_list: return None - landmarks = results.multi_face_landmarks[0].landmark + landmarks = face_landmarks_list[0] landmark_points = [(lm.x * w, lm.y * h, lm.z) for lm in landmarks] return self._estimate_head_pose(landmark_points, w, h) diff --git a/app/infrastructure/ml/quality/quality_assessor.py b/app/infrastructure/ml/quality/quality_assessor.py index a489218..5598fe1 100644 --- a/app/infrastructure/ml/quality/quality_assessor.py +++ b/app/infrastructure/ml/quality/quality_assessor.py @@ -190,24 +190,42 @@ def _estimate_pose(face_image: np.ndarray): Tuple (yaw_degrees, pitch_degrees) or (None, None) if unavailable. """ try: - import mediapipe as mp + # Ported 2026-05-12 from mp.solutions.face_mesh to the Tasks API. + # Creating a fresh FaceLandmarker per call matches the old + # `with FaceMesh(...)` lifecycle and is cheap enough at the call + # frequency this static method sees (one PnP solve per quality + # check). + from app.infrastructure.ml.landmarks.face_landmarker_loader import ( + create_face_landmarker, + to_mp_image, + ) h, w = face_image.shape[:2] - mp_face_mesh = mp.solutions.face_mesh - with mp_face_mesh.FaceMesh( + face_landmarker = create_face_landmarker( static_image_mode=True, - max_num_faces=1, - refine_landmarks=False, - min_detection_confidence=0.5, - ) as face_mesh: + num_faces=1, + min_face_detection_confidence=0.5, + ) + if face_landmarker is None: + return None, None + try: rgb = cv2.cvtColor(face_image, cv2.COLOR_BGR2RGB) - results = face_mesh.process(rgb) - - if not results.multi_face_landmarks: + mp_image = to_mp_image(rgb) + result = face_landmarker.detect(mp_image) + finally: + # Tasks API requires explicit close to release the underlying + # C++ graph; without this we'd leak FDs on long-running workers. + try: + face_landmarker.close() + except Exception: # noqa: BLE001 + pass + + face_landmarks_list = result.face_landmarks or [] + if not face_landmarks_list: return None, None - lm = results.multi_face_landmarks[0].landmark + lm = face_landmarks_list[0] # 2D image points (selected canonical landmarks) # Indices: nose tip=1, chin=152, left eye left=33, right eye right=263, diff --git a/tests/demo_local.py b/tests/demo_local.py index 00921c6..43ac485 100644 --- a/tests/demo_local.py +++ b/tests/demo_local.py @@ -375,9 +375,9 @@ def __init__(self, camera_id: int = 0, mode: str = "all"): # ML Components self._deepface = None - self._mp_face_mesh = None # Legacy Solutions API - self._mp_face_landmarker = None # New Tasks API - self._mp_use_tasks_api = False + # 2026-05-12: legacy Solutions API removed from mediapipe 0.10.35; + # we keep only the Tasks-API path. + self._mp_face_landmarker = None self._mediapipe_loaded = False self._quality_assessor = SimpleQualityAssessor() self._liveness_detector = SimpleLivenessDetector() @@ -465,55 +465,34 @@ def _init_ml(self): print("[2/3] Loading MediaPipe (468 landmarks)...") self._mediapipe_loaded = False - self._mp_face_mesh = None - self._mp_use_tasks_api = False try: - import mediapipe as mp - - # Try new Tasks API first (MediaPipe 0.10.14+) - if hasattr(mp, 'tasks'): - try: - from mediapipe.tasks import python as mp_tasks - from mediapipe.tasks.python import vision - - # Download model if needed - import urllib.request - import os - model_path = "face_landmarker.task" - if not os.path.exists(model_path): - print(" Downloading face landmark model...") - url = "https://storage.googleapis.com/mediapipe-models/face_landmarker/face_landmarker/float16/1/face_landmarker.task" - urllib.request.urlretrieve(url, model_path) - - base_options = mp_tasks.BaseOptions(model_asset_path=model_path) - options = vision.FaceLandmarkerOptions( - base_options=base_options, - output_face_blendshapes=False, - output_facial_transformation_matrixes=False, - num_faces=5 - ) - self._mp_face_landmarker = vision.FaceLandmarker.create_from_options(options) - self._mp_use_tasks_api = True - self._mediapipe_loaded = True - print(" MediaPipe Tasks API ready!") - except Exception as e: - print(f" Tasks API failed: {e}") - - # Fall back to legacy solutions API - if not self._mediapipe_loaded and hasattr(mp, 'solutions'): - self._mp_face_mesh = mp.solutions.face_mesh.FaceMesh( - static_image_mode=False, - max_num_faces=5, - refine_landmarks=True, - min_detection_confidence=0.5, - min_tracking_confidence=0.5 + from mediapipe.tasks import python as mp_tasks + from mediapipe.tasks.python import vision + + # Download model if needed + import urllib.request + import os + model_path = "face_landmarker.task" + if not os.path.exists(model_path): + print(" Downloading face landmark model...") + url = ( + "https://storage.googleapis.com/mediapipe-models/" + "face_landmarker/face_landmarker/float16/latest/" + "face_landmarker.task" ) - self._mediapipe_loaded = True - print(" MediaPipe Solutions API ready!") - - if not self._mediapipe_loaded: - print(" MediaPipe: No compatible API found") + urllib.request.urlretrieve(url, model_path) + + base_options = mp_tasks.BaseOptions(model_asset_path=model_path) + options = vision.FaceLandmarkerOptions( + base_options=base_options, + output_face_blendshapes=False, + output_facial_transformation_matrixes=False, + num_faces=5, + ) + self._mp_face_landmarker = vision.FaceLandmarker.create_from_options(options) + self._mediapipe_loaded = True + print(" MediaPipe Tasks API ready!") except Exception as e: print(f" MediaPipe unavailable: {e}") @@ -716,8 +695,8 @@ def detect_landmarks(self, frame: np.ndarray) -> List[List[Tuple[int, int]]]: h, w = frame.shape[:2] try: - # Use Tasks API (MediaPipe 0.10.14+) - if self._mp_use_tasks_api and hasattr(self, '_mp_face_landmarker'): + # MediaPipe Tasks API (legacy Solutions API removed in 0.10.35). + if self._mp_face_landmarker is not None: import mediapipe as mp rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb) @@ -732,16 +711,6 @@ def detect_landmarks(self, frame: np.ndarray) -> List[List[Tuple[int, int]]]: all_landmarks.append(points) self._landmarks_cache = all_landmarks - # Use legacy Solutions API - elif self._mp_face_mesh is not None: - rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) - results = self._mp_face_mesh.process(rgb) - if not results.multi_face_landmarks: - self._landmarks_cache = [] - else: - self._landmarks_cache = [[(int(lm.x * w), int(lm.y * h)) for lm in face.landmark] - for face in results.multi_face_landmarks] - self._last_landmarks_time = current_time except Exception as e: diff --git a/tests/integration/test_no_mp_solutions_runtime.py b/tests/integration/test_no_mp_solutions_runtime.py new file mode 100644 index 0000000..5c4caf0 --- /dev/null +++ b/tests/integration/test_no_mp_solutions_runtime.py @@ -0,0 +1,100 @@ +"""Integration: confirm no runtime code path imports mediapipe.solutions. + +Added 2026-05-12 after the mp.solutions → mp.tasks.vision port. + +The legacy ``mp.solutions`` namespace was removed in mediapipe 0.10.35. Any +remaining call site would fail at runtime with: + + AttributeError: module 'mediapipe' has no attribute 'solutions' + +The /verify route is the primary symptom surface today (every call was +logging ``Landmark detection failed: module 'mediapipe' has no attribute +'solutions'``). Rather than spinning up the full FastAPI stack and asserting +on the absence of a log line — which is brittle — this test does two things: + +1. AST-scans the app/ tree for executable references to mp.solutions. Any + non-comment, non-docstring reference fails the test. +2. Verifies the ported call sites can import without crashing. + +Comments and docstrings that *mention* the migration are still permitted. +""" + +from __future__ import annotations + +import ast +import pathlib + +REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent.parent +APP_ROOT = REPO_ROOT / "app" + + +def _executable_references_to_mp_solutions(py_file: pathlib.Path) -> list[str]: + """Return locations of executable mp.solutions references in py_file. + + AST walks the module; anything inside a string literal (docstring or + comment-equivalent string assignment) is invisible to the AST, so this + naturally ignores comments and docstring references. + """ + try: + tree = ast.parse(py_file.read_text(encoding="utf-8")) + except SyntaxError: + return [] + + offenders: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Attribute): + # Walk back along the attribute chain to find the root identifier. + current = node + attr_chain = [] + while isinstance(current, ast.Attribute): + attr_chain.append(current.attr) + current = current.value + if isinstance(current, ast.Name): + attr_chain.append(current.id) + attr_chain.reverse() + # Match patterns: mp.solutions, mediapipe.solutions + joined = ".".join(attr_chain) + if joined.startswith("mp.solutions") or joined.startswith( + "mediapipe.solutions" + ): + offenders.append(f"{py_file}:{node.lineno}: {joined}") + return offenders + + +def test_no_executable_mp_solutions_references_in_app() -> None: + """No production code under app/ may reference mp.solutions executable-side.""" + offenders: list[str] = [] + for py_file in APP_ROOT.rglob("*.py"): + if "__pycache__" in py_file.parts: + continue + offenders.extend(_executable_references_to_mp_solutions(py_file)) + + assert not offenders, ( + "Found executable references to mp.solutions / mediapipe.solutions:\n" + + "\n".join(offenders) + + "\n\nThe mp.solutions namespace was removed in mediapipe 0.10.35. " + "Port to mp.tasks.vision.FaceLandmarker — see " + "app/infrastructure/ml/landmarks/face_landmarker_loader.py." + ) + + +def test_ported_modules_import_cleanly() -> None: + """The four ported modules must import without crashing in prod env.""" + # If any of these still referenced mp.solutions executable-side, the + # import itself would not crash (the reference is inside lazy-init), but + # any subsequent .detect() call would. The previous test (AST) catches + # the executable references; this test catches import-time syntax / + # name errors introduced by the port. + from app.infrastructure.ml.landmarks import face_landmarker_loader # noqa: F401 + from app.infrastructure.ml.landmarks.mediapipe_landmarks import ( # noqa: F401 + MediaPipeLandmarkDetector, + ) + from app.infrastructure.ml.liveness.active_liveness_detector import ( # noqa: F401 + ActiveLivenessDetector, + ) + from app.infrastructure.ml.proctoring.mediapipe_gaze_tracker import ( # noqa: F401 + MediaPipeGazeTracker, + ) + from app.infrastructure.ml.quality.quality_assessor import ( # noqa: F401 + QualityAssessor, + ) diff --git a/tests/unit/infrastructure/test_active_liveness_detector_tasks_api.py b/tests/unit/infrastructure/test_active_liveness_detector_tasks_api.py new file mode 100644 index 0000000..c628c24 --- /dev/null +++ b/tests/unit/infrastructure/test_active_liveness_detector_tasks_api.py @@ -0,0 +1,85 @@ +"""Tests for ActiveLivenessDetector after the mp.solutions → mp.tasks port. + +Added 2026-05-12. Verifies the detector adapts to the new +``result.face_landmarks[0][i].x`` shape and still returns a coherent +LivenessResult. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest import mock + +import numpy as np +import pytest + +from app.domain.entities.liveness_result import LivenessResult +from app.infrastructure.ml.liveness.active_liveness_detector import ( + ActiveLivenessDetector, +) + + +def _fake_face(num: int = 478) -> list: + return [ + SimpleNamespace(x=0.5 + (i % 7) * 0.01, y=0.5 + (i % 13) * 0.01, z=0.0) + for i in range(num) + ] + + +def _fake_result_with_face() -> SimpleNamespace: + return SimpleNamespace(face_landmarks=[_fake_face()]) + + +def _fake_empty_result() -> SimpleNamespace: + return SimpleNamespace(face_landmarks=[]) + + +class TestActiveLivenessDetectorTasksAPI: + @pytest.mark.asyncio + async def test_detect_returns_not_live_when_no_face(self) -> None: + detector = ActiveLivenessDetector() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect.return_value = _fake_empty_result() + + with mock.patch( + "app.infrastructure.ml.liveness.active_liveness_detector.create_face_landmarker", + return_value=fake_landmarker, + ), mock.patch( + "app.infrastructure.ml.liveness.active_liveness_detector.to_mp_image", + side_effect=lambda x: x, + ): + result = await detector.detect(np.zeros((100, 100, 3), dtype=np.uint8)) + + assert isinstance(result, LivenessResult) + assert result.is_live is False + assert result.liveness_score == 0.0 + assert result.details["eyes_open"] is False + assert result.details["smiling"] is False + + @pytest.mark.asyncio + async def test_detect_calls_landmarker_with_mp_image(self) -> None: + detector = ActiveLivenessDetector() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect.return_value = _fake_result_with_face() + + sentinel_mp_image = object() + with mock.patch( + "app.infrastructure.ml.liveness.active_liveness_detector.create_face_landmarker", + return_value=fake_landmarker, + ), mock.patch( + "app.infrastructure.ml.liveness.active_liveness_detector.to_mp_image", + return_value=sentinel_mp_image, + ): + await detector.detect(np.zeros((50, 50, 3), dtype=np.uint8)) + + fake_landmarker.detect.assert_called_once_with(sentinel_mp_image) + + @pytest.mark.asyncio + async def test_detect_propagates_runtime_error_when_landmarker_unavailable(self) -> None: + detector = ActiveLivenessDetector() + with mock.patch( + "app.infrastructure.ml.liveness.active_liveness_detector.create_face_landmarker", + return_value=None, + ): + with pytest.raises(RuntimeError): + await detector.detect(np.zeros((10, 10, 3), dtype=np.uint8)) diff --git a/tests/unit/infrastructure/test_face_landmarker_loader.py b/tests/unit/infrastructure/test_face_landmarker_loader.py new file mode 100644 index 0000000..afadf37 --- /dev/null +++ b/tests/unit/infrastructure/test_face_landmarker_loader.py @@ -0,0 +1,156 @@ +"""Tests for the shared MediaPipe FaceLandmarker (Tasks API) loader. + +Added 2026-05-12 alongside the mp.solutions → mp.tasks.vision port. The +loader is the single integration seam between the codebase and the new +MediaPipe API, so these tests focus on path/SHA resolution semantics rather +than on actual landmark detection (which requires a real model file). +""" + +from __future__ import annotations + +import hashlib +import os +import sys +from pathlib import Path +from unittest import mock + +import pytest + +# Import under test +from app.infrastructure.ml.landmarks import face_landmarker_loader + + +class TestResolveFaceLandmarkerModelPath: + """resolve_face_landmarker_model_path resolution order semantics.""" + + def test_env_override_wins(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + custom = tmp_path / "custom.task" + custom.write_bytes(b"stub") + monkeypatch.setenv("FACE_LANDMARKER_MODEL_PATH", str(custom)) + assert face_landmarker_loader.resolve_face_landmarker_model_path() == custom + + def test_env_pointing_to_missing_file_falls_through( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + # env var points at a non-existent file. The loader should NOT return + # that path. It should fall through to the next candidate(s); when + # none exist it returns None. + missing = tmp_path / "does_not_exist.task" + monkeypatch.setenv("FACE_LANDMARKER_MODEL_PATH", str(missing)) + # Stub the repo-relative default to also be missing. + with mock.patch.object( + face_landmarker_loader, "_DEFAULT_MODEL_PATH", tmp_path / "also_missing.task" + ): + # And cwd-relative path also missing — make cwd a tmp dir. + monkeypatch.chdir(tmp_path) + assert face_landmarker_loader.resolve_face_landmarker_model_path() is None + + def test_returns_none_when_nothing_found( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("FACE_LANDMARKER_MODEL_PATH", raising=False) + with mock.patch.object( + face_landmarker_loader, "_DEFAULT_MODEL_PATH", tmp_path / "nope.task" + ): + monkeypatch.chdir(tmp_path) + assert face_landmarker_loader.resolve_face_landmarker_model_path() is None + + +class TestVerifyModelSha256: + """Integrity check skip / match / mismatch behaviour.""" + + def test_skip_when_env_unset( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("FACE_LANDMARKER_MODEL_SHA256", raising=False) + f = tmp_path / "x.task" + f.write_bytes(b"any") + assert face_landmarker_loader.verify_model_sha256(f) is True + + def test_match(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + f = tmp_path / "x.task" + f.write_bytes(b"hello world") + digest = hashlib.sha256(b"hello world").hexdigest() + monkeypatch.setenv("FACE_LANDMARKER_MODEL_SHA256", digest) + assert face_landmarker_loader.verify_model_sha256(f) is True + + def test_match_case_insensitive( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + f = tmp_path / "x.task" + f.write_bytes(b"hello world") + digest = hashlib.sha256(b"hello world").hexdigest().upper() + monkeypatch.setenv("FACE_LANDMARKER_MODEL_SHA256", digest) + assert face_landmarker_loader.verify_model_sha256(f) is True + + def test_mismatch(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + f = tmp_path / "x.task" + f.write_bytes(b"hello world") + monkeypatch.setenv("FACE_LANDMARKER_MODEL_SHA256", "deadbeef" * 8) + assert face_landmarker_loader.verify_model_sha256(f) is False + + +class TestCreateFaceLandmarker: + """create_face_landmarker fail-soft behaviour when assets/imports are absent.""" + + def test_returns_none_when_model_missing( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("FACE_LANDMARKER_MODEL_PATH", raising=False) + with mock.patch.object( + face_landmarker_loader, "_DEFAULT_MODEL_PATH", tmp_path / "nope.task" + ): + monkeypatch.chdir(tmp_path) + assert face_landmarker_loader.create_face_landmarker() is None + + def test_returns_none_when_sha256_mismatch( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + f = tmp_path / "fake.task" + f.write_bytes(b"not-a-real-model") + monkeypatch.setenv("FACE_LANDMARKER_MODEL_PATH", str(f)) + monkeypatch.setenv("FACE_LANDMARKER_MODEL_SHA256", "deadbeef" * 8) + assert face_landmarker_loader.create_face_landmarker() is None + + def test_returns_none_when_mediapipe_import_fails( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + f = tmp_path / "fake.task" + f.write_bytes(b"stub") + monkeypatch.setenv("FACE_LANDMARKER_MODEL_PATH", str(f)) + monkeypatch.delenv("FACE_LANDMARKER_MODEL_SHA256", raising=False) + + # Simulate Tasks API import failure. The loader catches ImportError + # and returns None — verifying we never crash a request just because + # mediapipe is uninstallable on a particular host. + with mock.patch.dict( + sys.modules, + { + "mediapipe.tasks": None, + "mediapipe.tasks.python": None, + }, + ): + assert face_landmarker_loader.create_face_landmarker() is None + + +class TestToMpImage: + """to_mp_image helper wraps RGB arrays in mp.Image.""" + + def test_returns_mp_image_with_srgb_format(self) -> None: + pytest.importorskip("mediapipe") + import numpy as np + + # mp.Image instantiation pulls in libGLESv2.so.2 via the C bindings. + # CI images that strip GL/EGL libs cannot exercise the constructor; + # skip rather than fail-spuriously. The runtime container does have + # libgl1 installed (see Dockerfile) so this test passes there. + try: + wrapped = face_landmarker_loader.to_mp_image( + np.zeros((10, 10, 3), dtype=np.uint8) + ) + except OSError as exc: + pytest.skip(f"mediapipe native libs unavailable in this env: {exc}") + assert wrapped is not None + # Defensive: width/height are stable public attributes since 0.10.x. + assert wrapped.width == 10 + assert wrapped.height == 10 diff --git a/tests/unit/infrastructure/test_mediapipe_gaze_tracker_tasks_api.py b/tests/unit/infrastructure/test_mediapipe_gaze_tracker_tasks_api.py new file mode 100644 index 0000000..fd3ed47 --- /dev/null +++ b/tests/unit/infrastructure/test_mediapipe_gaze_tracker_tasks_api.py @@ -0,0 +1,100 @@ +"""Tests for MediaPipeGazeTracker after the mp.solutions → mp.tasks port. + +Added 2026-05-12. The gaze tracker uses VIDEO running-mode, so we verify +both that ``detect_for_video`` is invoked (NOT ``detect``) and that the +monotonic timestamp counter only ever increases. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest import mock + +import numpy as np +import pytest + +from app.infrastructure.ml.proctoring.mediapipe_gaze_tracker import ( + MediaPipeGazeTracker, +) + + +def _fake_face(num: int = 478): + # Spread x/y in a way that triggers the iris/eye-corner math without + # exercising degenerate divisions. + return [ + SimpleNamespace(x=0.4 + (i % 17) * 0.005, y=0.5 + (i % 19) * 0.005, z=0.0) + for i in range(num) + ] + + +def _fake_result_with_face() -> SimpleNamespace: + return SimpleNamespace(face_landmarks=[_fake_face()]) + + +def _fake_empty_result() -> SimpleNamespace: + return SimpleNamespace(face_landmarks=[]) + + +class TestMediaPipeGazeTrackerTasksAPI: + @pytest.mark.asyncio + async def test_analyze_uses_detect_for_video_not_detect(self) -> None: + tracker = MediaPipeGazeTracker() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect_for_video.return_value = _fake_result_with_face() + + with mock.patch( + "app.infrastructure.ml.proctoring.mediapipe_gaze_tracker.create_face_landmarker", + return_value=fake_landmarker, + ), mock.patch( + "app.infrastructure.ml.proctoring.mediapipe_gaze_tracker.to_mp_image", + side_effect=lambda x: x, + ): + image = np.zeros((480, 640, 3), dtype=np.uint8) + await tracker.analyze(image, session_id="s1") + + fake_landmarker.detect_for_video.assert_called_once() + fake_landmarker.detect.assert_not_called() + + @pytest.mark.asyncio + async def test_video_timestamps_strictly_monotonic(self) -> None: + tracker = MediaPipeGazeTracker() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect_for_video.return_value = _fake_result_with_face() + + with mock.patch( + "app.infrastructure.ml.proctoring.mediapipe_gaze_tracker.create_face_landmarker", + return_value=fake_landmarker, + ), mock.patch( + "app.infrastructure.ml.proctoring.mediapipe_gaze_tracker.to_mp_image", + side_effect=lambda x: x, + ): + image = np.zeros((100, 100, 3), dtype=np.uint8) + for _ in range(5): + await tracker.analyze(image, session_id="s1") + + timestamps = [c.args[1] for c in fake_landmarker.detect_for_video.call_args_list] + # The Tasks API rejects non-increasing timestamps with + # InvalidArgumentError, so this invariant is a hard correctness + # requirement. + assert all(timestamps[i] < timestamps[i + 1] for i in range(len(timestamps) - 1)) + + @pytest.mark.asyncio + async def test_no_face_returns_off_screen_result(self) -> None: + tracker = MediaPipeGazeTracker() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect_for_video.return_value = _fake_empty_result() + + with mock.patch( + "app.infrastructure.ml.proctoring.mediapipe_gaze_tracker.create_face_landmarker", + return_value=fake_landmarker, + ), mock.patch( + "app.infrastructure.ml.proctoring.mediapipe_gaze_tracker.to_mp_image", + side_effect=lambda x: x, + ): + image = np.zeros((100, 100, 3), dtype=np.uint8) + result = await tracker.analyze(image, session_id="s1") + + assert result.head_pose is None + assert result.gaze_direction is None + assert result.is_on_screen is False + assert result.confidence == 0.0 diff --git a/tests/unit/infrastructure/test_mediapipe_landmarks_tasks_api.py b/tests/unit/infrastructure/test_mediapipe_landmarks_tasks_api.py new file mode 100644 index 0000000..bc1d1d8 --- /dev/null +++ b/tests/unit/infrastructure/test_mediapipe_landmarks_tasks_api.py @@ -0,0 +1,126 @@ +"""Tests for MediaPipeLandmarkDetector after the mp.solutions → mp.tasks port. + +Added 2026-05-12. Real Tasks-API model loading requires the +`face_landmarker.task` asset on disk, which is not present in CI. These +tests therefore mock ``create_face_landmarker`` and assert the consumer +correctly adapts the new result shape (``result.face_landmarks[0][i].x``) +into the existing ``LandmarkResult`` domain entity. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest import mock + +import numpy as np +import pytest + +from app.domain.exceptions.feature_errors import LandmarkError +from app.infrastructure.ml.landmarks.mediapipe_landmarks import ( + MediaPipeLandmarkDetector, +) + + +def _make_fake_landmark(x: float, y: float, z: float = 0.0) -> SimpleNamespace: + """Stand-in for mediapipe.tasks.python.components.containers.NormalizedLandmark.""" + return SimpleNamespace(x=x, y=y, z=z) + + +def _make_fake_result(num_landmarks: int = 468) -> SimpleNamespace: + """Stand-in for mediapipe.tasks.python.vision.FaceLandmarkerResult. + + Tasks-API shape: ``result.face_landmarks`` is a list-of-lists where each + inner list is a flat sequence of NormalizedLandmark objects (no ``.landmark`` + accessor unlike the legacy Solutions API). + """ + face = [_make_fake_landmark(0.1 + i * 0.001, 0.2 + i * 0.001) for i in range(num_landmarks)] + return SimpleNamespace(face_landmarks=[face]) + + +class TestMediaPipeLandmarkDetectorTasksAPI: + def test_returns_landmarks_when_face_present(self) -> None: + detector = MediaPipeLandmarkDetector() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect.return_value = _make_fake_result(num_landmarks=468) + + with mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.create_face_landmarker", + return_value=fake_landmarker, + ), mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.to_mp_image", + side_effect=lambda x: x, # no-op + ): + image = np.zeros((480, 640, 3), dtype=np.uint8) + result = detector.detect(image) + + assert result.landmark_count == 468 + assert len(result.landmarks) == 468 + # First landmark uses x=0.1, y=0.2 (see _make_fake_landmark). The + # detector multiplies normalised coords by image dims and casts to int. + assert result.landmarks[0].x == int(0.1 * 640) + assert result.landmarks[0].y == int(0.2 * 480) + # 3D coordinate is dropped when include_3d=False (default). + assert result.landmarks[0].z is None + + def test_include_3d_propagates_z(self) -> None: + detector = MediaPipeLandmarkDetector() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect.return_value = _make_fake_result(num_landmarks=10) + + with mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.create_face_landmarker", + return_value=fake_landmarker, + ), mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.to_mp_image", + side_effect=lambda x: x, + ): + image = np.zeros((100, 100, 3), dtype=np.uint8) + result = detector.detect(image, include_3d=True) + + assert result.landmarks[5].z is not None + + def test_raises_landmark_error_when_no_face_detected(self) -> None: + detector = MediaPipeLandmarkDetector() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect.return_value = SimpleNamespace(face_landmarks=[]) + + with mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.create_face_landmarker", + return_value=fake_landmarker, + ), mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.to_mp_image", + side_effect=lambda x: x, + ): + with pytest.raises(LandmarkError): + detector.detect(np.zeros((10, 10, 3), dtype=np.uint8)) + + def test_raises_when_loader_returns_none(self) -> None: + """If model asset is missing the loader returns None — caller must surface a clear error.""" + detector = MediaPipeLandmarkDetector() + + with mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.create_face_landmarker", + return_value=None, + ): + with pytest.raises(LandmarkError): + detector.detect(np.zeros((10, 10, 3), dtype=np.uint8)) + + def test_landmarker_is_cached_across_calls(self) -> None: + detector = MediaPipeLandmarkDetector() + fake_landmarker = mock.MagicMock() + fake_landmarker.detect.return_value = _make_fake_result() + + with mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.create_face_landmarker", + return_value=fake_landmarker, + ) as create_mock, mock.patch( + "app.infrastructure.ml.landmarks.mediapipe_landmarks.to_mp_image", + side_effect=lambda x: x, + ): + image = np.zeros((100, 100, 3), dtype=np.uint8) + detector.detect(image) + detector.detect(image) + detector.detect(image) + + # Lazy init: create_face_landmarker should only run once. + assert create_mock.call_count == 1