From 5d1ecc68830f8b54a7d65a10b07780ff98bdd37e Mon Sep 17 00:00:00 2001 From: KoyoNoguchi Date: Sat, 3 Oct 2026 23:58:25 +0900 Subject: [PATCH] =?UTF-8?q?=E5=90=84=E3=83=A9=E3=82=A4=E3=83=96=E3=83=A9?= =?UTF-8?q?=E3=83=AA=E3=82=92=E6=9C=80=E6=96=B0=E7=89=88=E3=81=AB=E5=A4=89?= =?UTF-8?q?=E6=9B=B4=EF=BC=8Cmediapipe=E3=81=AE=E4=BB=95=E6=A7=98=E5=A4=89?= =?UTF-8?q?=E6=9B=B4=E3=81=AB=E5=AF=BE=E5=BF=9C=EF=BC=8E=20=E3=82=AA?= =?UTF-8?q?=E3=82=A4=E3=83=A9=E3=83=BC=E8=A7=92=E8=A8=88=E6=B8=AC=E3=82=92?= =?UTF-8?q?mediapipe=E3=81=AEFaceLandmarker=E3=81=A7=E8=A1=8C=E3=81=86?= =?UTF-8?q?=E3=82=88=E3=81=86=E3=81=AB=E5=A4=89=E6=9B=B4=E3=81=97=EF=BC=8C?= =?UTF-8?q?yaw=E5=80=A4=E3=83=90=E3=82=A4=E3=82=A2=E3=82=B9=E5=95=8F?= =?UTF-8?q?=E9=A1=8C=E3=82=92=E8=A7=A3=E6=B1=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/default.yaml | 3 ++- src/config_loader.py | 5 +++-- src/face_processor.py | 48 +++++++++++++++++++++++++++++-------------- src/head_pose.py | 19 +++++++++++++++++ 4 files changed, 57 insertions(+), 18 deletions(-) diff --git a/config/default.yaml b/config/default.yaml index 5510d43..8f81f2a 100644 --- a/config/default.yaml +++ b/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 diff --git a/src/config_loader.py b/src/config_loader.py index 9b5e4a4..b3fbf1b 100644 --- a/src/config_loader.py +++ b/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, diff --git a/src/face_processor.py b/src/face_processor.py index d599d2a..e5ba864 100644 --- a/src/face_processor.py +++ b/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,22 +82,30 @@ 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) - estimation = self.estimator.estimate(landmarks, width, height) + 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: return HeadPoseResult( @@ -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) ] diff --git a/src/head_pose.py b/src/head_pose.py index 3aeefc3..b5d103a 100644 --- a/src/head_pose.py +++ b/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) # 再投影誤差は無いので棄却判定はスキップされる