ilaandClaude Opus 4.8
04422d9ca0
A: PoseAdapter.set_confidence_threshold is applied on start, so the
settings model-confidence field actually affects inference.
B: config source.mode ('stream'|'replay') is explicit; app no longer
guesses the source type from the URL prefix.
C: FallStateMachine takes a session_id and from_config generates a
unique one per run, so event ids never collide across restarts
(no screenshot overwrite or duplicate JSONL identity in a day).
51 tests pass.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
import numpy as np
|
|
from pathlib import Path
|
|
|
|
from v1.config import AppConfig, EventConfig
|
|
from v1.fall_policy import FallEvidencePolicy
|
|
from v1.fall_state import FallState, FallStateMachine
|
|
from v1.pipeline import FallPipeline
|
|
from v1.pose import Keypoint, PersonPose
|
|
from v1.tracking import PersonTracker
|
|
from v1.video_source import FramePacket, SourceStatus
|
|
|
|
|
|
class _SequencePoseAdapter:
|
|
def __init__(self, frames):
|
|
self._frames = iter(frames)
|
|
|
|
def infer(self, _image):
|
|
return next(self._frames)
|
|
|
|
|
|
def _pose(horizontal=False):
|
|
points = [Keypoint(float(index), float(index), 0.9) for index in range(17)]
|
|
if horizontal:
|
|
points[5] = Keypoint(20.0, 80.0, 0.9)
|
|
points[6] = Keypoint(30.0, 80.0, 0.9)
|
|
points[11] = Keypoint(70.0, 100.0, 0.9)
|
|
points[12] = Keypoint(80.0, 100.0, 0.9)
|
|
else:
|
|
points[5] = Keypoint(30.0, 10.0, 0.9)
|
|
points[6] = Keypoint(40.0, 10.0, 0.9)
|
|
points[11] = Keypoint(30.0, 30.0, 0.9)
|
|
points[12] = Keypoint(40.0, 30.0, 0.9)
|
|
return PersonPose(
|
|
box_xyxy=(20.0, 20.0, 160.0, 160.0),
|
|
box_confidence=0.9,
|
|
keypoints=tuple(points),
|
|
)
|
|
|
|
|
|
def _packet(timestamp):
|
|
return FramePacket(
|
|
image=np.zeros((180, 180, 3), dtype=np.uint8),
|
|
timestamp_monotonic=timestamp,
|
|
status=SourceStatus.CONNECTED,
|
|
)
|
|
|
|
|
|
def _pipeline(frames):
|
|
machine = FallStateMachine(
|
|
confirm_window_seconds=1.0,
|
|
recovery_window_seconds=2.0,
|
|
config_version="cfg-test-pipeline",
|
|
)
|
|
return (
|
|
FallPipeline(
|
|
pose_adapter=_SequencePoseAdapter(frames),
|
|
tracker=PersonTracker(),
|
|
policy=FallEvidencePolicy(suspect_window_seconds=0.5),
|
|
state_machine=machine,
|
|
keypoint_confidence_threshold=0.4,
|
|
),
|
|
machine,
|
|
)
|
|
|
|
|
|
def _config():
|
|
return AppConfig(
|
|
source_id="lobby-camera-01",
|
|
source_url="rtsp://not-written-to-disk/live",
|
|
model_path=Path("models/best.pt"),
|
|
model_sha256="a" * 64,
|
|
confidence_threshold=0.25,
|
|
event=EventConfig(
|
|
keypoint_confidence_threshold=0.4,
|
|
suspect_window_seconds=0.5,
|
|
confirm_window_seconds=1.0,
|
|
recovery_window_seconds=2.0,
|
|
cooldown_seconds=3.0,
|
|
),
|
|
event_dir=Path("artifacts/events"),
|
|
)
|
|
|
|
|
|
def test_pipeline_confirms_a_recent_drop_that_remains_horizontal():
|
|
pipeline, machine = _pipeline([(_pose(),), (_pose(horizontal=True),), (_pose(horizontal=True),)])
|
|
|
|
pipeline.process(_packet(0.0))
|
|
pipeline.process(_packet(0.1))
|
|
result = pipeline.process(_packet(1.1))
|
|
|
|
assert len(result.events) == 1
|
|
assert result.events[0].config_version == "cfg-test-pipeline"
|
|
assert machine.state_of("P-0001") is FallState.CONFIRMED
|
|
|
|
|
|
def test_pipeline_rejects_a_suspect_when_the_track_is_missing_for_one_frame():
|
|
pipeline, machine = _pipeline([(_pose(),), (_pose(horizontal=True),), (), (_pose(horizontal=True),)])
|
|
|
|
pipeline.process(_packet(0.0))
|
|
pipeline.process(_packet(0.1))
|
|
pipeline.process(_packet(0.2))
|
|
result = pipeline.process(_packet(1.1))
|
|
|
|
assert result.events == ()
|
|
assert machine.state_of("P-0001") is FallState.NORMAL
|
|
|
|
|
|
def test_pipeline_from_config_uses_runtime_version_for_confirmed_event():
|
|
config = _config()
|
|
pipeline = FallPipeline.from_config(
|
|
config,
|
|
pose_adapter=_SequencePoseAdapter(
|
|
[(_pose(),), (_pose(horizontal=True),), (_pose(horizontal=True),)]
|
|
),
|
|
)
|
|
|
|
pipeline.process(_packet(0.0))
|
|
pipeline.process(_packet(0.1))
|
|
result = pipeline.process(_packet(1.1))
|
|
|
|
assert result.events[0].config_version == config.runtime_config_version
|
|
|
|
|
|
def test_from_config_gives_each_run_a_unique_event_id():
|
|
config = _config()
|
|
|
|
def confirm(pipeline):
|
|
pipeline.process(_packet(0.0))
|
|
pipeline.process(_packet(0.1))
|
|
return pipeline.process(_packet(1.1)).events[0].event_id
|
|
|
|
frames = [(_pose(),), (_pose(horizontal=True),), (_pose(horizontal=True),)]
|
|
first = confirm(FallPipeline.from_config(config, _SequencePoseAdapter(list(frames))))
|
|
second = confirm(FallPipeline.from_config(config, _SequencePoseAdapter(list(frames))))
|
|
|
|
assert first != second
|