Files

T

ila 1f6bbd8f98 feat(v1): add temporal fall state machine

2026-07-21 09:59:53 +08:00

18 KiB

Raw Blame History

Silver Pose V1 Implementation Plan

For agentic workers: REQUIRED SKILL: Use subagent-driven-development (recommended) or executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build a Python desktop V1 that detects a confirmed sudden-fall event from one RTSP or replayed video source and produces red overlay, sound, popup, and screenshot evidence within the agreed 1–3 second target.

Architecture: V1 keeps video acquisition, Pose inference, tracking, pose-quality evidence, per-person temporal state, UI rendering, and alert side effects in separate modules. All decisions are driven by one validated configuration object; the UI receives state and events but never decides a fall.

Tech Stack: Python 3.8.10, PyQt5, OpenCV, Ultralytics 8.3.205, ByteTrack, pytest, JSON/JSONL, Windows winsound.

File structure

Path	Responsibility
`v1/config.py`	Typed configuration and environment-variable resolution.
`v1/video_source.py`	Replay/RTSP frame source, timestamps, connection state and reconnect policy.
`v1/pose.py`	YOLO Pose adapter and model fingerprint verification.
`v1/tracking.py`	Stable person IDs.
`v1/evidence.py`	Pose-quality gate and scale-independent fall evidence.
`v1/fall_state.py`	Deterministic per-ID state machine and `FallEvent`.
`v1/alerts.py`	Idempotent sound, popup request, screenshot and JSONL artifact work.
`v1/gui.py`	PyQt display, controls and event presentation.
`v1/tests/`	Pure logic and replay tests.
`v1/scripts/replay_cases.py`	Replay labelled videos and emit event report.

Task 1: T-101 Create the V1 package, secure configuration and test baseline

Files:

Create: v1/__init__.py
Create: v1/config.py
Create: v1/config.example.json
Create: v1/requirements.txt
Create: v1/tests/test_config.py
Modify: .gitignore
Modify: init.ps1, docs/03-tech-stack.md, docs/current-state.md
Step 1: Write the failing configuration test

from v1.config import load_config

def test_load_config_resolves_rtsp_environment_variable(tmp_path, monkeypatch):
    config_file = tmp_path / "config.json"
    config_file.write_text(
        '{"source":{"id":"cam","rtsp_url_env":"SILVER_POSE_RTSP_URL"},'
        '"model":{"path":"models/best.pt","sha256":"abc","confidence_threshold":0.25},'
        '"event":{"keypoint_confidence_threshold":0.4,"suspect_window_seconds":0.5,'
        '"confirm_window_seconds":1.0,"recovery_window_seconds":2.0,"cooldown_seconds":10.0},'
        '"artifacts":{"event_dir":"artifacts"}}',
        encoding="utf-8",
    )
    monkeypatch.setenv("SILVER_POSE_RTSP_URL", "rtsp://example")
    assert load_config(config_file).source_url == "rtsp://example"

Step 2: Run the test to verify it fails

Run: python -m pytest v1/tests/test_config.py -v Expected: FAIL because v1.config does not exist.

Step 3: Implement the smallest configuration interface

import json
import os
from dataclasses import dataclass
from pathlib import Path

class ConfigError(ValueError):
    pass

@dataclass(frozen=True)
class EventConfig:
    keypoint_confidence_threshold: float
    suspect_window_seconds: float
    confirm_window_seconds: float
    recovery_window_seconds: float
    cooldown_seconds: float

@dataclass(frozen=True)
class AppConfig:
    source_id: str
    source_url: str
    model_path: Path
    model_sha256: str
    confidence_threshold: float
    event: EventConfig
    event_dir: Path

def load_config(path: Path) -> AppConfig:
    raw = json.loads(path.read_text(encoding="utf-8"))
    url = os.environ.get(raw["source"]["rtsp_url_env"])
    if not url:
        raise ConfigError("missing RTSP environment variable")
    event_raw = raw["event"]
    return AppConfig(
        source_id=raw["source"]["id"],
        source_url=url,
        model_path=Path(raw["model"]["path"]),
        model_sha256=raw["model"]["sha256"],
        confidence_threshold=float(raw["model"]["confidence_threshold"]),
        event=EventConfig(
            keypoint_confidence_threshold=float(event_raw["keypoint_confidence_threshold"]),
            suspect_window_seconds=float(event_raw["suspect_window_seconds"]),
            confirm_window_seconds=float(event_raw["confirm_window_seconds"]),
            recovery_window_seconds=float(event_raw["recovery_window_seconds"]),
            cooldown_seconds=float(event_raw["cooldown_seconds"]),
        ),
        event_dir=Path(raw["artifacts"]["event_dir"]),
    )

