silver_pose/v1/pipeline.py

"""Compose Pose, tracking, evidence policy, and temporal fall state."""

import itertools
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, Optional, Sequence, Tuple

from v1.config import AppConfig
from v1.evidence import PoseEvidence, assess_pose_quality, extract_evidence
from v1.fall_policy import FallEvidencePolicy
from v1.fall_state import Evidence, FallEvent, FallState, FallStateMachine
from v1.tracking import PersonTracker, TrackedPersonPose
from v1.video_source import FramePacket, SourceStatus


_SESSION_COUNTER = itertools.count(1)


def new_session_id() -> str:
    """Return a process-unique, human-readable run id for event traceability."""

    return "{0}-{1:03d}".format(
        datetime.now().strftime("%Y%m%d-%H%M%S"), next(_SESSION_COUNTER)
    )


@dataclass(frozen=True)
class PersonAnalysis:
    tracked_pose: TrackedPersonPose
    pose_evidence: PoseEvidence
    state: FallState
    evidence: Evidence


@dataclass(frozen=True)
class FrameAnalysis:
    packet: FramePacket
    people: Tuple[PersonAnalysis, ...]
    events: Tuple[FallEvent, ...]


class FallPipeline:
    """Run one source frame through the V1 event decision flow."""

    def __init__(
        self,
        pose_adapter,
        tracker: PersonTracker,
        policy: FallEvidencePolicy,
        state_machine: FallStateMachine,
        keypoint_confidence_threshold: float,
        require_lower_body: bool = True,
        horizontal_angle_threshold_degrees: float = 35.0,
    ) -> None:
        if not 0.0 <= keypoint_confidence_threshold <= 1.0:
            raise ValueError("keypoint_confidence_threshold must be between 0 and 1")
        self._pose_adapter = pose_adapter
        self._tracker = tracker
        self._policy = policy
        self._state_machine = state_machine
        self._keypoint_confidence_threshold = float(keypoint_confidence_threshold)
        self._require_lower_body = bool(require_lower_body)
        self._horizontal_angle_threshold_degrees = float(horizontal_angle_threshold_degrees)
        self._previous_evidence: Dict[str, PoseEvidence] = {}
        self._active_track_ids = set()

    @classmethod
    def from_config(
        cls, config: AppConfig, pose_adapter, session_id: Optional[str] = None
    ) -> "FallPipeline":
        """Create one immutable runtime decision flow from validated config.

        Each run gets a unique ``session_id`` so confirmed-event IDs never collide
        across monitoring restarts within the same day (no screenshot overwrite or
        duplicate JSONL identity).
        """

        return cls(
            pose_adapter=pose_adapter,
            tracker=PersonTracker(),
            policy=FallEvidencePolicy(
                config.event.suspect_window_seconds,
                require_rapid_drop=config.event.require_rapid_drop,
            ),
            state_machine=FallStateMachine(
                confirm_window_seconds=config.event.confirm_window_seconds,
                recovery_window_seconds=config.event.recovery_window_seconds,
                cooldown_seconds=config.event.cooldown_seconds,
                config_version=config.runtime_config_version,
                session_id=session_id or new_session_id(),
            ),
            keypoint_confidence_threshold=config.event.keypoint_confidence_threshold,
            require_lower_body=config.event.require_lower_body,
            horizontal_angle_threshold_degrees=config.event.horizontal_angle_threshold_degrees,
        )

    def process(self, packet: FramePacket) -> FrameAnalysis:
        if packet.status is not SourceStatus.CONNECTED or packet.image is None:
            events = self._reject_active_tracks(packet.timestamp_monotonic)
            return FrameAnalysis(packet=packet, people=(), events=tuple(events))

        height, width = packet.image.shape[:2]
        poses = self._pose_adapter.infer(packet.image)
        tracked_poses = self._tracker.update(
            poses,
            detected_at_monotonic=packet.timestamp_monotonic,
            frame_size=(width, height),
        )
        current_ids = {tracked.track_id for tracked in tracked_poses}
        events = self._reject_missing_tracks(current_ids, packet.timestamp_monotonic)
        people = []
        for tracked in tracked_poses:
            quality = assess_pose_quality(
                tracked.pose,
                threshold=self._keypoint_confidence_threshold,
                require_lower_body=self._require_lower_body,
            )
            pose_evidence = extract_evidence(
                tracked.pose,
                quality=quality,
                previous=self._previous_evidence.get(tracked.track_id),
                horizontal_angle_threshold_degrees=self._horizontal_angle_threshold_degrees,
            )
            state_before = self._state_machine.state_of(tracked.track_id)
            evidence = self._policy.evaluate(
                tracked.track_id,
                pose_evidence,
                now=packet.timestamp_monotonic,
                state=state_before,
            )
            events.extend(
                self._state_machine.update(
                    tracked.track_id, evidence, now=packet.timestamp_monotonic
                )
            )
            if pose_evidence.accepted:
                self._previous_evidence[tracked.track_id] = pose_evidence
            else:
                self._previous_evidence.pop(tracked.track_id, None)
            people.append(
                PersonAnalysis(
                    tracked_pose=tracked,
                    pose_evidence=pose_evidence,
                    state=self._state_machine.state_of(tracked.track_id),
                    evidence=evidence,
                )
            )
        self._active_track_ids = current_ids
        return FrameAnalysis(packet=packet, people=tuple(people), events=tuple(events))

    def _reject_missing_tracks(self, current_ids: set, now: float) -> list:
        missing_ids = self._active_track_ids - current_ids
        events = []
        for track_id in sorted(missing_ids):
            events.extend(self._reject_track(track_id, now))
        return events

    def _reject_active_tracks(self, now: float) -> list:
        events = []
        for track_id in sorted(self._active_track_ids):
            events.extend(self._reject_track(track_id, now))
        self._active_track_ids = set()
        return events

    def _reject_track(self, track_id: str, now: float) -> Sequence[FallEvent]:
        rejected = Evidence(accepted=False, is_fall_candidate=False)
        self._previous_evidence.pop(track_id, None)
        self._policy.evaluate(
            track_id,
            PoseEvidence(False, False, False, None, None, None, "missing_pose"),
            now=now,
            state=self._state_machine.state_of(track_id),
        )
        return self._state_machine.update(track_id, rejected, now=now)