silver_pose/v1/fall_policy.py

"""Translate per-track pose facts into state-machine evidence."""

from typing import Dict

from v1.evidence import PoseEvidence
from v1.fall_state import Evidence, FallState


class FallEvidencePolicy:
    """Require a recent rapid drop before a horizontal pose becomes a candidate."""

    def __init__(
        self, suspect_window_seconds: float, require_rapid_drop: bool = True
    ) -> None:
        self._suspect_window_seconds = float(suspect_window_seconds)
        self._require_rapid_drop = bool(require_rapid_drop)
        self._rapid_drop_at: Dict[str, float] = {}

    def evaluate(
        self,
        track_id: str,
        pose_evidence: PoseEvidence,
        now: float,
        state: FallState,
    ) -> Evidence:
        timestamp = float(now)
        if not pose_evidence.accepted:
            self._rapid_drop_at.pop(track_id, None)
            return Evidence(accepted=False, is_fall_candidate=False)
        if pose_evidence.rapid_vertical_change:
            self._rapid_drop_at[track_id] = timestamp

        candidate = False
        if state is FallState.SUSPECT:
            candidate = pose_evidence.horizontal_pose
        elif pose_evidence.horizontal_pose:
            if not self._require_rapid_drop:
                candidate = True
            elif track_id in self._rapid_drop_at:
                candidate = (
                    timestamp - self._rapid_drop_at[track_id] <= self._suspect_window_seconds
                )

        recovery = (
            state in (FallState.CONFIRMED, FallState.RECOVERING)
            and not pose_evidence.horizontal_pose
            and not pose_evidence.rapid_vertical_change
        )
        return Evidence(
            accepted=pose_evidence.accepted,
            is_fall_candidate=candidate,
            is_recovery_candidate=recovery,
        )