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.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): def test_restore_runtime_state_uses_stable_occupancy_when_raw_metrics_flicker(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: diagnostics_path = Path(tmpdir) / "runtime_diagnostics.jsonl" diagnostics_path.write_text( json.dumps( { "ts": "2026-05-29T10:05:26+08:00", "zone_counts": {"2": 1}, "diagnostics": { "baseline_ready": True, "zones": { "2": { "baseline_mean_luma": 165.0, "baseline_texture": 16.0, "baseline_dark_fraction": 0.0, "baseline_bright_fraction": 0.0, "mean_delta": 17.077, "texture_delta": 8.819, "dark_fraction": 0.0357, "bright_fraction": 0.0, "raw_occupied": False, "occupied": True, "empty_streak": 1, }, }, }, } ), encoding="utf-8", ) _, zone_counts = restore_runtime_state( diagnostics_path, { "runtime": { "occupancy_mean_delta": 55.0, "occupancy_texture_delta": 18.0, "occupancy_dark_fraction": 0.06, "occupancy_texture_dark_fraction": 0.04, } }, ) self.assertEqual(zone_counts, {"2": 1}) def test_restore_runtime_state_uses_dark_fraction_rules(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: diagnostics_path = Path(tmpdir) / "runtime_diagnostics.jsonl" diagnostics_path.write_text( json.dumps( { "ts": "2026-05-29T10:00:00+08:00", "zone_counts": {"1": 1, "4": 1}, "diagnostics": { "baseline_ready": True, "zones": { "1": { "baseline_mean_luma": 165.0, "baseline_texture": 16.0, "baseline_dark_fraction": 0.0, "baseline_bright_fraction": 0.0, "mean_delta": 40.0, "texture_delta": 18.0, "dark_fraction": 0.10, "bright_fraction": 0.0, }, "4": { "baseline_mean_luma": 177.0, "baseline_texture": 9.0, "baseline_dark_fraction": 0.0, "baseline_bright_fraction": 0.0, "mean_delta": 16.0, "texture_delta": 40.0, "dark_fraction": 0.0769, "bright_fraction": 0.3077, }, }, }, } ), encoding="utf-8", ) baselines, zone_counts = restore_runtime_state( diagnostics_path, { "runtime": { "occupancy_mean_delta": 55.0, "occupancy_texture_delta": 18.0, "occupancy_dark_fraction": 0.06, "occupancy_texture_dark_fraction": 0.04, } }, ) self.assertEqual(zone_counts, {"1": 1, "4": 0}) self.assertEqual(baselines["1"].dark_fraction, 0.0) 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()