519 lines
18 KiB
Markdown
519 lines
18 KiB
Markdown
# 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`
|
||
|
||
- [x] **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
|
||
```
|
||
|
||
- [x] **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.
|
||
|
||
- [x] **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.
|
||
|
||
- [x] **Step 4: Run tests**
|
||
|
||
Run: `python -m pytest v1/tests/test_video_source.py -v`
|
||
Expected: PASS.
|
||
|
||
- [x] **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`
|
||
|
||
- [x] **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
|
||
```
|
||
|
||
- [x] **Step 2: Run tests to verify failure**
|
||
|
||
Run: `python -m pytest v1/tests/test_pose.py -v`
|
||
Expected: FAIL because `PoseAdapter` is missing.
|
||
|
||
- [x] **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.
|
||
|
||
- [x] **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.
|
||
|
||
- [x] **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.
|