2026-07-20 22:05:55 +08:00
# 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–
**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`
2026-07-21 09:45:42 +08:00
- [x] **Step 1: Write the failing configuration test**
2026-07-20 22:05:55 +08:00
```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"
```
2026-07-21 09:45:42 +08:00
- [x] **Step 2: Run the test to verify it fails**
2026-07-20 22:05:55 +08:00
Run: `python -m pytest v1/tests/test_config.py -v`
Expected: FAIL because `v1.config` does not exist.
2026-07-21 09:45:42 +08:00
- [x] **Step 3: Implement the smallest configuration interface**
2026-07-20 22:05:55 +08:00
```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` .
2026-07-21 09:45:42 +08:00
- [x] **Step 4: Run baseline tests and update the standard command**
2026-07-20 22:05:55 +08:00
Run: `python -m pytest v1/tests -v`
2026-07-21 09:45:42 +08:00
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.
2026-07-20 22:05:55 +08:00
2026-07-21 09:45:42 +08:00
- [x] **Step 5: Commit**
2026-07-20 22:05:55 +08:00
```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`
2026-07-21 09:49:28 +08:00
- [x] **Step 1: Write replay and broken-source tests**
2026-07-20 22:05:55 +08:00
```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
```
2026-07-21 09:49:28 +08:00
- [x] **Step 2: Run tests to verify they fail**
2026-07-20 22:05:55 +08:00
Run: `python -m pytest v1/tests/test_video_source.py -v`
Expected: FAIL because `VideoSource` and `SourceStatus` do not exist.
2026-07-21 09:49:28 +08:00
- [x] **Step 3: Implement the source contract**
2026-07-20 22:05:55 +08:00
```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.
2026-07-21 09:49:28 +08:00
- [x] **Step 4: Run tests**
2026-07-20 22:05:55 +08:00
Run: `python -m pytest v1/tests/test_video_source.py -v`
Expected: PASS.
2026-07-21 09:49:28 +08:00
- [x] **Step 5: Commit**
2026-07-20 22:05:55 +08:00
```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–
- [ ] **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.
