Files
planet/motion_agent/recognizer.py
linkong 899e3bce43
Some checks failed
ci / backend (push) Has been cancelled
ci / frontend (push) Has been cancelled
release / images (push) Has been cancelled
ci / delivery (push) Has been cancelled
release: bump version to 0.71.0
2026-06-11 16:47:24 +08:00

502 lines
18 KiB
Python

"""Gesture recognizer interfaces.
The production recognizer is intentionally thin in this skeleton. It validates
that optional CV dependencies are available and keeps the protocol independent
from any specific model implementation.
"""
from __future__ import annotations
import os
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Protocol
from .cameras import MotionAgentDependencyError
from .events import GestureName, SkeletonEvent, SkeletonJoint, now_ms
POSE_MODEL_URL = (
"https://storage.googleapis.com/mediapipe-models/pose_landmarker/"
"pose_landmarker_lite/float16/latest/pose_landmarker_lite.task"
)
POSE_MODEL_CACHE_PATH = Path.home() / ".cache" / "planet" / "motion_agent" / "pose_landmarker_lite.task"
POSE_JOINTS = [
(0, "nose"),
(7, "left_ear"),
(8, "right_ear"),
(11, "left_shoulder"),
(12, "right_shoulder"),
(13, "left_elbow"),
(14, "right_elbow"),
(15, "left_wrist"),
(16, "right_wrist"),
]
POSE_BONES = [
("left_shoulder", "left_elbow"),
("left_elbow", "left_wrist"),
("right_shoulder", "right_elbow"),
("right_elbow", "right_wrist"),
("left_shoulder", "right_shoulder"),
]
LEFT_WRIST_LAYER_DELTA_Y = 0.05
HEAD_TILT_DELTA_Y = 0.035
ARM_PATTERN_TERMINAL_TOLERANCE_DEG = 32
ARM_PATTERN_UPPER_TOLERANCE_DEG = 34
ARM_PATTERN_MIN_SEGMENT = 0.045
ARM_PATTERN_MIN_SIDE_REACH = 0.06
ARM_PATTERN_MIN_VERTICAL_REACH = 0.055
MIN_GESTURE_INTENSITY = 0.45
ARM_PATTERN_INTENSITY_SCALE = 5
WRIST_LAYER_INTENSITY_SCALE = 9
HEAD_TILT_INTENSITY_SCALE = 12
ZOOM_CLOSE_WRIST_SPREAD_FACTOR = 1.28
ZOOM_SUPPRESS_WRIST_SPREAD_FACTOR = 1.18
ZOOM_TREND_MIN_WRIST_DELTA = 0.010
ZOOM_TREND_HEIGHT_TOLERANCE = 0.18
ZOOM_TREND_INTENSITY_SCALE = 14
ZOOM_TREND_MIN_INTENSITY = 0.6
LEFT_ARM_REST_HANGING_BELOW_SHOULDER = 0.13
ZOOM_HOLD_HEIGHT_TOLERANCE = 0.18
ZOOM_HOLD_WRIST_BELOW_SHOULDER_LIMIT = 0.10
ZOOM_HOLD_SPREAD_FACTOR = 1.30
ZOOM_HOLD_CLOSE_FACTOR = 0.85
ZOOM_HOLD_ELBOW_OUT_FACTOR = 0.25
@dataclass(frozen=True)
class GestureObservation:
gesture: GestureName
confidence: float
intensity: float = 1.0
timestamp_ms: int | None = None
camera_id: str = "unknown"
class GestureRecognizer(Protocol):
name: str
def recognize(self, frame: Any) -> GestureObservation | None:
"""Return a gesture observation for the current frame."""
def debug_skeleton(
self,
frame: Any,
*,
camera_id: str,
mode: str,
matched_gesture: GestureName | None = None,
confidence: float = 0.0,
) -> SkeletonEvent | None:
"""Return normalized skeleton debug data when available."""
class MediaPipeGestureRecognizer:
name = "mediapipe-opencv"
def __init__(self) -> None:
try:
import cv2 # noqa: F401
import mediapipe as mp
from mediapipe.tasks.python.core import base_options as base_options_module
from mediapipe.tasks.python.vision import pose_landmarker
from mediapipe.tasks.python.vision.core import vision_task_running_mode
except ImportError as exc:
raise MotionAgentDependencyError(
"MediaPipe and OpenCV are required for live gesture recognition. "
"Add mediapipe and opencv-python with uv, or use --dry-run for protocol testing."
) from exc
model_path = _resolve_pose_model_path()
options = pose_landmarker.PoseLandmarkerOptions(
base_options=base_options_module.BaseOptions(model_asset_path=str(model_path)),
running_mode=vision_task_running_mode.VisionTaskRunningMode.VIDEO,
num_poses=1,
)
self._mp = mp
self._cv2 = cv2
self._landmarker = pose_landmarker.PoseLandmarker.create_from_options(options)
self._previous_joints: list[SkeletonJoint] = []
self._latest_joints: list[SkeletonJoint] = []
self._latest_timestamp_ms = 0
self._state: dict[str, str | None] = {"active_pattern_gesture": None}
def recognize(self, frame: Any) -> GestureObservation | None:
joints = self._detect_joints(frame)
self._latest_joints = joints
self._latest_timestamp_ms = now_ms()
if not joints:
self._previous_joints = []
self._state["active_pattern_gesture"] = None
return None
observation = _recognize_gesture(joints, self._previous_joints, self._state)
self._previous_joints = joints
if observation is None:
return None
return GestureObservation(
gesture=observation["gesture"],
confidence=observation["confidence"],
intensity=observation["intensity"],
timestamp_ms=self._latest_timestamp_ms,
)
def debug_skeleton(
self,
frame: Any,
*,
camera_id: str,
mode: str,
matched_gesture: GestureName | None = None,
confidence: float = 0.0,
) -> SkeletonEvent | None:
_ = frame
if not self._latest_joints:
return None
return SkeletonEvent(
joints=self._latest_joints,
bones=POSE_BONES,
matched_gesture=matched_gesture,
confidence=confidence,
camera_id=camera_id,
mode=mode,
)
def _detect_joints(self, frame: Any) -> list[SkeletonJoint]:
rgb = self._cv2.cvtColor(frame, self._cv2.COLOR_BGR2RGB)
image = self._mp.Image(image_format=self._mp.ImageFormat.SRGB, data=rgb)
result = self._landmarker.detect_for_video(image, now_ms())
if not result.pose_landmarks:
return []
landmarks = result.pose_landmarks[0]
joints: list[SkeletonJoint] = []
for index, joint_id in POSE_JOINTS:
if index >= len(landmarks):
continue
point = landmarks[index]
x = _clamp01(float(point.x))
y = _clamp01(float(point.y))
confidence = _clamp01(float(getattr(point, "visibility", getattr(point, "presence", 1.0))))
joints.append(SkeletonJoint(joint_id, x, y, confidence))
return joints
class NullGestureRecognizer:
name = "dry-run"
def recognize(self, frame: Any) -> GestureObservation | None:
_ = frame
return None
def debug_skeleton(
self,
frame: Any,
*,
camera_id: str,
mode: str,
matched_gesture: GestureName | None = None,
confidence: float = 0.0,
) -> SkeletonEvent | None:
_ = frame
now = now_ms()
sway = ((now // 250) % 6 - 2.5) * 0.015
joints = [
SkeletonJoint("head", 0.5, 0.18, 1.0),
SkeletonJoint("neck", 0.5, 0.3, 1.0),
SkeletonJoint("left_shoulder", 0.38, 0.34, 1.0),
SkeletonJoint("right_shoulder", 0.62, 0.34, 1.0),
SkeletonJoint("left_elbow", 0.31 + sway, 0.48, 0.95),
SkeletonJoint("right_elbow", 0.69 - sway, 0.48, 0.95),
SkeletonJoint("left_wrist", 0.24 + sway, 0.62, 0.9),
SkeletonJoint("right_wrist", 0.76 - sway, 0.62, 0.9),
]
bones = [
("head", "neck"),
("neck", "left_shoulder"),
("neck", "right_shoulder"),
("left_shoulder", "left_elbow"),
("left_elbow", "left_wrist"),
("right_shoulder", "right_elbow"),
("right_elbow", "right_wrist"),
]
return SkeletonEvent(
joints=joints,
bones=bones,
matched_gesture=matched_gesture,
confidence=confidence,
camera_id=camera_id,
mode=mode,
)
def _resolve_pose_model_path() -> Path:
configured = os.getenv("MOTION_AGENT_POSE_MODEL_PATH")
path = Path(configured).expanduser() if configured else POSE_MODEL_CACHE_PATH
if path.exists():
return path
path.parent.mkdir(parents=True, exist_ok=True)
try:
urllib.request.urlretrieve(POSE_MODEL_URL, path)
except Exception as exc:
raise MotionAgentDependencyError(
"MediaPipe pose model is missing and could not be downloaded. "
f"Set MOTION_AGENT_POSE_MODEL_PATH to a local .task file or download {POSE_MODEL_URL}."
) from exc
return path
def _clamp01(value: float) -> float:
return max(0.0, min(1.0, value))
def _get_joint(joints: list[SkeletonJoint], joint_id: str) -> SkeletonJoint | None:
return next((joint for joint in joints if joint.id == joint_id), None)
def _vector_between(start: SkeletonJoint | None, end: SkeletonJoint | None) -> dict[str, float] | None:
if start is None or end is None:
return None
dx = end.x - start.x
dy = end.y - start.y
return {"dx": dx, "dy": dy, "length": (dx * dx + dy * dy) ** 0.5}
def _vector_angle_deg(vector: dict[str, float]) -> float:
import math
return math.atan2(vector["dy"], vector["dx"]) * 180 / math.pi
def _normalize_angle_delta(angle: float, target: float) -> float:
delta = angle - target
while delta > 180:
delta -= 360
while delta < -180:
delta += 360
return abs(delta)
def _is_angle_near(angle: float, target: float, tolerance_deg: float) -> bool:
return _normalize_angle_delta(angle, target) <= tolerance_deg
def _is_horizontal_arm(upper_vector: dict[str, float] | None) -> bool:
if not upper_vector or upper_vector["length"] < ARM_PATTERN_MIN_SEGMENT:
return False
angle = _vector_angle_deg(upper_vector)
return _is_angle_near(angle, 0, ARM_PATTERN_UPPER_TOLERANCE_DEG) or _is_angle_near(
angle, 180, ARM_PATTERN_UPPER_TOLERANCE_DEG
)
def _is_terminal_toward(vector: dict[str, float] | None, target_angle: float) -> bool:
if not vector or vector["length"] < ARM_PATTERN_MIN_SEGMENT:
return False
return _is_angle_near(_vector_angle_deg(vector), target_angle, ARM_PATTERN_TERMINAL_TOLERANCE_DEG)
def _gesture(gesture: GestureName, confidence: float, intensity: float) -> dict[str, Any]:
return {"gesture": gesture, "confidence": confidence, "intensity": intensity}
def _get_right_arm_pattern(
right_shoulder: SkeletonJoint,
right_elbow: SkeletonJoint,
right_wrist: SkeletonJoint,
) -> dict[str, Any] | None:
upper = _vector_between(right_shoulder, right_elbow)
terminal = _vector_between(right_elbow, right_wrist)
if not upper or not terminal:
return None
intensity = min(1.0, max(MIN_GESTURE_INTENSITY, terminal["length"] * ARM_PATTERN_INTENSITY_SCALE))
if _is_terminal_toward(terminal, 180) and right_wrist.x < right_shoulder.x - ARM_PATTERN_MIN_SIDE_REACH:
return _gesture("rotate_right", 0.82, intensity)
if _is_terminal_toward(terminal, 0) and right_wrist.x > right_shoulder.x + ARM_PATTERN_MIN_SIDE_REACH:
return _gesture("rotate_left", 0.82, intensity)
if (
_is_horizontal_arm(upper)
and _is_terminal_toward(terminal, -90)
and right_wrist.y < right_elbow.y - ARM_PATTERN_MIN_VERTICAL_REACH
):
return _gesture("rotate_up", 0.8, intensity)
if (
_is_horizontal_arm(upper)
and _is_terminal_toward(terminal, 90)
and right_wrist.y > right_elbow.y + ARM_PATTERN_MIN_VERTICAL_REACH
):
return _gesture("rotate_down", 0.8, intensity)
return None
def _is_zoom_candidate_pose(
left_shoulder: SkeletonJoint,
left_elbow: SkeletonJoint,
left_wrist: SkeletonJoint,
right_shoulder: SkeletonJoint,
right_elbow: SkeletonJoint,
right_wrist: SkeletonJoint,
shoulder_width: float,
) -> bool:
wrists_apart = abs(right_wrist.x - left_wrist.x)
both_hands_outside = (
left_wrist.x < left_elbow.x - ARM_PATTERN_MIN_SIDE_REACH * 0.25
and left_wrist.x < left_shoulder.x - ARM_PATTERN_MIN_SIDE_REACH * 0.55
and right_wrist.x > right_elbow.x + ARM_PATTERN_MIN_SIDE_REACH * 0.25
and right_wrist.x > right_shoulder.x + ARM_PATTERN_MIN_SIDE_REACH * 0.55
)
both_elbows_participating = (
left_elbow.x <= left_shoulder.x + ARM_PATTERN_MIN_SIDE_REACH
and right_elbow.x >= right_shoulder.x - ARM_PATTERN_MIN_SIDE_REACH
)
hands_near_center = (
left_elbow.x < left_shoulder.x - ARM_PATTERN_MIN_SIDE_REACH * 0.5
and right_elbow.x > right_shoulder.x + ARM_PATTERN_MIN_SIDE_REACH * 0.5
and left_wrist.x > left_elbow.x
and right_wrist.x < right_elbow.x
and wrists_apart < shoulder_width * ZOOM_CLOSE_WRIST_SPREAD_FACTOR
)
return (
both_hands_outside
and both_elbows_participating
and wrists_apart > shoulder_width * ZOOM_SUPPRESS_WRIST_SPREAD_FACTOR
) or hands_near_center
def _is_left_arm_at_rest(left_shoulder: SkeletonJoint, left_elbow: SkeletonJoint, left_wrist: SkeletonJoint) -> bool:
return (
left_wrist.y >= left_shoulder.y + LEFT_ARM_REST_HANGING_BELOW_SHOULDER
and left_wrist.x >= left_shoulder.x - ARM_PATTERN_MIN_SIDE_REACH
and left_elbow.x >= left_shoulder.x - ARM_PATTERN_MIN_SIDE_REACH
)
def _get_zoom_trend(
left_wrist: SkeletonJoint,
right_wrist: SkeletonJoint,
previous_left_wrist: SkeletonJoint | None,
previous_right_wrist: SkeletonJoint | None,
) -> dict[str, Any] | None:
if previous_left_wrist is None or previous_right_wrist is None:
return None
if abs(right_wrist.y - left_wrist.y) > ZOOM_TREND_HEIGHT_TOLERANCE:
return None
left_moved = abs(left_wrist.x - previous_left_wrist.x)
right_moved = abs(right_wrist.x - previous_right_wrist.x)
if left_moved < ZOOM_TREND_MIN_WRIST_DELTA or right_moved < ZOOM_TREND_MIN_WRIST_DELTA:
return None
spread_delta = abs(right_wrist.x - left_wrist.x) - abs(previous_right_wrist.x - previous_left_wrist.x)
min_spread_delta = ZOOM_TREND_MIN_WRIST_DELTA * 2
intensity = min(1.0, max(ZOOM_TREND_MIN_INTENSITY, (left_moved + right_moved) * ZOOM_TREND_INTENSITY_SCALE))
if spread_delta > min_spread_delta:
return _gesture("zoom_in", 0.88, intensity)
if spread_delta < -min_spread_delta:
return _gesture("zoom_out", 0.86, intensity)
return None
def _get_zoom_hold_pose(
left_shoulder: SkeletonJoint,
left_elbow: SkeletonJoint,
left_wrist: SkeletonJoint,
right_shoulder: SkeletonJoint,
right_elbow: SkeletonJoint,
right_wrist: SkeletonJoint,
shoulder_width: float,
) -> dict[str, Any] | None:
if abs(right_wrist.y - left_wrist.y) > ZOOM_HOLD_HEIGHT_TOLERANCE:
return None
avg_shoulder_y = (left_shoulder.y + right_shoulder.y) / 2
wrists_raised = (
left_wrist.y <= avg_shoulder_y + ZOOM_HOLD_WRIST_BELOW_SHOULDER_LIMIT
and right_wrist.y <= avg_shoulder_y + ZOOM_HOLD_WRIST_BELOW_SHOULDER_LIMIT
)
if not wrists_raised:
return None
span = abs(right_wrist.x - left_wrist.x)
if span > shoulder_width * ZOOM_HOLD_SPREAD_FACTOR:
return _gesture("zoom_in", 0.82, 0.8)
elbows_outward = (
abs(left_elbow.x - left_shoulder.x) > shoulder_width * ZOOM_HOLD_ELBOW_OUT_FACTOR
and abs(right_elbow.x - right_shoulder.x) > shoulder_width * ZOOM_HOLD_ELBOW_OUT_FACTOR
)
if elbows_outward and span < shoulder_width * ZOOM_HOLD_CLOSE_FACTOR:
return _gesture("zoom_out", 0.80, 0.7)
return None
def _apply_pose_latch(observation: dict[str, Any] | None, state: dict[str, str | None]) -> dict[str, Any] | None:
if observation is None:
return None
if state.get("active_pattern_gesture") == observation["gesture"]:
return None
state["active_pattern_gesture"] = observation["gesture"]
return observation
def _recognize_gesture(
joints: list[SkeletonJoint],
previous_joints: list[SkeletonJoint],
state: dict[str, str | None],
) -> dict[str, Any] | None:
left_ear = _get_joint(joints, "left_ear")
right_ear = _get_joint(joints, "right_ear")
left_wrist = _get_joint(joints, "left_wrist")
right_wrist = _get_joint(joints, "right_wrist")
left_elbow = _get_joint(joints, "left_elbow")
right_elbow = _get_joint(joints, "right_elbow")
left_shoulder = _get_joint(joints, "left_shoulder")
right_shoulder = _get_joint(joints, "right_shoulder")
previous_left_wrist = _get_joint(previous_joints, "left_wrist")
previous_right_wrist = _get_joint(previous_joints, "right_wrist")
if not all([left_wrist, right_wrist, left_elbow, right_elbow, left_shoulder, right_shoulder]):
return None
assert left_wrist and right_wrist and left_elbow and right_elbow and left_shoulder and right_shoulder
trend = _get_zoom_trend(left_wrist, right_wrist, previous_left_wrist, previous_right_wrist)
if trend:
state["active_pattern_gesture"] = trend["gesture"]
return trend
shoulder_width = max(0.08, abs(right_shoulder.x - left_shoulder.x))
left_raised = left_wrist.y < left_shoulder.y - 0.05
right_raised = right_wrist.y < right_shoulder.y - 0.05
left_delta_y = left_wrist.y - previous_left_wrist.y if previous_left_wrist else 0.0
head_tilt_y = right_ear.y - left_ear.y if left_ear and right_ear else 0.0
if not right_raised and left_raised and left_delta_y < -LEFT_WRIST_LAYER_DELTA_Y:
return _gesture("layer_prev", 0.78, min(1.0, abs(left_delta_y) * WRIST_LAYER_INTENSITY_SCALE))
if not right_raised and left_raised and left_delta_y > LEFT_WRIST_LAYER_DELTA_Y:
return _gesture("layer_next", 0.78, min(1.0, abs(left_delta_y) * WRIST_LAYER_INTENSITY_SCALE))
if head_tilt_y < -HEAD_TILT_DELTA_Y:
return _gesture("focus_prev", 0.78, min(1.0, abs(head_tilt_y) * HEAD_TILT_INTENSITY_SCALE))
if head_tilt_y > HEAD_TILT_DELTA_Y:
return _gesture("focus_next", 0.78, min(1.0, abs(head_tilt_y) * HEAD_TILT_INTENSITY_SCALE))
zoom_pattern = _get_zoom_hold_pose(
left_shoulder,
left_elbow,
left_wrist,
right_shoulder,
right_elbow,
right_wrist,
shoulder_width,
)
if zoom_pattern:
state["active_pattern_gesture"] = zoom_pattern["gesture"]
return zoom_pattern
rotate_allowed = not _is_zoom_candidate_pose(
left_shoulder,
left_elbow,
left_wrist,
right_shoulder,
right_elbow,
right_wrist,
shoulder_width,
) and _is_left_arm_at_rest(left_shoulder, left_elbow, left_wrist)
rotate_pattern = _get_right_arm_pattern(right_shoulder, right_elbow, right_wrist) if rotate_allowed else None
if rotate_pattern:
return _apply_pose_latch(rotate_pattern, state)
state["active_pattern_gesture"] = None
return None