各ライブラリを最新版に変更，mediapipeの仕様変更に対応．

オイラー角計測をmediapipeのFaceLandmarkerで行うように変更し，yaw値バイアス問題を解決

config/default.yaml

@@ -85,9 +85,10 @@ face:
   refine_landmarks: true
   min_detection_confidence: 0.40
   min_tracking_confidence: 0.40
+  model_path: models/face_landmarker.task
 head_pose:
   enabled: true
-  method: solvepnp
+  method: facial_matrix
   smoothing_enabled: true
   # none | ema | median_ema | hampel_ema | one_euro | median_one_euro (see src/yaw_filters.py)
   smoothing_method: ema

src/config_loader.py

@@ -194,6 +194,7 @@ class FaceSettings:
     refine_landmarks: bool
     min_detection_confidence: float
     min_tracking_confidence: float
+    model_path: str
 
 
 @dataclass(frozen=True, slots=True)
@@ -609,9 +610,9 @@ def _validate(config: AppConfig) -> None:
     if not 0.0 <= face.min_detection_confidence <= 1.0 or not 0.0 <= face.min_tracking_confidence <= 1.0:
         raise ConfigurationError("face confidence thresholds must be in [0, 1]")
     pose = config.head_pose
-    if pose.method != "solvepnp" or pose.smoothing_method not in FILTER_KINDS:
+    if pose.method not in {"solvepnp", "facial_matrix"} or pose.smoothing_method not in FILTER_KINDS:
         raise ConfigurationError(
-            f"head_pose method must be solvepnp and smoothing_method one of {list(FILTER_KINDS)}")
+            f"head_pose method must be solvepnp or facial_matrix and smoothing_method one of {list(FILTER_KINDS)}")
     if not 0.0 < pose.ema_alpha <= 1.0 or pose.max_abs_angle <= 0:
         raise ConfigurationError("head_pose EMA alpha and maximum angle are invalid")
     numeric = (pose.smoothing_window, pose.hampel_threshold, pose.one_euro_min_cutoff, pose.one_euro_beta,

src/face_processor.py

@@ -6,6 +6,7 @@ import logging
 import math
 from collections import deque
 from typing import Any
+from pathlib import Path
 
 from .config_loader import FaceSettings, HeadPoseSettings
 from .exceptions import FaceProcessingError
@@ -32,16 +33,25 @@ class FaceProcessor:
     def _initialize(self) -> None:
         try:
             import mediapipe as mp
+            from mediapipe.tasks import python as mp_python
+            from mediapipe.tasks.python import vision
 
-            self._mesh = mp.solutions.face_mesh.FaceMesh(
-                static_image_mode=self.config.static_image_mode,
-                max_num_faces=self.config.max_num_faces,
-                refine_landmarks=self.config.refine_landmarks,
-                min_detection_confidence=self.config.min_detection_confidence,
+            model_path = Path(self.config.model_path)
+            if not model_path.is_file():
+                raise FileNotFoundError(f"face_landmarker model not found: {model_path}")
+            options = vision.FaceLandmarkerOptions(
+                base_options=mp_python.BaseOptions(model_asset_path=str(model_path)),
+                running_mode=vision.RunningMode.IMAGE,  # ROIごとに独立して推論するためIMAGEモード
+                num_faces=self.config.max_num_faces,
+                min_face_detection_confidence=self.config.min_detection_confidence,
+                min_face_presence_confidence=self.config.min_detection_confidence,
                 min_tracking_confidence=self.config.min_tracking_confidence,
+                output_facial_transformation_matrixes=True,
             )
+            self._mp = mp
+            self._mesh = vision.FaceLandmarker.create_from_options(options)
         except Exception as exc:
-            raise FaceProcessingError(f"Failed to initialize MediaPipe Face Mesh: {exc}") from exc
+            raise FaceProcessingError(f"Failed to initialize MediaPipe FaceLandmarker: {exc}") from exc
 
     def process(
         self, frame: Any, tracked_persons: list[TrackedPerson], frame_info: FrameInfo,
@@ -72,21 +82,29 @@ class FaceProcessor:
         try:
             image = frame[roi.y1:roi.y2, roi.x1:roi.x2]
             import cv2
+            import numpy as np
 
             target = self.config.roi_upscale_target
             longest = max(image.shape[:2])
             if target > 0 and longest < target:
                 scale = target / longest
                 image = cv2.resize(image, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC)
-            output = self._mesh.process(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
-            faces = getattr(output, "multi_face_landmarks", None) or []
+            rgb = np.ascontiguousarray(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
+            output = self._mesh.detect(self._mp.Image(image_format=self._mp.ImageFormat.SRGB, data=rgb))
+            faces = output.face_landmarks or []
             if not faces:
                 return self._failure(person.track_id, info, "face_not_detected", roi)
-            selected = min(faces, key=self._face_center_distance)
-            landmarks = self._convert_landmarks(selected.landmark, roi)
+            index = min(range(len(faces)), key=lambda k: self._face_center_distance(faces[k]))
+            landmarks = self._convert_landmarks(faces[index], roi)
         except Exception:
             LOGGER.exception("MediaPipe failed for track %d on frame %d", person.track_id, info.frame_number)
             return self._failure(person.track_id, info, "mediapipe_error", roi)
+        if self.head_pose_config.method == "facial_matrix":
+            matrices = output.facial_transformation_matrixes or []
+            if index >= len(matrices):
+                return self._failure(person.track_id, info, "facial_matrix_missing", roi)
+            estimation = self.estimator.estimate_from_matrix(matrices[index])
+        else:
             estimation = self.estimator.estimate(landmarks, width, height)
         interocular = self._interocular_px(landmarks)
         if not estimation.success:
@@ -118,9 +136,9 @@ class FaceProcessor:
 
     @staticmethod
     def _face_center_distance(face: Any) -> float:
-        landmarks = face.landmark
-        center_x = sum(point.x for point in landmarks) / len(landmarks)
-        center_y = sum(point.y for point in landmarks) / len(landmarks)
+        points = getattr(face, "landmark", face)
+        center_x = sum(point.x for point in points) / len(points)
+        center_y = sum(point.y for point in points) / len(points)
         return (center_x - 0.5) ** 2 + (center_y - 0.5) ** 2
 
     @staticmethod
@@ -129,7 +147,7 @@ class FaceProcessor:
             FaceLandmark(
                 index=index, x=roi.x1 + float(point.x) * roi.width,
                 y=roi.y1 + float(point.y) * roi.height, z=float(point.z) * roi.width,
-                visibility=float(point.visibility) if hasattr(point, "visibility") else None,
+                visibility=float(point.visibility) if getattr(point, "visibility", None) is not None else None,
             )
             for index, point in enumerate(raw_landmarks)
         ]

src/head_pose.py

@@ -111,3 +111,22 @@ class HeadPoseEstimator:
         if math.isfinite(roll):
             roll = (roll + 90.0) % 180.0 - 90.0
         return float(pitch), float(yaw), float(roll)
+
+    @staticmethod
+    def facial_matrix_to_euler(matrix) -> tuple[float, float, float]:
+        import cv2
+        import numpy as np
+        m = np.asarray(matrix, dtype=np.float64).reshape(4, 4)
+        rotation = m[:3, :3]
+        rotation = rotation / np.linalg.norm(rotation, axis=0)  # スケール成分を除去
+        angles = cv2.RQDecomp3x3(rotation)[0]
+        return float(angles[0]), float(angles[1]), float(angles[2])
+
+    def estimate_from_matrix(self, matrix) -> HeadPoseEstimation:
+        if not self.config.enabled:
+            return HeadPoseEstimation(False, failure_reason="head_pose_disabled")
+        try:
+            angles = self.facial_matrix_to_euler(matrix)
+        except Exception:
+            return HeadPoseEstimation(False, failure_reason="facial_matrix_failed")
+        return self.result_from_angles(angles)  # 再投影誤差は無いので棄却判定はスキップされる