config.local.json, artifacts/, testdata/private/, *.onnx and camera credentials must be ignored; config.example.json contains only rtsp_url_env.

Step 4: Run baseline tests and update the standard command

Run: python -m pytest v1/tests -v Expected: PASS. Update init.ps1 so it checks the required runtime imports and runs this command; keep v1/requirements.txt as an explicit installation command rather than mutating the environment during verification.

Step 5: Commit

git add .gitignore init.ps1 v1 docs/03-tech-stack.md docs/current-state.md docs/06-tasks.md progress.md
git commit -m "feat(v1): add secure configuration baseline"

Task 2: T-102 Implement replayable frame acquisition before RTSP

Files:

Create: v1/video_source.py
Create: v1/tests/test_video_source.py
Step 1: Write replay and broken-source tests

def test_file_source_emits_monotonic_timestamps(sample_video):
    source = VideoSource(sample_video, reconnect=False)
    first = source.read()
    second = source.read()
    assert first.timestamp_monotonic < second.timestamp_monotonic

def test_missing_source_returns_error_state(tmp_path):
    source = VideoSource(tmp_path / "missing.mp4", reconnect=False)
    assert source.read().status == SourceStatus.ERROR

Step 2: Run tests to verify they fail

Run: python -m pytest v1/tests/test_video_source.py -v Expected: FAIL because VideoSource and SourceStatus do not exist.

Step 3: Implement the source contract

from typing import Optional

class SourceStatus(str, Enum):
    CONNECTED = "connected"
    RETRYING = "retrying"
    ERROR = "error"
    EOF = "eof"

@dataclass(frozen=True)
class FramePacket:
    image: Optional[np.ndarray]
    timestamp_monotonic: float
    status: SourceStatus
    error: Optional[str] = None

read() returns an error packet for failures; it never emits a synthetic person or fall event. For RTSP, reconnect with bounded backoff from configuration.

Step 4: Run tests

Run: python -m pytest v1/tests/test_video_source.py -v Expected: PASS.

Step 5: Commit

git add v1/video_source.py v1/tests/test_video_source.py docs progress.md
git commit -m "feat(v1): add replayable video source"

Task 3: T-103 Add a verified Pose adapter

Files:

Create: v1/pose.py
Create: v1/tests/test_pose.py
Step 1: Write adapter shape tests

def test_pose_adapter_rejects_non_pose_model(tmp_path):
    with pytest.raises(ModelValidationError):
        PoseAdapter(tmp_path / "not-a-pose-model.pt", expected_sha256="abc")

def test_person_pose_has_seventeen_keypoints(fake_yolo_result):
    poses = PoseAdapter.from_results(fake_yolo_result)
    assert len(poses[0].keypoints) == 17

Step 2: Run tests to verify failure

Run: python -m pytest v1/tests/test_pose.py -v Expected: FAIL because PoseAdapter is missing.

Step 3: Implement only the adapter contract

from typing import Sequence, Tuple

@dataclass(frozen=True)
class PersonPose:
    box_xyxy: Tuple[float, float, float, float]
    box_confidence: float
    keypoints: Sequence[Keypoint]

class PoseAdapter:
    def infer(self, image: np.ndarray) -> Sequence[PersonPose]:
        results = self._model(image, conf=self._confidence, verbose=False)
        return self.from_results(results[0])

At construction, hash the model, require task pose, class person, and exactly 17 three-value keypoints.

Step 4: Run tests and a real model smoke

Run: python -m pytest v1/tests/test_pose.py -v; python -c "from ultralytics import YOLO; assert YOLO('demo/best.pt').task == 'pose'" Expected: PASS and no assertion error.

Step 5: Commit

git add v1/pose.py v1/tests/test_pose.py docs progress.md
git commit -m "feat(v1): add verified pose adapter"

Task 4: T-104 Build quality gating and tracking

Files:

Create: v1/evidence.py
Create: v1/tracking.py
Create: v1/tests/test_evidence.py
Step 1: Write quality tests

def test_missing_ankles_rejects_pose(person_pose_without_ankles):
    quality = assess_pose_quality(person_pose_without_ankles, threshold=0.4)
    assert quality.accepted is False
    assert quality.reason == "required_joint_low_confidence"

def test_horizontal_body_is_evidence_not_event(horizontal_pose):
    evidence = extract_evidence(horizontal_pose, previous=None)
    assert evidence.horizontal_pose is True

