# 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` - [x] **Step 1: Write the failing configuration test** ```python 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" ``` - [x] **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. - [x] **Step 3: Implement the smallest configuration interface** ```python 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`. - [x] **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. - [x] **Step 5: Commit** ```powershell 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** ```python 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** ```python 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** ```powershell 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** ```python 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** ```python 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** ```powershell 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** ```python 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** ```python 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** ```powershell 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** ```python 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** ```python 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** ```powershell 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** ```python 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** ```python 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** ```powershell 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** ```json { "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** ```python 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** ```python 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** ```powershell 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.