Spaces:

build-small-hackathon
/

small-functional-movement-screening

Paused

small-functional-movement-screening / formscout /types.py

feat: ZeroGPU startup fix + FMS scoring sheet download + judge prompt alignment

3732b46 verified 4 months ago

6.39 kB

"""
FormScout typed agent contracts.
Every agent accepts and returns frozen dataclasses defined here.
Validate at every boundary — never accept raw dicts across agent boundaries.
"""
from __future__ import annotations

from dataclasses import dataclass, field


@dataclass(frozen=True)
class IngestResult:
    """Output of IngestAgent — decoded video frames + metadata."""
    frames: list  # list of np.ndarray HWC BGR
    fps: float
    duration: float
    n_people: int
    width: int
    height: int
    confidence: float = 1.0
    notes: str = ""


@dataclass(frozen=True)
class SegmentResult:
    """Output of SegmentationAgent — per-frame athlete masks."""
    athlete_track_id: int
    masks: list  # list of np.ndarray bool HW per frame
    confidence: float = 1.0
    notes: str = ""


@dataclass(frozen=True)
class Pose2DResult:
    """Output of Pose2DAgent — per-frame 2D keypoints (COCO 17-joint)."""
    keypoints: list  # list[dict[int, dict]] frame→joint→{x,y,conf}
    fps: float
    confidence: float = 0.0
    notes: str = ""


@dataclass(frozen=True)
class Body3DResult:
    """Output of Body3DAgent — optional 3D joint positions."""
    used: bool
    joints_3d: list  # list[dict] frame→joint→{x,y,z} — empty if used=False
    confidence: float = 0.0
    notes: str = ""


@dataclass(frozen=True)
class MovementResult:
    """Output of MovementClassifierAgent — which FMS test is being performed."""
    test_name: str  # "deep_squat"|"hurdle_step"|...|"unknown"
    side: str       # "left"|"right"|"na"
    confidence: float = 0.0
    notes: str = ""

    def __post_init__(self):
        valid_tests = {
            "deep_squat", "hurdle_step", "inline_lunge",
            "shoulder_mobility", "active_slr",
            "trunk_stability_pushup", "rotary_stability", "unknown",
        }
        if self.test_name not in valid_tests:
            raise ValueError(f"test_name must be one of {valid_tests}, got '{self.test_name}'")
        valid_sides = {"left", "right", "na"}
        if self.side not in valid_sides:
            raise ValueError(f"side must be one of {valid_sides}, got '{self.side}'")


@dataclass(frozen=True)
class BiomechFeatures:
    """Output of BiomechanicsAgent — measured angles, alignments, timing."""
    test_name: str
    view: str           # "2d" | "3d"
    side: str           # "left"|"right"|"na"
    angles: dict        # named angle → degrees
    alignments: dict    # named alignment → value
    symmetry_delta: float | None  # |left - right| or None for non-bilateral
    timing: dict        # event name → frame index
    confidence: float = 0.0
    notes: str = ""

    def __post_init__(self):
        if self.view not in ("2d", "3d"):
            raise ValueError(f"view must be '2d' or '3d', got '{self.view}'")


@dataclass(frozen=True)
class ScoreResult:
    """Output of ScoringAgent (ST-GCN head) — provisional numeric score."""
    score: int  # 0–3
    rationale: str
    confidence: float
    needs_human: bool = False
    notes: str = ""

    def __post_init__(self):
        if not self.needs_human and not (0 <= self.score <= 3):
            raise ValueError(f"score must be 0–3, got {self.score}")


@dataclass(frozen=True)
class RetrievalResult:
    """Output of RetrievalAgent — similar scored exemplars from the index."""
    exemplars: list  # list of {clip_id, score, similarity, rationale}
    confidence: float = 1.0
    notes: str = ""


@dataclass(frozen=True)
class JudgeResult:
    """Output of JudgeAgent — final VLM-scored result with rationale."""
    score: int | None   # 0–3 or None if needs_human=True
    rationale: str
    compensation_tags: list
    corrective_hint: str
    confidence: float
    needs_human: bool = False
    notes: str = ""

    def __post_init__(self):
        if not self.needs_human and self.score is not None:
            if not (0 <= self.score <= 3):
                raise ValueError(f"score must be 0–3 when needs_human=False, got {self.score}")
        if self.needs_human and self.score is not None:
            raise ValueError("score must be None when needs_human=True")


