File size: 8,802 Bytes
e2d95ed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 | """
From raw RGB frames to model input, using exactly the preprocessing chain that
training used (``backend.preprocessing.landmarks`` + ``backend.preprocessing.sequence``).
Used by the webcam loop, the batch video runner and the API's video endpoint, so
the live path and the training path cannot diverge.
"""
from __future__ import annotations
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Dict, List, Optional, Sequence
import numpy as np
import config
from backend.preprocessing import landmarks as lm
from backend.preprocessing import sequence as seq_ops
from backend.preprocessing.mediapipe_extractor import MediaPipeExtractor
@dataclass
class BufferStats:
frames_seen: int = 0
frames_with_pose: int = 0
frames_with_left: int = 0
frames_with_right: int = 0
frames_with_nothing: int = 0
started_at: float = field(default_factory=time.time)
@property
def elapsed(self) -> float:
return max(1e-6, time.time() - self.started_at)
@property
def fps(self) -> float:
return self.frames_seen / self.elapsed
def as_dict(self) -> Dict[str, float]:
return {
"frames_seen": self.frames_seen,
"pose_rate": self.frames_with_pose / max(1, self.frames_seen),
"left_hand_rate": self.frames_with_left / max(1, self.frames_seen),
"right_hand_rate": self.frames_with_right / max(1, self.frames_seen),
"empty_rate": self.frames_with_nothing / max(1, self.frames_seen),
"fps": self.fps,
}
class FrameBuffer:
"""Rolling window of landmark frames -> fixed-length feature sequence.
Landmarks are normalised *per frame at insertion time*, so the tensor handed
to the model is built the same way whether the frames came from a webcam, an
uploaded video or the dataset.
"""
def __init__(self, length: Optional[int] = None, extractor: Optional[MediaPipeExtractor] = None,
stride: int = 1):
self.length = int(length or config.SEQUENCE_LENGTH)
self.stride = max(1, int(stride))
self.extractor = extractor or MediaPipeExtractor()
self._frames: Deque[np.ndarray] = deque(maxlen=self.length)
self.last_landmarks = None
self.last_stage_timings = {"mediapipe_ms": 0.0, "preprocessing_ms": 0.0, "buffer_ms": 0.0}
self.stats = BufferStats()
self._n_since_push = 0
# ------------------------------------------------------------------
def push(self, rgb: np.ndarray, timestamp_ms: Optional[int] = None) -> Optional[np.ndarray]:
"""Add one frame; returns ``(length, 225)`` features once the window is full."""
push_started = time.perf_counter()
self.stats.frames_seen += 1
stage_started = time.perf_counter()
landmarks = self.extractor.extract_frame(rgb, timestamp_ms=timestamp_ms)
mediapipe_ms = (time.perf_counter() - stage_started) * 1000.0
self.last_landmarks = landmarks
self.stats.frames_with_pose += int(landmarks.pose is not None)
self.stats.frames_with_left += int(landmarks.left_hand is not None)
self.stats.frames_with_right += int(landmarks.right_hand is not None)
self.stats.frames_with_nothing += int(landmarks.is_empty())
stage_started = time.perf_counter()
coords, mask = lm.pack_frame(landmarks)
feats, _ = lm.normalise_sequence(coords[None, ...], mask[None, ...])
preprocessing_ms = (time.perf_counter() - stage_started) * 1000.0
stage_started = time.perf_counter()
self._frames.append(feats[0])
self._n_since_push += 1
buffering_ms = (time.perf_counter() - stage_started) * 1000.0
self.last_stage_timings = {
"mediapipe_ms": round(mediapipe_ms, 3),
"preprocessing_ms": round(preprocessing_ms, 3),
"buffer_ms": round(buffering_ms, 3),
"frame_total_ms": round((time.perf_counter() - push_started) * 1000.0, 3),
}
if len(self._frames) < self.length or (self._n_since_push % self.stride):
return None
return self.features()
# ------------------------------------------------------------------
def features(self) -> Optional[np.ndarray]:
if len(self._frames) < self.length:
return None
return np.stack(self._frames).astype(np.float32)
def ready(self) -> bool:
return len(self._frames) >= self.length
def fill_ratio(self) -> float:
return len(self._frames) / self.length
def reset(self) -> None:
self._frames.clear()
self._n_since_push = 0
self.last_landmarks = None
def close(self) -> None:
self.extractor.close()
# --------------------------------------------------------------------------
# Offline helpers (batch inference without a camera)
# --------------------------------------------------------------------------
def features_from_rgb_frames(frames: Sequence[np.ndarray], extractor: MediaPipeExtractor,
length: Optional[int] = None, return_landmarks: bool = False):
"""Extract + normalise + resample a whole clip of RGB frames.
When ``return_landmarks=True``, also returns the raw (pre-normalisation)
per-frame landmark coordinates, resampled with the exact same method to
the same ``length`` timesteps as the features -- i.e. these are literally
the landmarks behind the sequence the model consumes, not a separate
estimate, so a caller can honestly visualise "what the model saw".
"""
length = int(length or config.SEQUENCE_LENGTH)
coords, masks = [], []
for i, f in enumerate(frames):
c, m = lm.pack_frame(extractor.extract_frame(f, timestamp_ms=i * 33))
coords.append(c)
masks.append(m)
if not coords:
return (None, None) if return_landmarks else None
coords = np.stack(coords)
mask = np.stack(masks)
usable = (mask > 0).any(axis=1)
if int(usable.sum()) < config.MIN_VALID_FRAMES:
return (None, None) if return_landmarks else None
feats, _ = lm.normalise_sequence(coords, mask)
feats = seq_ops.resample_sequence(feats, length, method=config.SEQ_RESAMPLE_METHOD)
if not return_landmarks:
return feats
t, n, c = coords.shape
coords_rs = seq_ops.resample_sequence(coords.reshape(t, n * c), length,
method=config.SEQ_RESAMPLE_METHOD).reshape(length, n, c)
return feats, coords_rs
def features_from_video(path, extractor: Optional[MediaPipeExtractor] = None,
max_frames: Optional[int] = None,
length: Optional[int] = None, return_landmarks: bool = False):
"""Decode a video file and return (features, stats) for one clip.
When ``return_landmarks=True``, ``stats["landmarks"]`` carries the raw
per-frame coordinates behind the returned feature sequence (see
``features_from_rgb_frames``), as a plain nested list ready for JSON.
"""
import cv2
own = extractor is None
ex = extractor or MediaPipeExtractor()
stats = None
frames: List[np.ndarray] = []
try:
cap = cv2.VideoCapture(str(path))
if not cap.isOpened():
return None, {"error": "video_open_failed", "path": str(path)}
n = 0
limit = int(max_frames or config.MAX_FRAMES_READ)
while n < limit:
ok, frame = cap.read()
if not ok:
break
n += 1
frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
cap.release()
if not frames:
return None, {"error": "no_frames_decoded", "path": str(path)}
stats = {"path": str(path), "frames_decoded": len(frames)}
if return_landmarks:
feats, coords_rs = features_from_rgb_frames(frames, ex, length=length, return_landmarks=True)
if coords_rs is not None:
stats["landmarks"] = coords_rs.tolist()
else:
feats = features_from_rgb_frames(frames, ex, length=length)
return feats, stats
finally:
if own:
ex.close()
def dataset_features(filepath: str, length: Optional[int] = None) -> Optional[np.ndarray]:
"""Features for one dataset clip (used for demo/verification without a camera)."""
from backend.preprocessing import include50
coords, mask, _ = include50.read_keypoints_for(filepath)
if coords.shape[0] == 0:
return None
usable = (mask > 0).any(axis=1)
if int(usable.sum()) < config.MIN_VALID_FRAMES:
return None
feats, _ = lm.normalise_sequence(coords, mask)
return seq_ops.resample_sequence(feats, int(length or config.SEQUENCE_LENGTH))
|