Initial commit: Happy Up monorepo through Sprint 5.

Document-driven MVP with FastAPI backend, Vue H5, WeChat mini shell, product demo, and Docker dev stack. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-23 11:42:40 +08:00
commit 1aaef71f52
116 changed files with 10550 additions and 0 deletions
@@ -0,0 +1,191 @@
+"""Pure posture metric helpers — testable without MediaPipe/OpenCV."""
+
+from __future__ import annotations
+
+import math
+from statistics import mean, pstdev
+
+# MediaPipe Pose landmark indices
+NOSE = 0
+LEFT_SHOULDER = 11
+RIGHT_SHOULDER = 12
+LEFT_HIP = 23
+RIGHT_HIP = 24
+
+Landmark = dict[str, float]
+FrameLandmarks = dict[int, Landmark]
+
+METRIC_NAMES = ("头前伸", "高低肩", "骨盆倾斜")
+
+
+def _point(frame: FrameLandmarks, idx: int) -> Landmark | None:
+    point = frame.get(idx)
+    if not point or point.get("visibility", 0) < 0.5:
+        return None
+    return point
+
+
+def _shoulder_width(frame: FrameLandmarks) -> float | None:
+    left = _point(frame, LEFT_SHOULDER)
+    right = _point(frame, RIGHT_SHOULDER)
+    if not left or not right:
+        return None
+    width = abs(right["x"] - left["x"])
+    return width if width > 1e-4 else None
+
+
+def score_head_forward(frame: FrameLandmarks) -> float | None:
+    nose = _point(frame, NOSE)
+    left = _point(frame, LEFT_SHOULDER)
+    right = _point(frame, RIGHT_SHOULDER)
+    width = _shoulder_width(frame)
+    if not nose or not left or not right or not width:
+        return None
+    mid_x = (left["x"] + right["x"]) / 2
+    offset = abs(nose["x"] - mid_x) / width
+    return min(100.0, max(0.0, offset * 180))
+
+
+def score_shoulder_asymmetry(frame: FrameLandmarks) -> float | None:
+    left = _point(frame, LEFT_SHOULDER)
+    right = _point(frame, RIGHT_SHOULDER)
+    width = _shoulder_width(frame)
+    if not left or not right or not width:
+        return None
+    diff = abs(left["y"] - right["y"]) / width
+    return min(100.0, max(0.0, diff * 220))
+
+
+def score_pelvic_tilt(frame: FrameLandmarks) -> float | None:
+    left = _point(frame, LEFT_HIP)
+    right = _point(frame, RIGHT_HIP)
+    if not left or not right:
+        return None
+    angle = abs(math.degrees(math.atan2(right["y"] - left["y"], right["x"] - left["x"])))
+    tilt = min(angle, 180 - angle)
+    return min(100.0, max(0.0, tilt * 4.5))
+
+
+def value_to_level(value: float) -> str:
+    if value < 35:
+        return "normal"
+    if value < 55:
+        return "low"
+    if value < 75:
+        return "medium"
+    return "high"
+
+
+def aggregate_metric(values: list[float]) -> tuple[float, str, float]:
+    avg = mean(values)
+    level = value_to_level(avg)
+    confidence = min(0.98, 0.72 + min(len(values), 24) * 0.01)
+    return round(avg, 1), level, round(confidence, 2)
+
+
+def build_screening_report(frames: list[FrameLandmarks]) -> dict:
+    head_vals: list[float] = []
+    shoulder_vals: list[float] = []
+    pelvic_vals: list[float] = []
+
+    for frame in frames:
+        head = score_head_forward(frame)
+        shoulder = score_shoulder_asymmetry(frame)
+        pelvic = score_pelvic_tilt(frame)
+        if head is not None:
+            head_vals.append(head)
+        if shoulder is not None:
+            shoulder_vals.append(shoulder)
+        if pelvic is not None:
+            pelvic_vals.append(pelvic)
+
+    if not head_vals and not shoulder_vals and not pelvic_vals:
+        raise ValueError("insufficient_pose_frames")
+
+    metrics = []
+    for name, values in zip(METRIC_NAMES, (head_vals, shoulder_vals, pelvic_vals), strict=True):
+        if not values:
+            continue
+        value, level, confidence = aggregate_metric(values)
+        metrics.append({"name": name, "value": value, "level": level, "confidence": confidence})
+
+    worst = max((m["value"] for m in metrics), default=0)
+    risk_level = value_to_level(worst)
+    if risk_level == "normal":
+        summary = "体态指标整体正常，建议保持日常活动与姿势习惯"
+    elif risk_level == "low":
+        summary = "存在轻度体态偏差，建议开始基础纠正训练"
+    elif risk_level == "medium":
+        summary = "建议关注头前伸与高低肩，开始针对性训练"
+    else:
+        summary = "多项指标偏高，建议尽快安排专业评估与干预"
+
+    recommendations = []
+    head_metric = next((m for m in metrics if m["name"] == "头前伸"), None)
+    shoulder_metric = next((m for m in metrics if m["name"] == "高低肩"), None)
+    if head_metric and head_metric["level"] in ("medium", "high"):
+        recommendations.append("每日肩胛稳定训练 5 分钟")
+        recommendations.append("颈后肌群拉伸 3 组")
+    if shoulder_metric and shoulder_metric["level"] in ("medium", "high"):
+        recommendations.append("对称性肩带激活训练 2 组")
+    if not recommendations:
+        recommendations.append("保持每日 20 分钟户外活动")
+    recommendations.append("28 天后建议复测对比")
+
+    return {
+        "riskLevel": risk_level,
+        "summary": summary,
+        "metrics": metrics,
+        "recommendations": recommendations,
+        "disclaimer": "本报告用于健康管理建议，不构成医疗诊断。",
+    }
+
+
+def build_movement_report(frames: list[FrameLandmarks]) -> dict:
+    if len(frames) < 3:
+        raise ValueError("insufficient_pose_frames")
+
+    nose_y = []
+    shoulder_angles = []
+    for frame in frames:
+        nose = _point(frame, NOSE)
+        left = _point(frame, LEFT_SHOULDER)
+        right = _point(frame, RIGHT_SHOULDER)
+        if nose:
+            nose_y.append(nose["y"])
+        if left and right:
+            shoulder_angles.append(
+                math.degrees(math.atan2(right["y"] - left["y"], right["x"] - left["x"]))
+            )
+
+    stability = 90.0
+    if len(nose_y) >= 3:
+        stability = max(55.0, 100.0 - pstdev(nose_y) * 900)
+
+    angle_score = 85.0
+    if len(shoulder_angles) >= 3:
+        angle_score = max(50.0, 100.0 - pstdev(shoulder_angles) * 2.5)
+
+    rhythm = min(100.0, 70 + len(frames) * 1.2)
+    trajectory = min(100.0, stability + 5)
+    completion = min(100.0, 60 + len(frames) * 2)
+    score = round((trajectory + angle_score + rhythm + stability + completion) / 5)
+
+    return {
+        "taskType": "movement_scoring",
+        "score": score,
+        "repsCompleted": max(1, len(frames) // 4),
+        "durationSeconds": len(frames) * 8,
+        "dimensions": {
+            "trajectory": round(trajectory),
+            "angle": round(angle_score),
+            "rhythm": round(rhythm),
+            "stability": round(stability),
+            "completion": round(completion),
+        },
+        "baselineCompare": {
+            "headNeckAngle": {"screening": 21, "current": max(8, 21 - score // 10), "delta": -(score // 10)},
+            "shoulderDiffMm": {"screening": 12, "current": max(3, 12 - score // 12), "delta": -(score // 12)},
+        },
+        "disclaimer": "本报告用于运动训练反馈，不构成医疗诊断。",
+    }