@dataclass(frozen=True)
class ReportResult:
    """Output of ReportAgent — assembled scorecard."""
    per_test: list  # list of dicts with test_name, score, judge_result, features
    composite: int | None  # None if any test unscored
    asymmetries: list  # list of {test, left_score, right_score, delta}
    overlay_video_path: str | None
    pdf_path: str | None
    low_confidence_flags: list
    disagreement_flags: list
    scoresheet_path: str | None = None  # one-page FMS scoring sheet (separate download)
    notes: str = ""


@dataclass(frozen=True)
class SessionEntry:
    """One accumulated analysis in a screening session.

    Display fields (test_name…keyframe_path) feed the PDF/JSON/MD artifacts;
    the trailing typed objects (movement…judge) feed ReportAgent.run().
    """
    test_name: str
    side: str
    score: int | None
    needs_human: bool
    rationale: str
    compensation_tags: list
    corrective_hint: str
    measurements: dict
    confidence: float
    view: str
    keyframe_path: str | None
    movement: MovementResult
    features: BiomechFeatures
    rubric_score: ScoreResult
    judge: JudgeResult | None
    laban: dict | None = None          # Laban Effort factors + labels + body emphasis
    flexion: dict | None = None        # relevant joint angles at key frame: {name: {deg, openness}}
    chart_paths: dict | None = None    # {"angle"|"velocity"|"radar"|"flexion": png path}


