"""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 @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, ) -> 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._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), 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, ) 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 ) pose_evidence = extract_evidence( tracked.pose, quality=quality, previous=self._previous_evidence.get(tracked.track_id), ) 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), ) ) 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)