docs: add fall detection harness foundation
This commit is contained in:
@@ -0,0 +1,518 @@
|
||||
# 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**
|
||||
|
||||
```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"
|
||||
```
|
||||
|
||||
- [ ] **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**
|
||||
|
||||
```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`.
|
||||
|
||||
- [ ] **Step 4: Run baseline tests and update the standard command**
|
||||
|
||||
Run: `python -m pytest v1/tests -v`
|
||||
Expected: PASS. Update `init.ps1` so it installs `v1/requirements.txt` and runs this command.
|
||||
|
||||
- [ ] **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.
|
||||
Reference in New Issue
Block a user