nvti commited on
Commit
4108569
·
1 Parent(s): 77f35c0

feat(rep_counter): implement exercise-specific rep counters for push-ups, shoulder presses, and squats; add dependency on fastapi

Browse files
pyproject.toml CHANGED
@@ -5,6 +5,7 @@ description = "Gradio base app for Pozify with mocked pipeline steps and explici
5
  readme = "README.md"
6
  requires-python = ">=3.11"
7
  dependencies = [
 
8
  "gradio>=4.44.0",
9
  "mediapipe>=0.10.35",
10
  "numpy>=1.26.0",
 
5
  readme = "README.md"
6
  requires-python = ">=3.11"
7
  dependencies = [
8
+ "fastapi>=0.136.3",
9
  "gradio>=4.44.0",
10
  "mediapipe>=0.10.35",
11
  "numpy>=1.26.0",
requirements.txt CHANGED
@@ -38,7 +38,9 @@ contourpy==1.3.3
38
  cycler==0.12.1
39
  # via matplotlib
40
  fastapi==0.136.3
41
- # via gradio
 
 
42
  filelock==3.29.1
43
  # via huggingface-hub
44
  flatbuffers==25.12.19
 
38
  cycler==0.12.1
39
  # via matplotlib
40
  fastapi==0.136.3
41
+ # via
42
+ # gradio
43
+ # pozify
44
  filelock==3.29.1
45
  # via huggingface-hub
46
  flatbuffers==25.12.19
src/pozify/steps/rep_counter.py CHANGED
@@ -1,223 +1,10 @@
1
  from __future__ import annotations
2
 
3
- from dataclasses import asdict
4
  from typing import Any
5
 
6
- from pozify.contracts import ExerciseClassification, PoseSequence, Rep, Reps
7
- from pozify.steps.rep_signals import (
8
- SignalSample,
9
- angle_deg,
10
- average_axis,
11
- body_line_score,
12
- normalize_optional,
13
- samples_from_values,
14
- smooth_signal,
15
- )
16
- from pozify.steps.rep_state_machine import RepSegment, find_local_extrema, segment_low_high_low
17
-
18
-
19
- MIN_CYCLE_FRAMES = 12
20
- MIN_PHASE_FRAMES = 4
21
- MIN_USABLE_SIGNAL_SAMPLES = 9
22
- MIN_SIGNAL_RANGE = 0.22
23
-
24
-
25
- def _mean_optional(values: list[float | None]) -> float | None:
26
- usable = [value for value in values if value is not None]
27
- if not usable:
28
- return None
29
- return sum(usable) / len(usable)
30
-
31
-
32
- def _combine(primary: list[float | None], secondary: list[float | None], *, weight: float) -> list[float | None]:
33
- normalized_secondary = normalize_optional(secondary)
34
- combined: list[float | None] = []
35
- for primary_value, secondary_value in zip(primary, normalized_secondary, strict=False):
36
- if primary_value is None:
37
- combined.append(None)
38
- continue
39
- if secondary_value is None:
40
- combined.append(primary_value)
41
- continue
42
- combined.append(primary_value + secondary_value * weight)
43
- return combined
44
-
45
-
46
- def _primary_signal_for_exercise(sequence: PoseSequence, exercise: str) -> tuple[list[SignalSample], dict[str, Any]]:
47
- hip_y = [average_axis(frame, ("left_hip", "right_hip"), "y") for frame in sequence.frames]
48
- shoulder_y = [average_axis(frame, ("left_shoulder", "right_shoulder"), "y") for frame in sequence.frames]
49
- wrist_y = [average_axis(frame, ("left_wrist", "right_wrist"), "y") for frame in sequence.frames]
50
- knee_bend = [
51
- _mean_optional(
52
- [
53
- None if angle is None else max(0.0, 180.0 - angle)
54
- for angle in (
55
- angle_deg(frame, "left_hip", "left_knee", "left_ankle"),
56
- angle_deg(frame, "right_hip", "right_knee", "right_ankle"),
57
- )
58
- ]
59
- )
60
- for frame in sequence.frames
61
- ]
62
- elbow_bend = [
63
- _mean_optional(
64
- [
65
- None if angle is None else max(0.0, 180.0 - angle)
66
- for angle in (
67
- angle_deg(frame, "left_shoulder", "left_elbow", "left_wrist"),
68
- angle_deg(frame, "right_shoulder", "right_elbow", "right_wrist"),
69
- )
70
- ]
71
- )
72
- for frame in sequence.frames
73
- ]
74
- body_line = [body_line_score(frame) for frame in sequence.frames]
75
-
76
- if exercise == "squat":
77
- raw_signal = _combine(hip_y, knee_bend, weight=0.35)
78
- selected_signal = "hip_y_plus_knee_bend"
79
- elif exercise == "shoulder_press":
80
- inverted_wrist = [None if value is None else -value for value in wrist_y]
81
- raw_signal = _combine(inverted_wrist, [None if value is None else -value for value in elbow_bend], weight=0.2)
82
- selected_signal = "negative_wrist_y_plus_elbow_extension_proxy"
83
- else:
84
- chest_proxy = [
85
- _mean_optional([shoulder_value, hip_value])
86
- for shoulder_value, hip_value in zip(shoulder_y, hip_y, strict=False)
87
- ]
88
- raw_signal = _combine(chest_proxy, elbow_bend, weight=0.25)
89
- selected_signal = "chest_y_plus_elbow_bend"
90
-
91
- smoothed_signal = smooth_signal(raw_signal)
92
- normalized_signal = normalize_optional(smoothed_signal)
93
- samples = samples_from_values(sequence, normalized_signal)
94
- return samples, {
95
- "selected_signal": selected_signal,
96
- "raw_signal_range": (
97
- round(max((value for value in normalized_signal if value is not None), default=0.0)
98
- - min((value for value in normalized_signal if value is not None), default=0.0), 4)
99
- ),
100
- "usable_signal_samples": len(samples),
101
- "body_line_mean": round(_mean_optional(body_line) or 0.0, 4),
102
- }
103
-
104
-
105
- def _segments_to_reps(segments: list[RepSegment]) -> list[Rep]:
106
- return [
107
- Rep(
108
- rep_id=index + 1,
109
- start_frame=segment.start.frame_index,
110
- mid_frame=segment.middle.frame_index,
111
- end_frame=segment.end.frame_index,
112
- start_sec=round(segment.start.timestamp_sec, 3),
113
- mid_sec=round(segment.middle.timestamp_sec, 3),
114
- end_sec=round(segment.end.timestamp_sec, 3),
115
- )
116
- for index, segment in enumerate(segments)
117
- ]
118
-
119
-
120
- def _partial_reps(
121
- sequence: PoseSequence,
122
- segments: list[RepSegment],
123
- samples: list[SignalSample],
124
- *,
125
- signal_range: float,
126
- ) -> list[dict[str, Any]]:
127
- if not samples:
128
- return [{"reason": "low_signal_quality"}]
129
-
130
- partials: list[dict[str, Any]] = []
131
- if not segments:
132
- if signal_range >= MIN_SIGNAL_RANGE * 0.7:
133
- partials.append(
134
- {
135
- "reason": "insufficient_rom",
136
- "start_frame": samples[0].frame_index,
137
- "end_frame": samples[-1].frame_index,
138
- "start_sec": round(samples[0].timestamp_sec, 3),
139
- "end_sec": round(samples[-1].timestamp_sec, 3),
140
- }
141
- )
142
- return partials
143
-
144
- first_segment = segments[0]
145
- if first_segment.start.frame_index - samples[0].frame_index >= MIN_PHASE_FRAMES:
146
- partials.append(
147
- {
148
- "reason": "starts_mid_rep",
149
- "start_frame": samples[0].frame_index,
150
- "end_frame": first_segment.start.frame_index,
151
- "start_sec": round(samples[0].timestamp_sec, 3),
152
- "end_sec": round(first_segment.start.timestamp_sec, 3),
153
- }
154
- )
155
-
156
- last_segment = segments[-1]
157
- if samples[-1].frame_index - last_segment.end.frame_index >= MIN_PHASE_FRAMES:
158
- partials.append(
159
- {
160
- "reason": "ends_mid_rep",
161
- "start_frame": last_segment.end.frame_index,
162
- "end_frame": samples[-1].frame_index,
163
- "start_sec": round(last_segment.end.timestamp_sec, 3),
164
- "end_sec": round(samples[-1].timestamp_sec, 3),
165
- }
166
- )
167
- return partials
168
 
169
 
170
  def run(classification: ExerciseClassification, sequence: PoseSequence) -> tuple[Reps, dict[str, Any]]:
171
- if classification.exercise == "unknown":
172
- reps = Reps(exercise=classification.exercise, reps=[], partial_reps=[{"reason": "unknown_exercise"}])
173
- return reps, {"selected_signal": "none", "thresholds": {}, "extrema": [], "accepted_reps": []}
174
-
175
- samples, debug = _primary_signal_for_exercise(sequence, classification.exercise)
176
- signal_range = debug["raw_signal_range"]
177
- extrema = find_local_extrema(samples)
178
- segments = (
179
- segment_low_high_low(
180
- extrema,
181
- min_cycle_frames=MIN_CYCLE_FRAMES,
182
- min_phase_frames=MIN_PHASE_FRAMES,
183
- min_amplitude=max(MIN_SIGNAL_RANGE, signal_range * 0.35),
184
- )
185
- if len(samples) >= MIN_USABLE_SIGNAL_SAMPLES
186
- else []
187
- )
188
- partial_reps = _partial_reps(sequence, segments, samples, signal_range=signal_range)
189
- if sequence.pose_valid_ratio < 0.8:
190
- partial_reps.append({"reason": "low_pose_valid_ratio"})
191
-
192
- reps = Reps(
193
- exercise=classification.exercise,
194
- reps=_segments_to_reps(segments),
195
- partial_reps=partial_reps,
196
- )
197
- debug_payload = {
198
- **debug,
199
- "thresholds": {
200
- "min_cycle_frames": MIN_CYCLE_FRAMES,
201
- "min_phase_frames": MIN_PHASE_FRAMES,
202
- "min_amplitude": round(max(MIN_SIGNAL_RANGE, signal_range * 0.35), 4),
203
- },
204
- "extrema": [
205
- {
206
- "kind": extrema_item.kind,
207
- "frame_index": extrema_item.sample.frame_index,
208
- "timestamp_sec": round(extrema_item.sample.timestamp_sec, 3),
209
- "value": round(extrema_item.sample.value, 4),
210
- }
211
- for extrema_item in extrema
212
- ],
213
- "accepted_reps": [
214
- {
215
- "start": asdict(segment.start),
216
- "middle": asdict(segment.middle),
217
- "end": asdict(segment.end),
218
- "amplitude": round(segment.amplitude, 4),
219
- }
220
- for segment in segments
221
- ],
222
- }
223
- return reps, debug_payload
 
1
  from __future__ import annotations
2
 
 
3
  from typing import Any
4
 
5
+ from pozify.contracts import ExerciseClassification, PoseSequence, Reps
6
+ from pozify.steps.rep_counters import get_rep_counter
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7
 
8
 
9
  def run(classification: ExerciseClassification, sequence: PoseSequence) -> tuple[Reps, dict[str, Any]]:
10
+ return get_rep_counter(classification.exercise).count(sequence)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
src/pozify/steps/rep_counters/__init__.py ADDED
@@ -0,0 +1,5 @@
 
 
 
 
 
 
1
+ from pozify.steps.rep_counters.base import ExerciseRepCounter
2
+ from pozify.steps.rep_counters.registry import get_rep_counter
3
+
4
+ __all__ = ["ExerciseRepCounter", "get_rep_counter"]
5
+
src/pozify/steps/rep_counters/base.py ADDED
@@ -0,0 +1,180 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ from abc import ABC, abstractmethod
4
+ from dataclasses import asdict
5
+ from typing import Any
6
+
7
+ from pozify.contracts import PoseSequence, Rep, Reps
8
+ from pozify.steps.rep_signals import SignalSample, normalize_optional, samples_from_values, smooth_signal
9
+ from pozify.steps.rep_state_machine import RepSegment, find_local_extrema, segment_low_high_low
10
+
11
+
12
+ MIN_CYCLE_FRAMES = 12
13
+ MIN_PHASE_FRAMES = 4
14
+ MIN_USABLE_SIGNAL_SAMPLES = 9
15
+ MIN_SIGNAL_RANGE = 0.22
16
+
17
+
18
+ def mean_optional(values: list[float | None]) -> float | None:
19
+ usable = [value for value in values if value is not None]
20
+ if not usable:
21
+ return None
22
+ return sum(usable) / len(usable)
23
+
24
+
25
+ def combine(primary: list[float | None], secondary: list[float | None], *, weight: float) -> list[float | None]:
26
+ normalized_secondary = normalize_optional(secondary)
27
+ combined: list[float | None] = []
28
+ for primary_value, secondary_value in zip(primary, normalized_secondary, strict=False):
29
+ if primary_value is None:
30
+ combined.append(None)
31
+ continue
32
+ if secondary_value is None:
33
+ combined.append(primary_value)
34
+ continue
35
+ combined.append(primary_value + secondary_value * weight)
36
+ return combined
37
+
38
+
39
+ def normalized_samples(
40
+ sequence: PoseSequence,
41
+ raw_signal: list[float | None],
42
+ ) -> tuple[list[SignalSample], float]:
43
+ smoothed_signal = smooth_signal(raw_signal)
44
+ normalized_signal = normalize_optional(smoothed_signal)
45
+ samples = samples_from_values(sequence, normalized_signal)
46
+ signal_range = max((value for value in normalized_signal if value is not None), default=0.0) - min(
47
+ (value for value in normalized_signal if value is not None),
48
+ default=0.0,
49
+ )
50
+ return samples, round(signal_range, 4)
51
+
52
+
53
+ def segments_to_reps(segments: list[RepSegment]) -> list[Rep]:
54
+ return [
55
+ Rep(
56
+ rep_id=index + 1,
57
+ start_frame=segment.start.frame_index,
58
+ mid_frame=segment.middle.frame_index,
59
+ end_frame=segment.end.frame_index,
60
+ start_sec=round(segment.start.timestamp_sec, 3),
61
+ mid_sec=round(segment.middle.timestamp_sec, 3),
62
+ end_sec=round(segment.end.timestamp_sec, 3),
63
+ )
64
+ for index, segment in enumerate(segments)
65
+ ]
66
+
67
+
68
+ def partial_reps(
69
+ sequence: PoseSequence,
70
+ segments: list[RepSegment],
71
+ samples: list[SignalSample],
72
+ *,
73
+ signal_range: float,
74
+ ) -> list[dict[str, Any]]:
75
+ if not samples:
76
+ return [{"reason": "low_signal_quality"}]
77
+
78
+ partials: list[dict[str, Any]] = []
79
+ if not segments:
80
+ if signal_range >= MIN_SIGNAL_RANGE * 0.7:
81
+ partials.append(
82
+ {
83
+ "reason": "insufficient_rom",
84
+ "start_frame": samples[0].frame_index,
85
+ "end_frame": samples[-1].frame_index,
86
+ "start_sec": round(samples[0].timestamp_sec, 3),
87
+ "end_sec": round(samples[-1].timestamp_sec, 3),
88
+ }
89
+ )
90
+ return partials
91
+
92
+ first_segment = segments[0]
93
+ if first_segment.start.frame_index - samples[0].frame_index >= MIN_PHASE_FRAMES:
94
+ partials.append(
95
+ {
96
+ "reason": "starts_mid_rep",
97
+ "start_frame": samples[0].frame_index,
98
+ "end_frame": first_segment.start.frame_index,
99
+ "start_sec": round(samples[0].timestamp_sec, 3),
100
+ "end_sec": round(first_segment.start.timestamp_sec, 3),
101
+ }
102
+ )
103
+
104
+ last_segment = segments[-1]
105
+ if samples[-1].frame_index - last_segment.end.frame_index >= MIN_PHASE_FRAMES:
106
+ partials.append(
107
+ {
108
+ "reason": "ends_mid_rep",
109
+ "start_frame": last_segment.end.frame_index,
110
+ "end_frame": samples[-1].frame_index,
111
+ "start_sec": round(last_segment.end.timestamp_sec, 3),
112
+ "end_sec": round(samples[-1].timestamp_sec, 3),
113
+ }
114
+ )
115
+ return partials
116
+
117
+
118
+ class ExerciseRepCounter(ABC):
119
+ exercise: str
120
+ min_cycle_frames = MIN_CYCLE_FRAMES
121
+ min_phase_frames = MIN_PHASE_FRAMES
122
+ min_signal_range = MIN_SIGNAL_RANGE
123
+ min_usable_signal_samples = MIN_USABLE_SIGNAL_SAMPLES
124
+
125
+ @abstractmethod
126
+ def build_signal(self, sequence: PoseSequence) -> tuple[list[SignalSample], dict[str, Any]]:
127
+ """Build the exercise-specific normalized motion signal."""
128
+
129
+ def count(self, sequence: PoseSequence) -> tuple[Reps, dict[str, Any]]:
130
+ samples, debug = self.build_signal(sequence)
131
+ signal_range = debug["raw_signal_range"]
132
+ extrema = find_local_extrema(samples)
133
+ min_amplitude = max(self.min_signal_range, signal_range * 0.35)
134
+ segments = (
135
+ segment_low_high_low(
136
+ extrema,
137
+ min_cycle_frames=self.min_cycle_frames,
138
+ min_phase_frames=self.min_phase_frames,
139
+ min_amplitude=min_amplitude,
140
+ )
141
+ if len(samples) >= self.min_usable_signal_samples
142
+ else []
143
+ )
144
+ partials = partial_reps(sequence, segments, samples, signal_range=signal_range)
145
+ if sequence.pose_valid_ratio < 0.8:
146
+ partials.append({"reason": "low_pose_valid_ratio"})
147
+
148
+ reps = Reps(
149
+ exercise=self.exercise,
150
+ reps=segments_to_reps(segments),
151
+ partial_reps=partials,
152
+ )
153
+ debug_payload = {
154
+ **debug,
155
+ "thresholds": {
156
+ "min_cycle_frames": self.min_cycle_frames,
157
+ "min_phase_frames": self.min_phase_frames,
158
+ "min_amplitude": round(min_amplitude, 4),
159
+ },
160
+ "extrema": [
161
+ {
162
+ "kind": extrema_item.kind,
163
+ "frame_index": extrema_item.sample.frame_index,
164
+ "timestamp_sec": round(extrema_item.sample.timestamp_sec, 3),
165
+ "value": round(extrema_item.sample.value, 4),
166
+ }
167
+ for extrema_item in extrema
168
+ ],
169
+ "accepted_reps": [
170
+ {
171
+ "start": asdict(segment.start),
172
+ "middle": asdict(segment.middle),
173
+ "end": asdict(segment.end),
174
+ "amplitude": round(segment.amplitude, 4),
175
+ }
176
+ for segment in segments
177
+ ],
178
+ }
179
+ return reps, debug_payload
180
+
src/pozify/steps/rep_counters/push_up.py ADDED
@@ -0,0 +1,40 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ from typing import Any
4
+
5
+ from pozify.contracts import PoseSequence
6
+ from pozify.steps.rep_counters.base import ExerciseRepCounter, combine, mean_optional, normalized_samples
7
+ from pozify.steps.rep_signals import SignalSample, angle_deg, average_axis, body_line_score
8
+
9
+
10
+ class PushUpRepCounter(ExerciseRepCounter):
11
+ exercise = "push_up"
12
+
13
+ def build_signal(self, sequence: PoseSequence) -> tuple[list[SignalSample], dict[str, Any]]:
14
+ hip_y = [average_axis(frame, ("left_hip", "right_hip"), "y") for frame in sequence.frames]
15
+ shoulder_y = [average_axis(frame, ("left_shoulder", "right_shoulder"), "y") for frame in sequence.frames]
16
+ elbow_bend = [
17
+ mean_optional(
18
+ [
19
+ None if angle is None else max(0.0, 180.0 - angle)
20
+ for angle in (
21
+ angle_deg(frame, "left_shoulder", "left_elbow", "left_wrist"),
22
+ angle_deg(frame, "right_shoulder", "right_elbow", "right_wrist"),
23
+ )
24
+ ]
25
+ )
26
+ for frame in sequence.frames
27
+ ]
28
+ body_line = [body_line_score(frame) for frame in sequence.frames]
29
+ chest_proxy = [
30
+ mean_optional([shoulder_value, hip_value])
31
+ for shoulder_value, hip_value in zip(shoulder_y, hip_y, strict=False)
32
+ ]
33
+ samples, signal_range = normalized_samples(sequence, combine(chest_proxy, elbow_bend, weight=0.25))
34
+ return samples, {
35
+ "selected_signal": "chest_y_plus_elbow_bend",
36
+ "raw_signal_range": signal_range,
37
+ "usable_signal_samples": len(samples),
38
+ "body_line_mean": round(mean_optional(body_line) or 0.0, 4),
39
+ }
40
+
src/pozify/steps/rep_counters/registry.py ADDED
@@ -0,0 +1,34 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ from typing import Any
4
+
5
+ from pozify.contracts import PoseSequence, Reps
6
+ from pozify.steps.rep_counters.base import ExerciseRepCounter
7
+ from pozify.steps.rep_counters.push_up import PushUpRepCounter
8
+ from pozify.steps.rep_counters.shoulder_press import ShoulderPressRepCounter
9
+ from pozify.steps.rep_counters.squat import SquatRepCounter
10
+ from pozify.steps.rep_signals import SignalSample
11
+
12
+
13
+ class UnknownRepCounter(ExerciseRepCounter):
14
+ exercise = "unknown"
15
+
16
+ def build_signal(self, sequence: PoseSequence) -> tuple[list[SignalSample], dict[str, Any]]:
17
+ return [], {"selected_signal": "none", "thresholds": {}, "extrema": [], "accepted_reps": []}
18
+
19
+ def count(self, sequence: PoseSequence) -> tuple[Reps, dict[str, Any]]:
20
+ reps = Reps(exercise=self.exercise, reps=[], partial_reps=[{"reason": "unknown_exercise"}])
21
+ return reps, {"selected_signal": "none", "thresholds": {}, "extrema": [], "accepted_reps": []}
22
+
23
+
24
+ REP_COUNTERS: dict[str, ExerciseRepCounter] = {
25
+ "push_up": PushUpRepCounter(),
26
+ "shoulder_press": ShoulderPressRepCounter(),
27
+ "squat": SquatRepCounter(),
28
+ "unknown": UnknownRepCounter(),
29
+ }
30
+
31
+
32
+ def get_rep_counter(exercise: str) -> ExerciseRepCounter:
33
+ return REP_COUNTERS.get(exercise, REP_COUNTERS["unknown"])
34
+
src/pozify/steps/rep_counters/shoulder_press.py ADDED
@@ -0,0 +1,38 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ from typing import Any
4
+
5
+ from pozify.contracts import PoseSequence
6
+ from pozify.steps.rep_counters.base import ExerciseRepCounter, combine, mean_optional, normalized_samples
7
+ from pozify.steps.rep_signals import SignalSample, angle_deg, average_axis, body_line_score
8
+
9
+
10
+ class ShoulderPressRepCounter(ExerciseRepCounter):
11
+ exercise = "shoulder_press"
12
+
13
+ def build_signal(self, sequence: PoseSequence) -> tuple[list[SignalSample], dict[str, Any]]:
14
+ wrist_y = [average_axis(frame, ("left_wrist", "right_wrist"), "y") for frame in sequence.frames]
15
+ elbow_bend = [
16
+ mean_optional(
17
+ [
18
+ None if angle is None else max(0.0, 180.0 - angle)
19
+ for angle in (
20
+ angle_deg(frame, "left_shoulder", "left_elbow", "left_wrist"),
21
+ angle_deg(frame, "right_shoulder", "right_elbow", "right_wrist"),
22
+ )
23
+ ]
24
+ )
25
+ for frame in sequence.frames
26
+ ]
27
+ body_line = [body_line_score(frame) for frame in sequence.frames]
28
+ inverted_wrist = [None if value is None else -value for value in wrist_y]
29
+ inverted_elbow_bend = [None if value is None else -value for value in elbow_bend]
30
+ raw_signal = combine(inverted_wrist, inverted_elbow_bend, weight=0.2)
31
+ samples, signal_range = normalized_samples(sequence, raw_signal)
32
+ return samples, {
33
+ "selected_signal": "negative_wrist_y_plus_elbow_extension_proxy",
34
+ "raw_signal_range": signal_range,
35
+ "usable_signal_samples": len(samples),
36
+ "body_line_mean": round(mean_optional(body_line) or 0.0, 4),
37
+ }
38
+
src/pozify/steps/rep_counters/squat.py ADDED
@@ -0,0 +1,35 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ from typing import Any
4
+
5
+ from pozify.contracts import PoseSequence
6
+ from pozify.steps.rep_counters.base import ExerciseRepCounter, combine, mean_optional, normalized_samples
7
+ from pozify.steps.rep_signals import SignalSample, angle_deg, average_axis, body_line_score
8
+
9
+
10
+ class SquatRepCounter(ExerciseRepCounter):
11
+ exercise = "squat"
12
+
13
+ def build_signal(self, sequence: PoseSequence) -> tuple[list[SignalSample], dict[str, Any]]:
14
+ hip_y = [average_axis(frame, ("left_hip", "right_hip"), "y") for frame in sequence.frames]
15
+ knee_bend = [
16
+ mean_optional(
17
+ [
18
+ None if angle is None else max(0.0, 180.0 - angle)
19
+ for angle in (
20
+ angle_deg(frame, "left_hip", "left_knee", "left_ankle"),
21
+ angle_deg(frame, "right_hip", "right_knee", "right_ankle"),
22
+ )
23
+ ]
24
+ )
25
+ for frame in sequence.frames
26
+ ]
27
+ body_line = [body_line_score(frame) for frame in sequence.frames]
28
+ samples, signal_range = normalized_samples(sequence, combine(hip_y, knee_bend, weight=0.35))
29
+ return samples, {
30
+ "selected_signal": "hip_y_plus_knee_bend",
31
+ "raw_signal_range": signal_range,
32
+ "usable_signal_samples": len(samples),
33
+ "body_line_mean": round(mean_optional(body_line) or 0.0, 4),
34
+ }
35
+
tests/test_rep_counter.py CHANGED
@@ -9,6 +9,10 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
9
 
10
  from pozify.contracts import ExerciseClassification, PoseFrame, PoseSequence
11
  from pozify.steps import rep_counter
 
 
 
 
12
 
13
 
14
  def _frame(frame_index: int, landmarks: dict[str, dict[str, float]]) -> PoseFrame:
@@ -145,6 +149,11 @@ class RepCounterTests(unittest.TestCase):
145
  self.assertEqual(reps.partial_reps, [{"reason": "unknown_exercise"}])
146
  self.assertEqual(debug["selected_signal"], "none")
147
 
 
 
 
 
 
148
 
149
  if __name__ == "__main__":
150
  unittest.main()
 
9
 
10
  from pozify.contracts import ExerciseClassification, PoseFrame, PoseSequence
11
  from pozify.steps import rep_counter
12
+ from pozify.steps.rep_counters import get_rep_counter
13
+ from pozify.steps.rep_counters.push_up import PushUpRepCounter
14
+ from pozify.steps.rep_counters.shoulder_press import ShoulderPressRepCounter
15
+ from pozify.steps.rep_counters.squat import SquatRepCounter
16
 
17
 
18
  def _frame(frame_index: int, landmarks: dict[str, dict[str, float]]) -> PoseFrame:
 
149
  self.assertEqual(reps.partial_reps, [{"reason": "unknown_exercise"}])
150
  self.assertEqual(debug["selected_signal"], "none")
151
 
152
+ def test_exercises_resolve_to_specific_rep_counter_strategies(self) -> None:
153
+ self.assertIsInstance(get_rep_counter("push_up"), PushUpRepCounter)
154
+ self.assertIsInstance(get_rep_counter("shoulder_press"), ShoulderPressRepCounter)
155
+ self.assertIsInstance(get_rep_counter("squat"), SquatRepCounter)
156
+
157
 
158
  if __name__ == "__main__":
159
  unittest.main()
uv.lock CHANGED
@@ -1289,6 +1289,7 @@ name = "pozify"
1289
  version = "0.1.0"
1290
  source = { virtual = "." }
1291
  dependencies = [
 
1292
  { name = "gradio" },
1293
  { name = "mediapipe" },
1294
  { name = "numpy" },
@@ -1303,6 +1304,7 @@ dev = [
1303
 
1304
  [package.metadata]
1305
  requires-dist = [
 
1306
  { name = "gradio", specifier = ">=4.44.0" },
1307
  { name = "mediapipe", specifier = ">=0.10.35" },
1308
  { name = "numpy", specifier = ">=1.26.0" },
 
1289
  version = "0.1.0"
1290
  source = { virtual = "." }
1291
  dependencies = [
1292
+ { name = "fastapi" },
1293
  { name = "gradio" },
1294
  { name = "mediapipe" },
1295
  { name = "numpy" },
 
1304
 
1305
  [package.metadata]
1306
  requires-dist = [
1307
+ { name = "fastapi", specifier = ">=0.136.3" },
1308
  { name = "gradio", specifier = ">=4.44.0" },
1309
  { name = "mediapipe", specifier = ">=0.10.35" },
1310
  { name = "numpy", specifier = ">=1.26.0" },