@dataclass
class PipelineState:
    """Mutable state threaded through the Director."""
    video_path: str
    ingest: IngestResult | None = None
    segment: SegmentResult | None = None
    pose2d: Pose2DResult | None = None
    body3d: Body3DResult | None = None
    movement: MovementResult | None = None
    features: BiomechFeatures | None = None
    stgcn_score: ScoreResult | None = None
    retrieval: RetrievalResult | None = None
    judge: JudgeResult | None = None
    report: ReportResult | None = None
    errors: list = field(default_factory=list)
    warnings: list = field(default_factory=list)

	"""
	FormScout typed agent contracts.
	Every agent accepts and returns frozen dataclasses defined here.
	Validate at every boundary — never accept raw dicts across agent boundaries.
	"""
	from __future__ import annotations

	from dataclasses import dataclass, field


	@dataclass(frozen=True)
	class IngestResult:
	"""Output of IngestAgent — decoded video frames + metadata."""
	frames: list # list of np.ndarray HWC BGR
	fps: float
	duration: float
	n_people: int
	width: int
	height: int
	confidence: float = 1.0
	notes: str = ""


	@dataclass(frozen=True)
	class SegmentResult:
	"""Output of SegmentationAgent — per-frame athlete masks."""
	athlete_track_id: int
	masks: list # list of np.ndarray bool HW per frame
	confidence: float = 1.0
	notes: str = ""


	@dataclass(frozen=True)
	class Pose2DResult:
	"""Output of Pose2DAgent — per-frame 2D keypoints (COCO 17-joint)."""
	keypoints: list # list[dict[int, dict]] frame→joint→{x,y,conf}
	fps: float
	confidence: float = 0.0
	notes: str = ""


	@dataclass(frozen=True)
	class Body3DResult:
	"""Output of Body3DAgent — optional 3D joint positions."""
	used: bool
	joints_3d: list # list[dict] frame→joint→{x,y,z} — empty if used=False
	confidence: float = 0.0
	notes: str = ""


	@dataclass(frozen=True)
	class MovementResult:
	"""Output of MovementClassifierAgent — which FMS test is being performed."""
	test_name: str # "deep_squat"\|"hurdle_step"\|...\|"unknown"
	side: str # "left"\|"right"\|"na"
	confidence: float = 0.0
	notes: str = ""

	def __post_init__(self):
	valid_tests = {
	"deep_squat", "hurdle_step", "inline_lunge",
	"shoulder_mobility", "active_slr",
	"trunk_stability_pushup", "rotary_stability", "unknown",
	}
	if self.test_name not in valid_tests:
	raise ValueError(f"test_name must be one of {valid_tests}, got '{self.test_name}'")
	valid_sides = {"left", "right", "na"}
	if self.side not in valid_sides:
	raise ValueError(f"side must be one of {valid_sides}, got '{self.side}'")


	@dataclass(frozen=True)
	class BiomechFeatures:
	"""Output of BiomechanicsAgent — measured angles, alignments, timing."""
	test_name: str
	view: str # "2d" \| "3d"
	side: str # "left"\|"right"\|"na"
	angles: dict # named angle → degrees
	alignments: dict # named alignment → value
	symmetry_delta: float \| None # \|left - right\| or None for non-bilateral
	timing: dict # event name → frame index
	confidence: float = 0.0
	notes: str = ""

	def __post_init__(self):
	if self.view not in ("2d", "3d"):
	raise ValueError(f"view must be '2d' or '3d', got '{self.view}'")


	@dataclass(frozen=True)
	class ScoreResult:
	"""Output of ScoringAgent (ST-GCN head) — provisional numeric score."""
	score: int # 0–3
	rationale: str
	confidence: float
	needs_human: bool = False
	notes: str = ""

	def __post_init__(self):
	if not self.needs_human and not (0 <= self.score <= 3):
	raise ValueError(f"score must be 0–3, got {self.score}")


	@dataclass(frozen=True)
	class RetrievalResult:
	"""Output of RetrievalAgent — similar scored exemplars from the index."""
	exemplars: list # list of {clip_id, score, similarity, rationale}
	confidence: float = 1.0
	notes: str = ""


	@dataclass(frozen=True)
	class JudgeResult:
	"""Output of JudgeAgent — final VLM-scored result with rationale."""
	score: int \| None # 0–3 or None if needs_human=True
	rationale: str
	compensation_tags: list
	corrective_hint: str
	confidence: float
	needs_human: bool = False
	notes: str = ""

	def __post_init__(self):
	if not self.needs_human and self.score is not None:
	if not (0 <= self.score <= 3):
	raise ValueError(f"score must be 0–3 when needs_human=False, got {self.score}")
	if self.needs_human and self.score is not None:
	raise ValueError("score must be None when needs_human=True")


	@dataclass(frozen=True)
	class ReportResult:
	"""Output of ReportAgent — assembled scorecard."""
	per_test: list # list of dicts with test_name, score, judge_result, features
	composite: int \| None # None if any test unscored
	asymmetries: list # list of {test, left_score, right_score, delta}
	overlay_video_path: str \| None
	pdf_path: str \| None
	low_confidence_flags: list
	disagreement_flags: list
	scoresheet_path: str \| None = None # one-page FMS scoring sheet (separate download)
	notes: str = ""


	@dataclass(frozen=True)
	class SessionEntry:
	"""One accumulated analysis in a screening session.

	Display fields (test_name…keyframe_path) feed the PDF/JSON/MD artifacts;
	the trailing typed objects (movement…judge) feed ReportAgent.run().
	"""
	test_name: str
	side: str
	score: int \| None
	needs_human: bool
	rationale: str
	compensation_tags: list
	corrective_hint: str
	measurements: dict
	confidence: float
	view: str
	keyframe_path: str \| None
	movement: MovementResult
	features: BiomechFeatures
	rubric_score: ScoreResult
	judge: JudgeResult \| None
	laban: dict \| None = None # Laban Effort factors + labels + body emphasis
	flexion: dict \| None = None # relevant joint angles at key frame: {name: {deg, openness}}
	chart_paths: dict \| None = None # {"angle"\|"velocity"\|"radar"\|"flexion": png path}


	@dataclass
	class PipelineState:
	"""Mutable state threaded through the Director."""
	video_path: str
	ingest: IngestResult \| None = None
	segment: SegmentResult \| None = None
	pose2d: Pose2DResult \| None = None
	body3d: Body3DResult \| None = None
	movement: MovementResult \| None = None
	features: BiomechFeatures \| None = None
	stgcn_score: ScoreResult \| None = None
	retrieval: RetrievalResult \| None = None
	judge: JudgeResult \| None = None
	report: ReportResult \| None = None
	errors: list = field(default_factory=list)
	warnings: list = field(default_factory=list)