silver_pose/v1/fall_state.py

"""Deterministic per-person temporal fall event state machine."""

from dataclasses import dataclass
from enum import Enum
from typing import Dict, List, Optional


class FallState(str, Enum):
    NORMAL = "NORMAL"
    SUSPECT = "SUSPECT"
    CONFIRMED = "CONFIRMED"
    RECOVERING = "RECOVERING"


@dataclass(frozen=True)
class Evidence:
    accepted: bool
    is_fall_candidate: bool
    is_recovery_candidate: bool = False


@dataclass(frozen=True)
class FallEvent:
    event_id: str
    track_id: str
    config_version: str
    suspected_at_monotonic: float
    confirmed_at_monotonic: float
    latency_seconds: float
    state: FallState = FallState.CONFIRMED


@dataclass
class _Record:
    state: FallState
    suspect_started_at: Optional[float] = None
    confirmed_at: Optional[float] = None
    recovery_started_at: Optional[float] = None
    last_updated_at: Optional[float] = None


class FallStateMachine:
    """Confirm only uninterrupted, accepted fall evidence for one tracked person."""

    def __init__(
        self,
        confirm_window_seconds: float,
        recovery_window_seconds: float,
        config_version: str,
        cooldown_seconds: float = 0.0,
        session_id: str = "",
    ) -> None:
        if not 1.0 <= confirm_window_seconds <= 3.0:
            raise ValueError("confirm_window_seconds must be between 1 and 3 seconds")
        if recovery_window_seconds <= 0:
            raise ValueError("recovery_window_seconds must be positive")
        if cooldown_seconds < 0:
            raise ValueError("cooldown_seconds must be non-negative")
        if not isinstance(config_version, str) or not config_version.strip():
            raise ValueError("config_version must be a non-empty string")
        self._confirm_window_seconds = float(confirm_window_seconds)
        self._recovery_window_seconds = float(recovery_window_seconds)
        self._cooldown_seconds = float(cooldown_seconds)
        self._config_version = config_version.strip()
        self._session_id = str(session_id).strip()
        self._records: Dict[str, _Record] = {}
        self._next_event_number = 1

    def state_of(self, track_id: str) -> FallState:
        record = self._records.get(track_id)
        return record.state if record is not None else FallState.NORMAL

    def update(self, track_id: str, evidence: Evidence, now: float) -> List[FallEvent]:
        if not track_id:
            raise ValueError("track_id must be non-empty")
        timestamp = float(now)
        record = self._records.setdefault(track_id, _Record(state=FallState.NORMAL))
        if record.last_updated_at is not None and timestamp < record.last_updated_at:
            raise ValueError("timestamps must be monotonic per track")
        record.last_updated_at = timestamp

        if not evidence.accepted:
            self._reject_evidence(record)
            return []

        if record.state is FallState.NORMAL:
            if evidence.is_fall_candidate:
                record.state = FallState.SUSPECT
                record.suspect_started_at = timestamp
            return []

        if record.state is FallState.SUSPECT:
            if not evidence.is_fall_candidate:
                self._set_normal(record)
                return []
            if timestamp - record.suspect_started_at >= self._confirm_window_seconds:
                record.state = FallState.CONFIRMED
                record.confirmed_at = timestamp
                event = self._new_event(track_id, record.suspect_started_at, timestamp)
                record.suspect_started_at = None
                return [event]
            return []

        if record.state is FallState.CONFIRMED:
            if (
                evidence.is_recovery_candidate
                and timestamp - record.confirmed_at >= self._cooldown_seconds
            ):
                record.state = FallState.RECOVERING
                record.recovery_started_at = timestamp
            return []

        if record.state is FallState.RECOVERING:
            if evidence.is_fall_candidate:
                record.state = FallState.SUSPECT
                record.suspect_started_at = timestamp
                record.recovery_started_at = None
                return []
            if not evidence.is_recovery_candidate:
                record.state = FallState.CONFIRMED
                record.recovery_started_at = None
                return []
            if timestamp - record.recovery_started_at >= self._recovery_window_seconds:
                self._set_normal(record)
            return []

        raise RuntimeError("unknown fall state")

    def _reject_evidence(self, record: _Record) -> None:
        if record.state is FallState.SUSPECT:
            self._set_normal(record)
        elif record.state is FallState.RECOVERING:
            record.state = FallState.CONFIRMED
            record.recovery_started_at = None

    @staticmethod
    def _set_normal(record: _Record) -> None:
        record.state = FallState.NORMAL
        record.suspect_started_at = None
        record.confirmed_at = None
        record.recovery_started_at = None

    def _new_event(
        self, track_id: str, suspected_at: float, confirmed_at: float
    ) -> FallEvent:
        prefix = "FALL-{0}-".format(self._session_id) if self._session_id else "FALL-"
        event = FallEvent(
            event_id="{0}{1:06d}".format(prefix, self._next_event_number),
            track_id=track_id,
            config_version=self._config_version,
            suspected_at_monotonic=suspected_at,
            confirmed_at_monotonic=confirmed_at,
            latency_seconds=confirmed_at - suspected_at,
        )
        self._next_event_number += 1
        return event