Step 2: Run tests to verify failure

Run: python -m pytest v1/tests/test_evidence.py -v Expected: FAIL because quality and evidence functions are missing.

Step 3: Implement floating-point evidence

def assess_pose_quality(pose: PersonPose, threshold: float) -> PoseQuality:
    required = (5, 6, 11, 12, 13, 14, 15, 16)
    if any(pose.keypoints[index].confidence < threshold for index in required):
        return PoseQuality(False, "required_joint_low_confidence", 0)
    return PoseQuality(True, "accepted", len(required))

Use float coordinates, clamped cosine inputs, torso-normalized vertical motion, and no boolean alarm result in this module.

Step 4: Run tests

Run: python -m pytest v1/tests/test_evidence.py -v Expected: PASS.

Step 5: Commit

git add v1/evidence.py v1/tracking.py v1/tests/test_evidence.py docs progress.md
git commit -m "feat(v1): add pose quality evidence"

Task 5: T-105 Implement the deterministic per-person fall state machine

Files:

Create: v1/fall_state.py
Create: v1/tests/test_fall_state.py
Step 1: Write event timing tests

def test_confirmed_event_is_emitted_once_after_persistence():
    machine = FallStateMachine(confirm_window_seconds=1.0, recovery_window_seconds=2.0)
    assert machine.update("7", Evidence(True, True), now=0.0) == []
    events = machine.update("7", Evidence(True, True), now=1.1)
    assert len(events) == 1
    assert events[0].track_id == "7"
    assert machine.update("7", Evidence(True, True), now=1.2) == []

def test_brief_bend_returns_to_normal_without_event():
    machine = FallStateMachine(confirm_window_seconds=1.0, recovery_window_seconds=2.0)
    machine.update("7", Evidence(True, False), now=0.0)
    assert machine.update("7", Evidence(False, False), now=0.3) == []

def test_confirmed_person_recovers_before_new_event_is_allowed():
    machine = FallStateMachine(confirm_window_seconds=1.0, recovery_window_seconds=2.0)
    machine.update("7", Evidence(True, True), now=0.0)
    machine.update("7", Evidence(True, True), now=1.1)
    machine.update("7", Evidence(True, False, True), now=1.2)
    machine.update("7", Evidence(True, False, True), now=3.3)
    assert machine.state_of("7") is FallState.NORMAL

Step 2: Run tests to verify failure

Run: python -m pytest v1/tests/test_fall_state.py -v Expected: FAIL because FallStateMachine is missing.

Step 3: Implement the four-state contract

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
class _Record:
    state: FallState
    suspect_started_at: Optional[float] = None
    recovery_started_at: Optional[float] = None

@dataclass(frozen=True)
class FallEvent:
    event_id: str
    track_id: str
    confirmed_at: float

class FallStateMachine:
    def __init__(self, confirm_window_seconds: float, recovery_window_seconds: float):
        self._confirm_window_seconds = confirm_window_seconds
        self._recovery_window_seconds = recovery_window_seconds
        self._records: Dict[str, _Record] = {}
        self._next_event_number = 1

    def _new_event(self, track_id: str, now: float) -> FallEvent:
        event = FallEvent(f"fall-{self._next_event_number:06d}", track_id, now)
        self._next_event_number += 1
        return event

    def state_of(self, track_id: str) -> FallState:
        return self._records[track_id].state

    def update(self, track_id: str, evidence: Evidence, now: float) -> List[FallEvent]:
        record = self._records.setdefault(track_id, _Record(state=FallState.NORMAL))
        if not evidence.accepted:
            return []
        if record.state is FallState.NORMAL and evidence.is_fall_candidate:
            record.state, record.suspect_started_at = FallState.SUSPECT, now
            return []
        if record.state is FallState.SUSPECT and not evidence.is_fall_candidate:
            record.state, record.suspect_started_at = FallState.NORMAL, None
            return []
        if record.state is FallState.SUSPECT and now - record.suspect_started_at >= self._confirm_window_seconds:
            record.state = FallState.CONFIRMED
            return [self._new_event(track_id, now)]
        if record.state is FallState.CONFIRMED and evidence.is_recovery_candidate:
            record.state, record.recovery_started_at = FallState.RECOVERING, now
            return []
        if record.state is FallState.RECOVERING and evidence.is_fall_candidate:
            record.state, record.recovery_started_at = FallState.CONFIRMED, None
            return []
        if record.state is FallState.RECOVERING and not evidence.is_recovery_candidate:
            record.state, record.recovery_started_at = FallState.CONFIRMED, None
            return []
        if record.state is FallState.RECOVERING and now - record.recovery_started_at >= self._recovery_window_seconds:
            record.state, record.recovery_started_at = FallState.NORMAL, None
            return []
        return []

