hjwang/posefit-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(exercise): 优化死虫式训练姿态检测算法

Browse Source 
- 调整视频处理频率从每帧处理改为每2帧处理
- 添加膝角趋势平滑算法减少单帧抖动误判
- 改进对角伸展检测逻辑支持准备位手臂上举
- 优化状态机确保严格回到准备姿态才计数
- 添加姿态丢失时的候选帧清理机制
- 更新音频文件生成路径至resources目录
- 改进macOS音频生成使用AIFF格式提高质量
- 添加详细的帧处理日志输出间隔配置
This commit is contained in:
wsy182
2026-06-10 22:57:35 +08:00
parent ea0c007441
commit 6dee2a2ff3
9 changed files with 281 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/audio/generate.py
+29 -24
View File
@@ -21,27 +21,27 @@ def generate_rep_audio_files(
默认生成到：
app/audio/reps/0.wav
app/audio/reps/1.wav
resources/audio/reps/0.aiff # macOS
resources/audio/reps/0.wav # Windows / Linux
...
app/audio/reps/200.wav
resources/audio/reps/200.aiff 或 200.wav
服务启动时调用一次即可。
"""
output_dir.mkdir(parents=True, exist_ok=True)
system = platform.system().lower()
suffix = ".aiff" if system == "darwin" else ".wav"
missing_counts = [
count
for count in range(0, max_count + 1)
if overwrite or not _audio_path(output_dir, count).exists()
if overwrite or not _audio_path(output_dir, count, suffix=suffix).exists()
]
if not missing_counts:
logger.info("Rep audio files already prepared: {}", output_dir)
return
system = platform.system().lower()
logger.info(
"Preparing rep audio files, system={}, count={}, output_dir={}",
system,
@@ -79,22 +79,27 @@ def _generate_with_macos_say(
raise RuntimeError("macOS say command not found")
for count in counts:
audio_file = _audio_path(output_dir, count)
audio_file = _audio_path(output_dir, count, suffix=".aiff")
subprocess.run(
[
"say",
"-r",
str(rate),
"--file-format=WAVE",
"-o",
str(audio_file),
str(count),
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=True,
)
try:
subprocess.run(
[
"say",
"-r",
str(rate),
"--file-format=AIFF",
"-o",
str(audio_file),
str(count),
],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
check=True,
)
except subprocess.CalledProcessError as exc:
message = exc.stderr.strip() or f"exit status {exc.returncode}"
raise RuntimeError(f"Failed to generate {audio_file}: {message}") from exc
def _generate_with_pyttsx3(
@@ -114,11 +119,11 @@ def _generate_with_pyttsx3(
engine.setProperty("volume", 1.0)
for count in counts:
audio_file = _audio_path(output_dir, count)
audio_file = _audio_path(output_dir, count, suffix=".wav")
engine.save_to_file(str(count), str(audio_file))
engine.runAndWait()
def _audio_path(output_dir: Path, count: int) -> Path:
return output_dir / f"{count}.wav"
def _audio_path(output_dir: Path, count: int, *, suffix: str) -> Path:
return output_dir / f"{count}{suffix}"
app/exercises/dead_bug/detector.py
+2
View File
@@ -110,6 +110,7 @@ class DeadBugDetector:
annotated = bgr_frame.copy()
if pose_result is None or not pose_result.pose_landmarks:
self._state.mark_no_pose()
result = DeadBugResult(
rep_count=self._state.rep_count,
phase=DeadBugPhase.NO_POSE,
@@ -125,6 +126,7 @@ class DeadBugDetector:
draw_landmarks(annotated, landmarks, REQUIRED_LANDMARKS, visibility_threshold=self.visibility_threshold)
if not has_required_visibility(landmarks, REQUIRED_LANDMARKS, self.visibility_threshold):
self._state.mark_no_pose()
result = DeadBugResult(
rep_count=self._state.rep_count,
phase=DeadBugPhase.NO_POSE,
app/exercises/dead_bug/rules.py
+3 -3
View File
@@ -9,7 +9,7 @@ def has_required_visibility(landmarks: list[Point], required_indices: tuple[int,
def detect_diagonal_extension(metrics: DeadBugMetrics) -> str | None:
"""检测是否存在对角伸展（左臂+右腿 或 右臂+左腿）"""
"""检测对角伸展（腿部只允许单侧伸展，手臂允许准备位上举带来的识别重叠）"""
if metrics.left_leg_extended and metrics.right_leg_extended:
return None
@@ -21,7 +21,7 @@ def detect_diagonal_extension(metrics: DeadBugMetrics) -> str | None:
def is_ready_position(metrics: DeadBugMetrics) -> bool:
"""判断是否处于准备姿态（膝盖弯曲且四肢未伸展）"""
"""判断是否处于准备姿态（双膝弯曲且双腿未伸展；dead bug 准备位允许手臂上举）"""
knees_bent = metrics.left_knee_angle <= 140 and metrics.right_knee_angle <= 140
legs_not_extended = not metrics.left_leg_extended and not metrics.right_leg_extended
return knees_bent and legs_not_extended and detect_diagonal_extension(metrics) is None
return knees_bent and legs_not_extended
app/exercises/dead_bug/state_machine.py
+118 -8
View File
@@ -3,6 +3,13 @@ from __future__ import annotations
from app.exercises.dead_bug.rules import detect_diagonal_extension, is_ready_position
from app.exercises.dead_bug.types import DeadBugMetrics, DeadBugPhase, DeadBugResult
_SMOOTHING_ALPHA = 0.45
_TREND_DELTA_DEGREES = 1.5
_SIDE_ANGLE_MARGIN = 10.0
_EXTENSION_START_ANGLE = 125.0
_EXTENSION_PEAK_ANGLE = 150.0
_READY_KNEE_ANGLE = 140.0
class DeadBugStateMachine:
"""死虫式动作状态机：管理READY/EXTENDING/NEED_RESET/NO_POSE状态转换"""
@@ -17,11 +24,26 @@ class DeadBugStateMachine:
self._candidate_side: str | None = None
self._candidate_frames = 0
self._reset_frames = 0
self._smooth_left_knee_angle: float | None = None
self._smooth_right_knee_angle: float | None = None
self._left_knee_delta = 0.0
self._right_knee_delta = 0.0
def mark_no_pose(self) -> None:
"""姿态丢失时清掉候选帧；已确认的半程动作保留，等待重新可见后完成回收。"""
if self.phase == DeadBugPhase.READY:
self.phase = DeadBugPhase.NO_POSE
self.active_side = None
self._candidate_side = None
self._candidate_frames = 0
self._reset_frames = 0
def update(self, metrics: DeadBugMetrics) -> DeadBugResult:
"""根据传入指标更新状态机并返回本次结果"""
side = detect_diagonal_extension(metrics)
ready = is_ready_position(metrics)
self._update_knee_trends(metrics)
side = self._detect_motion_side(metrics)
ready = self._is_stable_ready(metrics)
if side is None:
self._candidate_side = None
@@ -33,13 +55,16 @@ class DeadBugStateMachine:
self._candidate_frames = 1
if self.phase in (DeadBugPhase.READY, DeadBugPhase.NO_POSE):
if ready:
self.phase = DeadBugPhase.READY
if self._candidate_frames >= self.extension_confirm_frames and side is not None:
self.phase = DeadBugPhase.EXTENDING
self.active_side = side
self._reset_frames = 0
elif self.phase == DeadBugPhase.EXTENDING:
if side == self.active_side:
if ready or self._active_knee_retracting():
self.phase = DeadBugPhase.NEED_RESET
self._reset_frames = 1 if ready else 0
elif self.phase == DeadBugPhase.NEED_RESET:
if ready:
self._reset_frames += 1
@@ -54,21 +79,106 @@ class DeadBugStateMachine:
self._reset_frames = 0
feedback = list(metrics.feedback)
if side is None and not ready:
display_side = detect_diagonal_extension(metrics)
if display_side is None and not ready:
feedback.append("Extend opposite arm and leg only")
if ready:
feedback.append("Ready position")
elif side == "left_arm_right_leg":
elif display_side == "left_arm_right_leg":
feedback.append("Left arm + right leg")
elif side == "right_arm_left_leg":
elif display_side == "right_arm_left_leg":
feedback.append("Right arm + left leg")
is_standard = side is not None and not metrics.feedback
is_standard = display_side is not None and not metrics.feedback
return DeadBugResult(
rep_count=self.rep_count,
phase=self.phase,
side=side,
side=display_side,
is_standard=is_standard,
feedback=feedback[:3],
metrics=metrics,
)
def _update_knee_trends(self, metrics: DeadBugMetrics) -> None:
"""更新平滑膝角和本帧变化量，用连续趋势抵消单帧抖动。"""
previous_left = self._smooth_left_knee_angle
previous_right = self._smooth_right_knee_angle
if previous_left is None:
self._smooth_left_knee_angle = metrics.left_knee_angle
self._left_knee_delta = 0.0
else:
self._smooth_left_knee_angle = (
_SMOOTHING_ALPHA * metrics.left_knee_angle
+ (1.0 - _SMOOTHING_ALPHA) * previous_left
)
self._left_knee_delta = self._smooth_left_knee_angle - previous_left
if previous_right is None:
self._smooth_right_knee_angle = metrics.right_knee_angle
self._right_knee_delta = 0.0
else:
self._smooth_right_knee_angle = (
_SMOOTHING_ALPHA * metrics.right_knee_angle
+ (1.0 - _SMOOTHING_ALPHA) * previous_right
)
self._right_knee_delta = self._smooth_right_knee_angle - previous_right
def _detect_motion_side(self, metrics: DeadBugMetrics) -> str | None:
"""基于膝角趋势推断正在伸展的对角侧，手臂只作为辅助校验。"""
raw_side = detect_diagonal_extension(metrics)
if raw_side is not None and self._side_has_extension_motion(raw_side):
return raw_side
left = self._smooth_left_knee_angle
right = self._smooth_right_knee_angle
if left is None or right is None:
return raw_side
both_legs_high = left >= _EXTENSION_PEAK_ANGLE and right >= _EXTENSION_PEAK_ANGLE
if both_legs_high and abs(left - right) < _SIDE_ANGLE_MARGIN:
return None
if (
right >= _EXTENSION_START_ANGLE
and right - left >= _SIDE_ANGLE_MARGIN
and (self._right_knee_delta >= _TREND_DELTA_DEGREES or right >= _EXTENSION_PEAK_ANGLE)
):
return "left_arm_right_leg"
if (
left >= _EXTENSION_START_ANGLE
and left - right >= _SIDE_ANGLE_MARGIN
and (self._left_knee_delta >= _TREND_DELTA_DEGREES or left >= _EXTENSION_PEAK_ANGLE)
):
return "right_arm_left_leg"
return raw_side
def _side_has_extension_motion(self, side: str) -> bool:
"""确认对应腿处于伸展区或仍在伸展趋势中。"""
if side == "left_arm_right_leg":
angle = self._smooth_right_knee_angle
delta = self._right_knee_delta
else:
angle = self._smooth_left_knee_angle
delta = self._left_knee_delta
if angle is None:
return True
return angle >= _EXTENSION_PEAK_ANGLE or (
angle >= _EXTENSION_START_ANGLE and delta >= _TREND_DELTA_DEGREES
)
def _is_stable_ready(self, metrics: DeadBugMetrics) -> bool:
"""准备位需要双腿回到屈膝区域；使用平滑膝角避免单帧阈值跳变。"""
left = self._smooth_left_knee_angle
right = self._smooth_right_knee_angle
if left is None or right is None:
return is_ready_position(metrics)
return left <= _READY_KNEE_ANGLE and right <= _READY_KNEE_ANGLE
def _active_knee_retracting(self) -> bool:
"""确认已伸展的那条腿开始回收。"""
if self.active_side == "left_arm_right_leg":
return self._right_knee_delta <= -_TREND_DELTA_DEGREES
if self.active_side == "right_arm_left_leg":
return self._left_knee_delta <= -_TREND_DELTA_DEGREES
return False
app/webrtc/video_receiver.py
+7 -2
View File
@@ -33,7 +33,12 @@ class VideoReceiver:
async def run(self) -> None:
"""持续接收视频帧并进行姿态检测、渲染和语音播报"""
logger.info("Start receiving video frames, process_every_n={}", config.video.process_every_n_frames)
log_every_n_frames = max(1, config.video.log_every_n_frames)
logger.info(
"Start receiving video frames, process_every_n={}, log_every_n={}",
config.video.process_every_n_frames,
log_every_n_frames,
)
frame_count = 0
processed_count = 0
@@ -70,7 +75,7 @@ class VideoReceiver:
display_img = last_annotated if last_annotated is not None else raw_img
show_frame(display_img)
if frame_count % 100 == 0:
if frame_count % log_every_n_frames == 0:
logger.info(
"Received {} frames, processed={}, raw_shape={}, reps={}, phase={}, feedback={}, {}",
frame_count,
config.yaml
+2 -1
View File
@@ -7,7 +7,8 @@ server:
max_ws_size: 10485760 # 10 MB
video:
process_every_n_frames: 1
process_every_n_frames: 2
log_every_n_frames: 2
model:
path: "./pose_models/pose_landmarker_full.task"
configs/models.py
+2 -1
View File
@@ -15,7 +15,8 @@ class ServerConfig:
@dataclass
class VideoConfig:
"""视频帧处理配置"""
process_every_n_frames: int = 1
process_every_n_frames: int = 2
log_every_n_frames: int = 2
@dataclass
tests/test_dead_bug_rules.py
+33
View File
@@ -49,6 +49,28 @@ class TestDeadBugRules:
)
assert detect_diagonal_extension(metrics) == "left_arm_right_leg"
def test_detect_diagonal_extension_allows_ready_arm_overlap(self):
"""测试：准备位双臂上举时，单腿对侧伸展仍应识别"""
metrics = DeadBugMetrics(
left_arm_extended=True, right_arm_extended=True,
left_leg_extended=False, right_leg_extended=True,
left_elbow_angle=160, right_elbow_angle=160,
left_knee_angle=90, right_knee_angle=160,
feedback=[],
)
assert detect_diagonal_extension(metrics) == "left_arm_right_leg"
def test_detect_diagonal_extension_rejects_both_legs(self):
"""测试：双腿同时伸展不应识别为可计数对角伸展"""
metrics = DeadBugMetrics(
left_arm_extended=True, right_arm_extended=True,
left_leg_extended=True, right_leg_extended=True,
left_elbow_angle=160, right_elbow_angle=160,
left_knee_angle=160, right_knee_angle=160,
feedback=[],
)
assert detect_diagonal_extension(metrics) is None
def test_is_ready_position(self):
"""测试：膝盖弯曲且四肢收缩应识别为准备姿态"""
metrics = DeadBugMetrics(
@@ -60,6 +82,17 @@ class TestDeadBugRules:
)
assert is_ready_position(metrics)
def test_is_ready_allows_arms_extended(self):
"""测试：dead bug 准备位允许双臂上举"""
metrics = DeadBugMetrics(
left_arm_extended=True, right_arm_extended=True,
left_leg_extended=False, right_leg_extended=False,
left_elbow_angle=160, right_elbow_angle=160,
left_knee_angle=100, right_knee_angle=100,
feedback=[],
)
assert is_ready_position(metrics)
def test_is_not_ready_legs_extended(self):
"""测试：腿部伸展时不识别为准备姿态"""
metrics = DeadBugMetrics(
tests/test_dead_bug_state_machine.py
+85
View File
@@ -26,6 +26,36 @@ class TestDeadBugStateMachine:
feedback=[],
)
def _both_legs_extended(self) -> DeadBugMetrics:
"""构建双腿同时伸展的非标准姿态"""
return DeadBugMetrics(
left_arm_extended=True, right_arm_extended=True,
left_leg_extended=True, right_leg_extended=True,
left_elbow_angle=160, right_elbow_angle=160,
left_knee_angle=160, right_knee_angle=160,
feedback=[],
)
def _arms_extended_ready_legs(self) -> DeadBugMetrics:
"""构建腿已收回但手臂未收回的姿态"""
return DeadBugMetrics(
left_arm_extended=True, right_arm_extended=True,
left_leg_extended=False, right_leg_extended=False,
left_elbow_angle=160, right_elbow_angle=160,
left_knee_angle=100, right_knee_angle=100,
feedback=[],
)
def _right_knee_angle(self, angle: float) -> DeadBugMetrics:
"""构建右膝角连续变化但伸展布尔值尚未稳定的姿态"""
return DeadBugMetrics(
left_arm_extended=True, right_arm_extended=True,
left_leg_extended=False, right_leg_extended=False,
left_elbow_angle=160, right_elbow_angle=160,
left_knee_angle=90, right_knee_angle=angle,
feedback=[],
)
def test_initial_state(self):
"""测试：状态机初始化后应为READY且计数为0"""
sm = DeadBugStateMachine()
@@ -46,3 +76,58 @@ class TestDeadBugStateMachine:
assert sm.phase == DeadBugPhase.READY
sm.update(self._extended_left())
assert sm.phase == DeadBugPhase.EXTENDING
def test_confirm_extension_from_knee_angle_trend(self):
"""测试：膝角连续上升时，不依赖单帧伸展布尔值也能确认伸展"""
sm = DeadBugStateMachine(extension_confirm_frames=2, reset_confirm_frames=2)
sm.update(self._right_knee_angle(100))
sm.update(self._right_knee_angle(130))
assert sm.phase == DeadBugPhase.READY
sm.update(self._right_knee_angle(145))
sm.update(self._right_knee_angle(150))
assert sm.phase == DeadBugPhase.EXTENDING
def test_full_rep_counts_once_after_strict_reset(self):
"""测试：确认伸展后，只有严格回到准备姿态才计一次"""
sm = DeadBugStateMachine(extension_confirm_frames=2, reset_confirm_frames=2)
sm.update(self._extended_left())
sm.update(self._extended_left())
assert sm.phase == DeadBugPhase.EXTENDING
sm.update(self._arms_extended_ready_legs())
assert sm.rep_count == 0
assert sm.phase == DeadBugPhase.NEED_RESET
sm.update(self._ready_metrics())
result = sm.update(self._ready_metrics())
assert result.rep_count == 1
assert sm.phase == DeadBugPhase.READY
result = sm.update(self._ready_metrics())
assert result.rep_count == 1
def test_both_legs_do_not_start_rep(self):
"""测试：双腿同时伸展不进入计数流程"""
sm = DeadBugStateMachine(extension_confirm_frames=2, reset_confirm_frames=2)
sm.update(self._both_legs_extended())
result = sm.update(self._both_legs_extended())
assert result.rep_count == 0
assert sm.phase == DeadBugPhase.READY
def test_no_pose_preserves_confirmed_rep_until_reset(self):
"""测试：已确认伸展后短暂丢姿态，回到准备位仍能完成计数"""
sm = DeadBugStateMachine(extension_confirm_frames=2, reset_confirm_frames=2)
sm.update(self._extended_left())
sm.update(self._extended_left())
assert sm.phase == DeadBugPhase.EXTENDING
sm.mark_no_pose()
sm.update(self._ready_metrics())
result = sm.update(self._ready_metrics())
assert result.rep_count == 1
assert sm.phase == DeadBugPhase.READY