"""
FormScout — Gradio app entrypoint.
Screening aid for Functional Movement Screen (FMS) scoring.
NOT a diagnosis. NOT an injury predictor.
Custom scout/trail themed UI with score dial, pipeline visualization,
rubric breakdown, and persistent safety banner.
"""
from __future__ import annotations
import os
import tempfile
import gradio as gr
from formscout.pipeline import Director
from formscout.rubric import score_test
from formscout.ui.theme import formscout_theme, FORMSCOUT_CSS
from formscout import config
from formscout import session as session_mod
from formscout.startup import ensure_checkpoints
# ─── ZeroGPU ──────────────────────────────────────────────────────────────────
# On an HF Spaces ZeroGPU runtime the heavy analysis MUST run inside an
# @spaces.GPU function, and that function must already exist at import time:
# ZeroGPU scans for one during startup and aborts the Space with
# "No @spaces.GPU function detected during startup" if none is registered.
# We decorate process_video (the Start Analysis handler) so a single GPU window
# covers the whole pipeline — pose, optional 3D, and the Qwen3-VL judge. Off a
# ZeroGPU Space the `spaces` package is absent (or its decorator is effect-free),
# so local runs and CPU Spaces are unaffected.
try:
import spaces
gpu_task = spaces.GPU(duration=config.ZEROGPU_DURATION)
except Exception: # local dev / non-ZeroGPU — decorate as a no-op
def gpu_task(fn):
return fn
ensure_checkpoints()
# ─── Constants ───────────────────────────────────────────────────────────────
DISCLAIMER = (
"⚠️ **Screening aid — not a diagnosis. "
"Pain or clearing tests require a clinician.**"
)
FMS_TESTS = [
("Deep Squat", "deep_squat"),
("Hurdle Step", "hurdle_step"),
("In-Line Lunge", "inline_lunge"),
("Shoulder Mobility", "shoulder_mobility"),
("Active Straight-Leg Raise", "active_slr"),
("Trunk Stability Push-Up", "trunk_stability_pushup"),
("Rotary Stability", "rotary_stability"),
]
SCORE_DESCRIPTIONS = {
3: "Movement performed to criterion — no compensation",
2: "Movement completed with compensation or regression",
1: "Unable to perform the movement pattern",
0: "Pain reported — clinician referral required",
}
# Per-test illustration shown as a movement guide (test_key -> filename).
_ASSET_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "assets", "screenings")
SCREENING_ASSETS = {
"deep_squat": "deep-squat.png",
"hurdle_step": "hurdle-step.png",
"inline_lunge": "inline-lunge.png",
"shoulder_mobility": "shoulder-mobility.png",
"active_slr": "active-straight-leg-raise.png",
"trunk_stability_pushup": "trunk-stability-push-up.png",
"rotary_stability": "rotary-stability.png",
}
def _screening_image(test_display_name: str) -> str | None:
"""Resolve the movement-guide image path for a dropdown display name."""
key = {name: val for name, val in FMS_TESTS}.get(test_display_name)
fname = SCREENING_ASSETS.get(key)
if not fname:
return None
path = os.path.join(_ASSET_DIR, fname)
return path if os.path.exists(path) else None
# ─── Processing ──────────────────────────────────────────────────────────────
@gpu_task
def process_video(video_path: str, test_name: str, side: str, model_key: str,
layers: list[str], session_state):
"""Analyse one clip and accumulate it into the screening session.
Decorated with @spaces.GPU on ZeroGPU: the whole pipeline (pose, optional 3D,
Qwen3-VL judge) runs inside one GPU window. The decorator is a no-op off-Space.
"""
if not video_path:
return (
session_state, _render_empty_state(), "Upload a video to begin analysis.",
"", "", None, "", _render_session_table(session_state),
gr.update(visible=False), gr.update(visible=False),
)
if session_state is None:
session_state = session_mod.new_session()
director = Director()
state = director.run(video_path, test_name=test_name, side=side, model_key=model_key)
score_html = _render_empty_state()
score_details = ""
if state.features:
result = score_test(state.features)
judge = state.judge
if judge and judge.score is not None:
score_html = _render_score_card(judge.score, judge.confidence, judge.needs_human)
score_details = _render_score_details_judge(judge, result, state.features)
elif judge and judge.needs_human:
score_html = _render_score_card(0, 0, True)
score_details = f"### Needs Clinician Review\n{judge.rationale}"
else:
score_html = _render_score_card(result.score, result.confidence, result.needs_human)
score_details = _render_score_details(result, state.features)
# Accumulate into the session (only when we have a real analysis)
if state.ingest and state.pose2d and state.judge:
draw_trails = "trails" in {lbl.lower().replace(" ", "_") for lbl in (layers or [])}
try:
session_mod.add_analysis(
session_state, ingest=state.ingest, pose2d=state.pose2d,
features=state.features, judge=state.judge,
test_name=test_name, side=side, draw_trails=draw_trails,
)
except Exception as e:
state.warnings.append(f"session accumulation failed: {e}")
pipeline_md = _render_pipeline_status(state)
alerts = _render_alerts(state)
overlay_path = None
vel_summary = ""
layer_set = {lbl.lower().replace(" ", "_") for lbl in (layers or [])}
if layer_set and state.ingest and state.pose2d:
try:
from formscout.agents.visualizer import PoseVisualizer, build_velocity_summary
vis = PoseVisualizer()
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
out_path = f.name
overlay_path = vis.render_video(state.ingest, state.pose2d, layer_set, out_path)
if overlay_path:
vel_summary = build_velocity_summary(state.pose2d.keypoints, vis.last_velocities)
except Exception as e:
alerts = (alerts or "") + f"\n⚠️ Visualizer error: {e}"
has_entries = bool(session_state and session_state.entries)
return (
session_state, score_html, pipeline_md, score_details, alerts,
overlay_path, vel_summary, _render_session_table(session_state),
gr.update(visible=has_entries), gr.update(visible=has_entries),
)
def _render_score_card(score: int, confidence: float, needs_human: bool) -> str:
"""Render the score dial as HTML."""
if needs_human:
return """
⚠️ Needs Clinician Review
Pain or clearing test detected — cannot auto-score
"""
def _render_empty_state() -> str:
"""Render placeholder when no video processed yet."""
return """
🏔️
Upload a video to begin
"""
def _render_score_details(result, features) -> str:
"""Render the rubric breakdown."""
parts = [f"### Rationale\n{result.rationale}\n"]
if features.angles:
parts.append("### Measurements")
for key, val in features.angles.items():
label = key.replace("_", " ").title()
parts.append(f"- **{label}:** {val:.1f}°")
if features.alignments:
parts.append("\n### Alignment Checks")
for key, val in features.alignments.items():
label = key.replace("_", " ").title()
icon = "✓" if val else "✗"
parts.append(f"- {icon} {label}")
if features.view == "2d":
parts.append(
"\n> ⚠️ *2D estimate — angles are camera-angle dependent. "
"For best accuracy, film from the side at hip height.*"
)
return "\n".join(parts)
def _render_score_details_judge(judge, rubric, features) -> str:
"""Render judge + rubric combined breakdown."""
parts = [f"### Judge Rationale\n{judge.rationale}\n"]
if judge.compensation_tags:
parts.append(f"**Compensations:** {', '.join(judge.compensation_tags)}")
if judge.corrective_hint:
parts.append(f"**Corrective:** {judge.corrective_hint}")
parts.append(f"\n### Rubric Score: {rubric.score}/3")
parts.append(f"*{rubric.rationale}*")
if features.angles:
parts.append("\n### Measurements")
for key, val in features.angles.items():
label = key.replace("_", " ").title()
parts.append(f"- **{label}:** {val:.1f}°" if isinstance(val, float) else f"- **{label}:** {val}")
if features.symmetry_delta is not None:
parts.append(f"\n### Asymmetry\n- **L/R Delta:** {features.symmetry_delta:.1f}°")
if features.view == "2d":
parts.append(
"\n> ⚠️ *2D estimate — angles are camera-angle dependent.*"
)
return "\n".join(parts)
def _render_pipeline_status(state) -> str:
"""Render pipeline step summary."""
parts = []
if state.ingest:
parts.append(
f"📹 **Ingest:** {len(state.ingest.frames)} frames · "
f"{state.ingest.fps:.0f}fps · {state.ingest.duration:.1f}s · "
f"{state.ingest.width}×{state.ingest.height}"
)
if state.pose2d:
n = sum(1 for kps in state.pose2d.keypoints if kps)
parts.append(
f"🦴 **Pose2D:** {n}/{len(state.pose2d.keypoints)} frames detected · "
f"conf={state.pose2d.confidence:.0%}"
)
if state.body3d:
if state.body3d.used:
parts.append(f"🧊 **Body3D:** active · conf={state.body3d.confidence:.0%}")
else:
parts.append("🧊 **Body3D:** 2D-only path (normal)")
if state.features:
parts.append(
f"📐 **Biomechanics:** view={state.features.view} · "
f"conf={state.features.confidence:.0%}"
)
return "\n\n".join(parts) if parts else "*Processing...*"
def _render_alerts(state) -> str:
"""Render errors and warnings."""
parts = []
if state.errors:
for e in state.errors:
parts.append(f"🚨 {e}")
if state.warnings:
for w in state.warnings:
parts.append(f"⚠️ {w}")
return "\n\n".join(parts)
def _render_session_table(session_state) -> str:
"""Render the accumulated 'Session so far' table as markdown."""
if not session_state or not session_state.entries:
return "*No clips analysed yet.*"
lines = ["| Test | Side | Score | Status |", "|---|---|---|---|"]
for e in session_state.entries:
test = e.test_name.replace("_", " ").title()
side = e.side if e.side in ("left", "right") else "—"
if e.needs_human:
score, status = "—", "⚠️ Clinician review"
else:
score, status = f"{e.score}/3", "✓ scored"
lines.append(f"| {test} | {side} | {score} | {status} |")
return "\n".join(lines)
def _finish_session(session_state):
"""Build the composite report + PDF for the whole session."""
if not session_state or not session_state.entries:
return ("⚠️ No clips analysed yet — analyse at least one clip first.",
None, None, None)
report, pdf_path = session_mod.finish_session(session_state)
if report is None:
return ("⚠️ Nothing to report.", None, None, None)
if report.composite is not None:
summary = [f"## Composite: {report.composite} / 21"]
else:
n = len(session_state.entries)
summary = [f"## Composite: Incomplete — {n}/7 tests scored",
"*(One or more tests need clinician review or were unscored.)*"]
if report.asymmetries:
summary.append("\n### Asymmetries")
for a in report.asymmetries:
test = a["test"].replace("_", " ").title()
summary.append(f"- **{test}:** L={a['left_score']} R={a['right_score']} (Δ {a['delta']})")
flags = list(report.low_confidence_flags) + list(report.disagreement_flags)
if flags:
summary.append("\n### Flags")
for fl in flags:
summary.append(f"- {fl}")
md_path = os.path.join(session_state.session_dir, "analysis.md")
md_out = md_path if os.path.exists(md_path) else None
return "\n".join(summary), pdf_path, md_out, report.scoresheet_path
# ─── App Builder ─────────────────────────────────────────────────────────────
def build_app() -> gr.Blocks:
"""Build the FormScout Gradio app with custom scout/trail theme."""
with gr.Blocks(title="FormScout — FMS Screening Aid") as app:
session_state = gr.State(None)
# Header
gr.HTML("""
🏔️ FormScout
Functional Movement Screen · Automated Scoring Aid