Use per-ID state, not global frame_has_fall. A source error and a rejected pose must not advance the machine.

Step 4: Run tests

Run: python -m pytest v1/tests/test_fall_state.py -v Expected: PASS.

Step 5: Commit

git add v1/fall_state.py v1/tests/test_fall_state.py docs progress.md
git commit -m "feat(v1): add temporal fall state machine"

Task 6: T-201 and T-202 Connect the event engine to PyQt and local artifacts

Files:

Create: v1/gui.py, v1/alerts.py, v1/app.py
Create: v1/tests/test_alerts.py
Step 1: Write idempotent artifact test

def test_alert_manager_writes_one_screenshot_and_one_log_record(tmp_path, fall_event, frame):
    manager = AlertManager(tmp_path)
    first = manager.handle(fall_event, frame)
    second = manager.handle(fall_event, frame)
    assert first.screenshot_path.exists()
    assert second.created is False
    assert len((tmp_path / "events.jsonl").read_text().splitlines()) == 1

Step 2: Run test to verify failure

Run: python -m pytest v1/tests/test_alerts.py -v Expected: FAIL because AlertManager is missing.

Step 3: Implement UI/event separation

class AlertManager:
    def handle(self, event: FallEvent, annotated_frame: np.ndarray) -> AlertResult:
        if event.event_id in self._handled:
            return AlertResult(created=False, screenshot_path=None)
        self._handled.add(event.event_id)
        # save screenshot, append JSONL, then request sound and popup
        return AlertResult(created=True, screenshot_path=path)

gui.py receives display frames and events through Qt signals. It renders state colors and popup requests but does not invoke inference or state transitions.

Step 4: Run test and manual GUI smoke

Run: python -m pytest v1/tests/test_alerts.py -v; python -m v1.app --config v1/config.example.json --source testdata/videos/smoke.mp4 Expected: pytest PASS; GUI shows source state and exits cleanly after replay.

Step 5: Commit

git add v1/gui.py v1/alerts.py v1/app.py v1/tests/test_alerts.py docs progress.md
git commit -m "feat(v1): add alerting desktop flow"

Task 7: T-203 through T-205 Establish event-level acceptance and V1 handoff

Files:

Create: v1/scripts/replay_cases.py
Create: testdata/expected_events.json
Modify: docs/02-requirements.md, docs/current-state.md, docs/06-tasks.md
Step 1: Create explicit expected cases

{
  "cases": [
    {"video":"fall-01.mp4","expected_event_count":1,"max_latency_seconds":3.0},
    {"video":"sit-01.mp4","expected_event_count":0,"max_latency_seconds":0.0},
    {"video":"bend-01.mp4","expected_event_count":0,"max_latency_seconds":0.0}
  ]
}

Step 2: Write the failing replay assertion

def test_replay_report_matches_expected_cases():
    report = replay_cases(Path("testdata/expected_events.json"))
    assert report.failed_cases == []

Step 3: Implement replay report generation

from typing import Optional

@dataclass(frozen=True)
class CaseResult:
    video: str
    expected_event_count: int
    actual_event_count: int
    max_latency_seconds: float
    actual_latency_seconds: Optional[float]

The command must return nonzero if an event count differs or a positive case exceeds 3.0 seconds.

Step 4: Run V1 acceptance

Run: python -m pytest v1/tests -v; python v1/scripts/replay_cases.py --manifest testdata/expected_events.json Expected: all tests PASS; every expected positive has one event in 1–3 seconds; every listed negative has zero events.

Step 5: Commit and mark V1 handoff

git add v1 testdata docs progress.md
git commit -m "test(v1): add fall event acceptance replay"

Only after this step, create the T-301 ONNX export task; do not start Go feature code earlier.

18 KiB Raw Blame History Unescape Escape

Silver Pose V1 Implementation Plan

File structure

Task 1: T-101 Create the V1 package, secure configuration and test baseline

Task 2: T-102 Implement replayable frame acquisition before RTSP

Task 3: T-103 Add a verified Pose adapter

Task 4: T-104 Build quality gating and tracking

Task 5: T-105 Implement the deterministic per-person fall state machine

Task 6: T-201 and T-202 Connect the event engine to PyQt and local artifacts

Task 7: T-203 through T-205 Establish event-level acceptance and V1 handoff

18 KiB

Raw Blame History