yangyl/cold_display_guard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: integrate trajectory runtime diagnostics

Browse Source
This commit is contained in:
Yoilun
2026-05-29 15:58:26 +08:00
parent 39cfc76fa2
commit 90aa5dd704
6 changed files with 314 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

189
tests/test_main.py
View File

@@ -3,9 +3,14 @@ from __future__ import annotations
import json
import tempfile
import unittest
from datetime import datetime
from pathlib import Path
from unittest.mock import patch
from cold_display_guard.main import restore_runtime_state
from cold_display_guard.frame_source import FrameCaptureError
from cold_display_guard.main import run, restore_runtime_state
from cold_display_guard.models import DisposalEvidence
from cold_display_guard.vision import Frame
class RuntimeRestoreTests(unittest.TestCase):
@@ -109,5 +114,187 @@ class RuntimeRestoreTests(unittest.TestCase):
self.assertEqual(baselines["4"].bright_fraction, 0.0)
class RuntimeLoopTests(unittest.TestCase):
def test_run_writes_disposal_evidence_and_trajectory_diagnostics(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
config_path, diagnostics_path = write_runtime_config(tmpdir)
captured_observations = []
tracker_calls = []
class FakeSource:
def __init__(self, **kwargs: object) -> None:
pass
def capture(self) -> Frame:
return Frame(width=2, height=2, rgb=bytes([0, 0, 0]) * 4)
class FakeDetector:
def __init__(self, *args: object) -> None:
pass
def observe(self, frame: Frame, when: datetime) -> tuple[dict[str, int], int, dict[str, object]]:
return {"1": 0}, 0, {"zones": {"1": {"occupied": False}}}
class FakeTracker:
def __init__(self, *args: object) -> None:
self.has_active_candidates = False
def observe(
self,
frame: Frame,
when: datetime,
zone_counts: dict[str, int],
) -> tuple[list[DisposalEvidence], dict[str, object]]:
tracker_calls.append(zone_counts)
return [
DisposalEvidence(
source_zone_id="1",
target="trash",
confidence=0.9,
method="motion",
track_points=[{"x": 0.1, "y": 0.2}],
item_class=None,
detector_score=None,
observed_at=when.isoformat(),
)
], {"active_candidates": 0, "emitted_evidence": 1}
class FakeEngine:
def __init__(self, settings: object) -> None:
pass
def process(self, observation: object) -> list[dict[str, object]]:
captured_observations.append(observation)
return []
with patch("cold_display_guard.main.RTSPFrameSource", FakeSource), patch(
"cold_display_guard.main.ZoneOccupancyDetector", FakeDetector
), patch("cold_display_guard.main.TrajectoryTracker", FakeTracker), patch(
"cold_display_guard.main.BatchEngine", FakeEngine
):
run(config_path, max_iterations=1)
diagnostics = [json.loads(line) for line in diagnostics_path.read_text(encoding="utf-8").splitlines()]
self.assertEqual(len(captured_observations), 1)
self.assertEqual(tracker_calls, [{"1": 0}])
self.assertEqual(captured_observations[0].disposal_evidence[0].source_zone_id, "1")
self.assertEqual(diagnostics[0]["disposal_evidence"][0]["source_zone_id"], "1")
self.assertEqual(diagnostics[0]["disposal_evidence"][0]["target"], "trash")
self.assertEqual(diagnostics[0]["diagnostics"]["trajectory"]["emitted_evidence"], 1)
def test_run_uses_trajectory_sample_interval_when_candidates_are_active(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
config_path, _ = write_runtime_config(tmpdir, sample_interval=5.0, trajectory_interval=1.0)
sleeps = []
tracker_calls = []
class FakeSource:
def __init__(self, **kwargs: object) -> None:
pass
def capture(self) -> Frame:
return Frame(width=2, height=2, rgb=bytes([0, 0, 0]) * 4)
class FakeDetector:
def __init__(self, *args: object) -> None:
pass
def observe(self, frame: Frame, when: datetime) -> tuple[dict[str, int], int, dict[str, object]]:
return {"1": 0}, 0, {}
class FakeTracker:
def __init__(self, *args: object) -> None:
self.has_active_candidates = False
def observe(
self,
frame: Frame,
when: datetime,
zone_counts: dict[str, int],
) -> tuple[list[DisposalEvidence], dict[str, object]]:
tracker_calls.append(zone_counts)
self.has_active_candidates = True
return [], {"active_candidates": 1}
with patch("cold_display_guard.main.RTSPFrameSource", FakeSource), patch(
"cold_display_guard.main.ZoneOccupancyDetector", FakeDetector
), patch("cold_display_guard.main.TrajectoryTracker", FakeTracker), patch(
"cold_display_guard.main.time.sleep", sleeps.append
):
run(config_path, max_iterations=2)
self.assertEqual(tracker_calls, [{"1": 0}, {"1": 0}])
self.assertEqual(sleeps, [1.0])
def test_capture_failure_diagnostics_keep_trajectory_schema(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
config_path, diagnostics_path = write_runtime_config(tmpdir)
class FailingSource:
def __init__(self, **kwargs: object) -> None:
pass
def capture(self) -> Frame:
raise FrameCaptureError("camera offline")
with patch("cold_display_guard.main.RTSPFrameSource", FailingSource):
run(config_path, max_iterations=1)
diagnostics = [json.loads(line) for line in diagnostics_path.read_text(encoding="utf-8").splitlines()]
self.assertEqual(diagnostics[0]["error"], "frame_capture_failed")
self.assertEqual(diagnostics[0]["disposal_evidence"], [])
self.assertEqual(diagnostics[0]["diagnostics"]["trajectory"]["reason"], "frame_capture_failed")
def write_runtime_config(
tmpdir: str,
sample_interval: float = 5.0,
trajectory_interval: float = 1.0,
) -> tuple[Path, Path]:
root = Path(tmpdir)
event_path = root / "events.jsonl"
diagnostics_path = root / "runtime_diagnostics.jsonl"
config_path = root / "config.toml"
config_path.write_text(
"\n".join(
[
'camera_id = "test-camera"',
'timezone = "UTC"',
"",
"[stream]",
'rtsp_url = "rtsp://example.invalid/stream"',
"",
"[thresholds]",
"max_dwell_seconds = 1200",
"trash_confirmation_seconds = 120",
"",
"[layout]",
"zone_count = 1",
'zone_ids = ["1"]',
"",
"[[zones]]",
'id = "1"',
"polygon = [[0.0, 0.0], [0.5, 0.0], [0.5, 0.5], [0.0, 0.5]]",
"",
"[trash]",
"roi = [[0.6, 0.6], [1.0, 0.6], [1.0, 1.0], [0.6, 1.0]]",
"",
"[runtime]",
f"sample_interval_seconds = {sample_interval}",
f"trajectory_sample_interval_seconds = {trajectory_interval}",
f'diagnostics_path = "{diagnostics_path}"',
"",
"[event_sink]",
f'path = "{event_path}"',
"",
]
),
encoding="utf-8",
)
return config_path, diagnostics_path
if __name__ == "__main__":
unittest.main()

35
tests/test_vision.py
View File

@@ -547,6 +547,41 @@ class VisionTests(unittest.TestCase):
self.assertEqual(settings.yolo_model_path, "")
self.assertEqual(settings.yolo_min_confidence, 0.65)
def test_runtime_vision_settings_read_trajectory_and_yolo_fields_from_config(self) -> None:
settings = load_runtime_vision_settings(
{
"runtime": {
"trajectory_enabled": False,
"trajectory_window_seconds": 11,
"trajectory_sample_interval_seconds": 0.5,
"trajectory_min_points": 4,
"trajectory_min_confidence": 0.8,
"trajectory_motion_delta": 25.0,
"trajectory_min_blob_area": 20,
"trajectory_max_blob_area_fraction": 0.25,
"trajectory_trash_entry_margin": 0.02,
"trajectory_backend": "motion",
"yolo_enabled": True,
"yolo_model_path": "models/yolo.onnx",
"yolo_min_confidence": 0.7,
}
}
)
self.assertFalse(settings.trajectory_enabled)
self.assertEqual(settings.trajectory_window_seconds, 11)
self.assertEqual(settings.trajectory_sample_interval_seconds, 0.5)
self.assertEqual(settings.trajectory_min_points, 4)
self.assertEqual(settings.trajectory_min_confidence, 0.8)
self.assertEqual(settings.trajectory_motion_delta, 25.0)
self.assertEqual(settings.trajectory_min_blob_area, 20)
self.assertEqual(settings.trajectory_max_blob_area_fraction, 0.25)
self.assertEqual(settings.trajectory_trash_entry_margin, 0.02)
self.assertEqual(settings.trajectory_backend, "motion")
self.assertTrue(settings.yolo_enabled)
self.assertEqual(settings.yolo_model_path, "models/yolo.onnx")
self.assertEqual(settings.yolo_min_confidence, 0.7)
if __name__ == "__main__":
unittest.main()