tthhanh commited on
Commit
8f2c6ef
·
1 Parent(s): 43779d9

feat(pose): use dense frames for real pose extraction

Browse files
src/pozify/steps/pose_landmarker.py CHANGED
@@ -33,7 +33,11 @@ def _env_pose_backend() -> str:
33
  return os.getenv("POZIFY_POSE_BACKEND", "mediapipe")
34
 
35
 
36
- def _iter_video_frames(manifest: VideoManifest) -> Iterator[tuple[int, Any]]:
 
 
 
 
37
  if not manifest.video_path or manifest.total_frames <= 0:
38
  return
39
 
@@ -41,8 +45,11 @@ def _iter_video_frames(manifest: VideoManifest) -> Iterator[tuple[int, Any]]:
41
  try:
42
  if not capture.isOpened():
43
  return
44
- sample_count = min(DEFAULT_POSE_SAMPLE_COUNT, manifest.total_frames)
45
- for frame_index in sample_frame_indices(manifest.total_frames, sample_count):
 
 
 
46
  capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
47
  ok, frame = capture.read()
48
  if ok and frame is not None:
@@ -98,7 +105,7 @@ def _run_with_backend(manifest: VideoManifest, backend: PoseBackend) -> PoseSequ
98
  valid_frames = 0
99
 
100
  with backend as extractor:
101
- for frame_index, frame in _iter_video_frames(manifest):
102
  rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
103
  detection = extractor.detect(rgb_frame, frame_index=frame_index)
104
  if detection.landmarks:
@@ -128,7 +135,7 @@ def _run_with_backend(manifest: VideoManifest, backend: PoseBackend) -> PoseSequ
128
  def _run_mock(manifest: VideoManifest) -> PoseSequence:
129
  frames: list[PoseFrame] = []
130
  backend = MockPoseBackend()
131
- for frame_index in sample_frame_indices(manifest.total_frames):
132
  detection = backend.detect(None, frame_index=frame_index)
133
  frames.append(
134
  PoseFrame(
 
33
  return os.getenv("POZIFY_POSE_BACKEND", "mediapipe")
34
 
35
 
36
+ def _iter_video_frames(
37
+ manifest: VideoManifest,
38
+ *,
39
+ sample_count: int | None,
40
+ ) -> Iterator[tuple[int, Any]]:
41
  if not manifest.video_path or manifest.total_frames <= 0:
42
  return
43
 
 
45
  try:
46
  if not capture.isOpened():
47
  return
48
+ if sample_count is None:
49
+ frame_indices = range(manifest.total_frames)
50
+ else:
51
+ frame_indices = sample_frame_indices(manifest.total_frames, min(sample_count, manifest.total_frames))
52
+ for frame_index in frame_indices:
53
  capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
54
  ok, frame = capture.read()
55
  if ok and frame is not None:
 
105
  valid_frames = 0
106
 
107
  with backend as extractor:
108
+ for frame_index, frame in _iter_video_frames(manifest, sample_count=None):
109
  rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
110
  detection = extractor.detect(rgb_frame, frame_index=frame_index)
111
  if detection.landmarks:
 
135
  def _run_mock(manifest: VideoManifest) -> PoseSequence:
136
  frames: list[PoseFrame] = []
137
  backend = MockPoseBackend()
138
+ for frame_index in sample_frame_indices(manifest.total_frames, DEFAULT_POSE_SAMPLE_COUNT):
139
  detection = backend.detect(None, frame_index=frame_index)
140
  frames.append(
141
  PoseFrame(
tests/test_pose_steps.py CHANGED
@@ -103,6 +103,28 @@ class PoseStepTests(unittest.TestCase):
103
  self.assertTrue(sequence.frames[0].pose_quality["critical_landmarks_visible"])
104
  self.assertEqual(sequence.pose_valid_ratio, 1.0)
105
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
106
  def test_pose_cleaning_interpolates_smooths_and_adds_normalized_fields(self) -> None:
107
  first_landmarks = _landmark_result(offset=0.0).pose_landmarks
108
  last_landmarks = _landmark_result(offset=0.2).pose_landmarks
 
103
  self.assertTrue(sequence.frames[0].pose_quality["critical_landmarks_visible"])
104
  self.assertEqual(sequence.pose_valid_ratio, 1.0)
105
 
106
+ def test_pose_landmarker_uses_dense_frames_for_real_backend(self) -> None:
107
+ path = self._write_video(frame_count=130)
108
+ manifest = VideoManifest(
109
+ video_path=str(path),
110
+ fps=30.0,
111
+ duration_sec=4.333,
112
+ total_frames=130,
113
+ sampled_frames=12,
114
+ width=640,
115
+ height=480,
116
+ codec="mp4v",
117
+ container="mp4",
118
+ brightness_mean=120.0,
119
+ blur_laplacian_var=100.0,
120
+ quality_warnings=[],
121
+ analysis_allowed=True,
122
+ )
123
+
124
+ sequence = pose_landmarker.run(manifest, backend=_FakePose())
125
+
126
+ self.assertEqual(len(sequence.frames), 130)
127
+
128
  def test_pose_cleaning_interpolates_smooths_and_adds_normalized_fields(self) -> None:
129
  first_landmarks = _landmark_result(offset=0.0).pose_landmarks
130
  last_landmarks = _landmark_result(offset=0.2).pose_landmarks