happy-up/apps/api/app/ai/pose_metrics.py

"""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]

from app.ai.calibration import CALIBRATION, 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 * CALIBRATION["head_forward_scale"]))


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 * CALIBRATION["shoulder_asymmetry_scale"]))


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 * CALIBRATION["pelvic_tilt_scale"]))


def value_to_level(value: float) -> str:
    if value < CALIBRATION["normal_max"]:
        return "normal"
    if value < CALIBRATION["low_max"]:
        return "low"
    if value < CALIBRATION["medium_max"]:
        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 worst >= CALIBRATION["review_required_min"]:
        risk_level = "review_required"
    if risk_level == "normal":
        summary = "体态指标整体正常，建议保持日常活动与姿势习惯"
    elif risk_level == "low":
        summary = "存在轻度体态偏差，建议开始基础纠正训练"
    elif risk_level == "medium":
        summary = "建议关注头前伸与高低肩，开始针对性训练"
    elif risk_level == "review_required":
        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": "本报告用于运动训练反馈，不构成医疗诊断。",
    }