tiena2cva commited on
Commit
e001594
·
1 Parent(s): 287f59f

feat(pose): add pluggable pose backends

Browse files
.gitignore CHANGED
@@ -4,3 +4,4 @@ __pycache__/
4
  .ruff_cache/
5
  .gradio/
6
  runs/
 
 
4
  .ruff_cache/
5
  .gradio/
6
  runs/
7
+ .env
src/pozify/steps/pose_landmarker.py CHANGED
@@ -48,7 +48,9 @@ def _iter_video_frames(
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()
@@ -68,7 +70,9 @@ def _pose_quality(landmarks: dict[str, dict[str, float]]) -> dict[str, Any]:
68
  "pose_warning": "pose_not_detected",
69
  }
70
 
71
- visibility_values = [landmark.get("visibility", 0.0) for landmark in landmarks.values()]
 
 
72
  critical_values = [
73
  landmarks[name].get("visibility", 0.0)
74
  for name in CRITICAL_LANDMARKS
@@ -86,15 +90,19 @@ def _pose_quality(landmarks: dict[str, dict[str, float]]) -> dict[str, Any]:
86
  return {
87
  "mean_visibility": round(sum(visibility_values) / len(visibility_values), 4),
88
  "critical_landmarks_visible": critical_visible,
89
- "full_body_visibility_proxy": round(sum(full_body_values) / len(full_body_values), 4)
90
- if full_body_values
91
- else 0.0,
 
 
92
  "landmark_count": len(landmarks),
93
  }
94
 
95
 
96
  def _empty_sequence() -> PoseSequence:
97
- return PoseSequence(frames=[], normalized=False, smoothing_method="none", pose_valid_ratio=0.0)
 
 
98
 
99
 
100
  def _run_with_backend(manifest: VideoManifest, backend: PoseBackend) -> PoseSequence:
@@ -113,7 +121,9 @@ def _run_with_backend(manifest: VideoManifest, backend: PoseBackend) -> PoseSequ
113
  frames.append(
114
  PoseFrame(
115
  frame_index=frame_index,
116
- timestamp_sec=round(frame_index / manifest.fps, 3) if manifest.fps else 0.0,
 
 
117
  landmarks=detection.landmarks,
118
  world_landmarks=detection.world_landmarks,
119
  pose_quality={
@@ -135,12 +145,16 @@ def _run_with_backend(manifest: VideoManifest, backend: PoseBackend) -> PoseSequ
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(
142
  frame_index=frame_index,
143
- timestamp_sec=round(frame_index / manifest.fps, 3) if manifest.fps else 0.0,
 
 
144
  landmarks=detection.landmarks,
145
  world_landmarks=detection.world_landmarks,
146
  pose_quality={
 
48
  if sample_count is None:
49
  frame_indices = range(manifest.total_frames)
50
  else:
51
+ frame_indices = sample_frame_indices(
52
+ manifest.total_frames, min(sample_count, manifest.total_frames)
53
+ )
54
  for frame_index in frame_indices:
55
  capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
56
  ok, frame = capture.read()
 
70
  "pose_warning": "pose_not_detected",
71
  }
72
 
73
+ visibility_values = [
74
+ landmark.get("visibility", 0.0) for landmark in landmarks.values()
75
+ ]
76
  critical_values = [
77
  landmarks[name].get("visibility", 0.0)
78
  for name in CRITICAL_LANDMARKS
 
90
  return {
91
  "mean_visibility": round(sum(visibility_values) / len(visibility_values), 4),
92
  "critical_landmarks_visible": critical_visible,
93
+ "full_body_visibility_proxy": (
94
+ round(sum(full_body_values) / len(full_body_values), 4)
95
+ if full_body_values
96
+ else 0.0
97
+ ),
98
  "landmark_count": len(landmarks),
99
  }
100
 
101
 
102
  def _empty_sequence() -> PoseSequence:
103
+ return PoseSequence(
104
+ frames=[], normalized=False, smoothing_method="none", pose_valid_ratio=0.0
105
+ )
106
 
107
 
108
  def _run_with_backend(manifest: VideoManifest, backend: PoseBackend) -> PoseSequence:
 
121
  frames.append(
122
  PoseFrame(
123
  frame_index=frame_index,
124
+ timestamp_sec=(
125
+ round(frame_index / manifest.fps, 3) if manifest.fps else 0.0
126
+ ),
127
  landmarks=detection.landmarks,
128
  world_landmarks=detection.world_landmarks,
129
  pose_quality={
 
145
  def _run_mock(manifest: VideoManifest) -> PoseSequence:
146
  frames: list[PoseFrame] = []
147
  backend = MockPoseBackend()
148
+ for frame_index in sample_frame_indices(
149
+ manifest.total_frames, DEFAULT_POSE_SAMPLE_COUNT
150
+ ):
151
  detection = backend.detect(None, frame_index=frame_index)
152
  frames.append(
153
  PoseFrame(
154
  frame_index=frame_index,
155
+ timestamp_sec=(
156
+ round(frame_index / manifest.fps, 3) if manifest.fps else 0.0
157
+ ),
158
  landmarks=detection.landmarks,
159
  world_landmarks=detection.world_landmarks,
160
  pose_quality={
tests/test_pose_steps.py CHANGED
@@ -21,7 +21,9 @@ from pozify.steps import pose_cleaning, pose_landmarker
21
  from pozify.steps.pose_backends import PoseDetection, landmark_list_to_dict
22
 
23
 
24
- def _landmark(x: float, y: float, z: float = 0.0, visibility: float = 0.9) -> SimpleNamespace:
 
 
25
  return SimpleNamespace(x=x, y=y, z=z, visibility=visibility)
26
 
27
 
@@ -67,7 +69,9 @@ class PoseStepTests(unittest.TestCase):
67
 
68
  def _write_video(self, frame_count: int = 4) -> Path:
69
  path = Path(self.temp_dir.name) / "pose.mp4"
70
- writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"mp4v"), 30.0, (640, 480))
 
 
71
  self.assertTrue(writer.isOpened())
72
  for frame_index in range(frame_count):
73
  frame = np.full((480, 640, 3), 120 + frame_index, dtype=np.uint8)
@@ -125,7 +129,9 @@ class PoseStepTests(unittest.TestCase):
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
131
  first = landmark_list_to_dict(first_landmarks)
 
21
  from pozify.steps.pose_backends import PoseDetection, landmark_list_to_dict
22
 
23
 
24
+ def _landmark(
25
+ x: float, y: float, z: float = 0.0, visibility: float = 0.9
26
+ ) -> SimpleNamespace:
27
  return SimpleNamespace(x=x, y=y, z=z, visibility=visibility)
28
 
29
 
 
69
 
70
  def _write_video(self, frame_count: int = 4) -> Path:
71
  path = Path(self.temp_dir.name) / "pose.mp4"
72
+ writer = cv2.VideoWriter(
73
+ str(path), cv2.VideoWriter_fourcc(*"mp4v"), 30.0, (640, 480)
74
+ )
75
  self.assertTrue(writer.isOpened())
76
  for frame_index in range(frame_count):
77
  frame = np.full((480, 640, 3), 120 + frame_index, dtype=np.uint8)
 
129
 
130
  self.assertEqual(len(sequence.frames), 130)
131
 
132
+ def test_pose_cleaning_interpolates_smooths_and_adds_normalized_fields(
133
+ self,
134
+ ) -> None:
135
  first_landmarks = _landmark_result(offset=0.0).pose_landmarks
136
  last_landmarks = _landmark_result(offset=0.2).pose_landmarks
137
  first = landmark_list_to_dict(first_landmarks)