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

feat: stabilize cold display runtime deployment

Browse Source
This commit is contained in:
Yoilun
2026-05-29 14:48:01 +08:00
parent ea5f9b1b07
commit 8b5bbff364
32 changed files with 5050 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/test_cli.py
View File

@@ -51,8 +51,9 @@ class CliTests(unittest.TestCase):
events = [json.loads(line) for line in output.getvalue().splitlines()]
self.assertEqual(
[event["event"] for event in events],
["batch_started", "batch_pending_disposal", "batch_discarded"],
["batch_started", "time_alarm", "batch_pending_disposal", "batch_discarded"],
)
self.assertEqual(events[1]["severity"], "alarm")
if __name__ == "__main__":

91
tests/test_config.py
View File

@@ -4,7 +4,7 @@ import tempfile
import unittest
from pathlib import Path
from cold_display_guard.config import load_settings
from cold_display_guard.config import load_settings, save_config_document
class ConfigTests(unittest.TestCase):
@@ -33,6 +33,95 @@ cols = 2
self.assertEqual(settings.trash_confirmation_seconds, 4)
self.assertEqual(settings.zone_ids, ("r1c1", "r1c2"))
def test_loads_numeric_zone_ids_for_custom_zone_count(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "config.toml"
path.write_text(
"""
camera_id = "cam_numeric"
[thresholds]
max_dwell_seconds = 1200
trash_confirmation_seconds = 120
[layout]
zone_count = 3
zone_ids = ["1", "2", "3"]
""".strip(),
encoding="utf-8",
)
settings = load_settings(path)
self.assertEqual(settings.camera_id, "cam_numeric")
self.assertEqual(settings.max_dwell_seconds, 1200)
self.assertEqual(settings.zone_ids, ("1", "2", "3"))
def test_rejects_more_than_ten_numeric_food_zones(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "config.toml"
path.write_text(
"""
[layout]
zone_ids = ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11"]
""".strip(),
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "1 to 10"):
load_settings(path)
def test_loads_numeric_zone_ids_from_zone_count_without_explicit_ids(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "config.toml"
path.write_text(
"""
[layout]
zone_count = 4
""".strip(),
encoding="utf-8",
)
settings = load_settings(path)
self.assertEqual(settings.zone_ids, ("1", "2", "3", "4"))
def test_rejects_numeric_zone_count_that_conflicts_with_zone_ids(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "config.toml"
path.write_text(
"""
[layout]
zone_count = 5
zone_ids = ["1", "2", "3"]
""".strip(),
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "zone_count"):
load_settings(path)
def test_save_config_document_round_trips_zone_count_and_numeric_labels(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "config.toml"
save_config_document(
path,
{
"layout": {"zone_count": 2, "zone_ids": ["1", "2"]},
"zones": [
{"id": "1", "label": "区域 1", "polygon": [[0, 0], [1, 0], [1, 1]]},
{"id": "2", "label": "区域 2", "polygon": [[0, 0], [0.5, 0], [0.5, 1]]},
],
"trash": {"roi": [[0, 0], [1, 0], [1, 1]]},
},
)
text = path.read_text(encoding="utf-8")
self.assertIn("zone_count = 2", text)
self.assertIn('label = "区域 1"', text)
self.assertIn("[trash]", text)
self.assertNotIn('"trash"', text.split("[layout]", maxsplit=1)[1].split("[[zones]]", maxsplit=1)[0])
if __name__ == "__main__":
unittest.main()

205
tests/test_engine.py
View File

@@ -36,6 +36,7 @@ class BatchEngineTests(unittest.TestCase):
self.assertEqual([event["event"] for event in events], ["batch_started"])
self.assertEqual(events[0]["zone_id"], "r1c1")
self.assertEqual(events[0]["current_count"], 3)
self.assertEqual(events[0]["severity"], "info")
def test_consumes_batch_when_removed_before_threshold(self) -> None:
self.engine.process(obs(self.t0, {"r1c1": 2}))
@@ -48,9 +49,29 @@ class BatchEngineTests(unittest.TestCase):
self.engine.process(obs(self.t0, {"r1c1": 2}))
events = self.engine.process(obs(self.t0 + timedelta(seconds=10), {"r1c1": 0}))
self.assertEqual([event["event"] for event in events], ["batch_pending_disposal"])
self.assertEqual(events[0]["dwell_seconds"], 10)
self.assertIn("disposal_deadline", events[0])
self.assertEqual([event["event"] for event in events], ["time_alarm", "batch_pending_disposal"])
self.assertEqual(events[0]["severity"], "alarm")
self.assertEqual(events[1]["dwell_seconds"], 10)
self.assertIn("disposal_deadline", events[1])
def test_removal_observation_at_threshold_emits_alarm_before_pending_disposal(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=1200,
trash_confirmation_seconds=120,
zone_ids=("1",),
)
engine = BatchEngine(settings)
engine.process(obs(self.t0, {"1": 1}))
events = engine.process(obs(self.t0 + timedelta(seconds=1200), {"1": 0}))
self.assertEqual([event["event"] for event in events], ["time_alarm", "batch_pending_disposal"])
self.assertEqual(events[0]["severity"], "alarm")
self.assertEqual(events[0]["current_count"], 1)
self.assertEqual(events[0]["zone_index"], 1)
self.assertEqual(events[1]["severity"], "warning")
self.assertEqual(events[1]["state"], "pending_disposal")
def test_trash_deposit_confirms_pending_disposal(self) -> None:
self.engine.process(obs(self.t0, {"r1c1": 2}))
@@ -59,12 +80,13 @@ class BatchEngineTests(unittest.TestCase):
self.assertEqual([event["event"] for event in events], ["batch_discarded"])
def test_missing_trash_deposit_raises_violation_after_deadline(self) -> None:
def test_missing_trash_deposit_escalates_warning_after_deadline(self) -> None:
self.engine.process(obs(self.t0, {"r1c1": 2}))
self.engine.process(obs(self.t0 + timedelta(seconds=11), {"r1c1": 0}))
events = self.engine.process(obs(self.t0 + timedelta(seconds=17), {"r1c1": 0}))
self.assertEqual([event["event"] for event in events], ["missing_disposal_violation"])
self.assertEqual([event["event"] for event in events], ["warning_escalated"])
self.assertEqual(events[0]["severity"], "warning")
self.assertEqual(events[0]["violation_reasons"], ["missing_disposal"])
def test_adding_food_before_zone_clears_raises_mixed_batch_violation(self) -> None:
@@ -72,6 +94,7 @@ class BatchEngineTests(unittest.TestCase):
events = self.engine.process(obs(self.t0 + timedelta(seconds=1), {"r1c1": 3}))
self.assertEqual([event["event"] for event in events], ["mixed_batch_violation"])
self.assertEqual(events[0]["severity"], "warning")
self.assertEqual(events[0]["reason"], "food_added_before_zone_cleared")
def test_count_decrease_keeps_same_batch_active(self) -> None:
@@ -93,6 +116,178 @@ class BatchEngineTests(unittest.TestCase):
)
self.assertEqual(events[0]["appeared_zones"], ["r1c2"])
def test_time_alarm_emits_once_while_batch_remains_in_zone(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=1200,
trash_confirmation_seconds=120,
zone_ids=("1",),
)
engine = BatchEngine(settings)
engine.process(obs(self.t0, {"1": 1}))
alarm_events = engine.process(obs(self.t0 + timedelta(seconds=1200), {"1": 1}))
repeated_events = engine.process(obs(self.t0 + timedelta(seconds=1300), {"1": 1}))
self.assertEqual([event["event"] for event in alarm_events], ["time_alarm"])
self.assertEqual(repeated_events, [])
self.assertEqual(alarm_events[0]["severity"], "alarm")
self.assertEqual(alarm_events[0]["zone_id"], "1")
self.assertEqual(alarm_events[0]["zone_index"], 1)
self.assertEqual(alarm_events[0]["zone_label"], "区域 1")
self.assertEqual(alarm_events[0]["dwell_seconds"], 1200)
self.assertEqual(alarm_events[0]["max_dwell_seconds"], 1200)
self.assertEqual(alarm_events[0]["current_count"], 1)
self.assertIn("alerted_at", alarm_events[0])
def test_alarmed_batch_removed_without_trash_deposit_escalates_warning(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=1200,
trash_confirmation_seconds=120,
zone_ids=("1",),
)
engine = BatchEngine(settings)
engine.process(obs(self.t0, {"1": 1}))
engine.process(obs(self.t0 + timedelta(seconds=1200), {"1": 1}))
pending_events = engine.process(obs(self.t0 + timedelta(seconds=1300), {"1": 0}))
warning_events = engine.process(obs(self.t0 + timedelta(seconds=1421), {"1": 0}))
self.assertEqual([event["event"] for event in pending_events], ["batch_pending_disposal"])
self.assertEqual(pending_events[0]["severity"], "warning")
self.assertEqual(pending_events[0]["state"], "pending_disposal")
self.assertEqual(pending_events[0]["zone_index"], 1)
self.assertEqual(pending_events[0]["ended_at"], (self.t0 + timedelta(seconds=1300)).isoformat())
self.assertEqual([event["event"] for event in warning_events], ["warning_escalated"])
self.assertEqual(warning_events[0]["severity"], "warning")
self.assertEqual(warning_events[0]["state"], "warning")
self.assertEqual(warning_events[0]["reason"], "alarmed_batch_removed_without_trash_deposit")
self.assertEqual(warning_events[0]["zone_label"], "区域 1")
def test_alarmed_batch_removed_with_trash_deposit_is_discarded(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=1200,
trash_confirmation_seconds=120,
zone_ids=("1",),
)
engine = BatchEngine(settings)
engine.process(obs(self.t0, {"1": 1}))
engine.process(obs(self.t0 + timedelta(seconds=1200), {"1": 1}))
engine.process(obs(self.t0 + timedelta(seconds=1300), {"1": 0}))
events = engine.process(obs(self.t0 + timedelta(seconds=1310), {"1": 0}, trash=True))
self.assertEqual([event["event"] for event in events], ["batch_discarded"])
self.assertEqual(events[0]["severity"], "info")
self.assertEqual(events[0]["state"], "discarded")
def test_same_observation_removal_and_trash_motion_discards_alerted_batch(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=1200,
trash_confirmation_seconds=120,
zone_ids=("1",),
)
engine = BatchEngine(settings)
engine.process(obs(self.t0, {"1": 1}))
engine.process(obs(self.t0 + timedelta(seconds=1200), {"1": 1}))
events = engine.process(obs(self.t0 + timedelta(seconds=1300), {"1": 0}, trash=True))
later_events = engine.process(obs(self.t0 + timedelta(seconds=1421), {"1": 0}))
self.assertEqual([event["event"] for event in events], ["batch_pending_disposal", "batch_discarded"])
self.assertEqual(events[1]["state"], "discarded")
self.assertEqual(later_events, [])
def test_same_observation_trash_motion_discards_multiple_newly_pending_batches(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=300,
trash_confirmation_seconds=120,
zone_ids=("1", "4"),
)
engine = BatchEngine(settings)
engine.process(obs(self.t0, {"1": 1, "4": 1}))
engine.process(obs(self.t0 + timedelta(seconds=300), {"1": 1, "4": 1}))
events = engine.process(obs(self.t0 + timedelta(seconds=360), {"1": 0, "4": 0}, trash=True))
later_events = engine.process(obs(self.t0 + timedelta(seconds=481), {"1": 0, "4": 0}))
self.assertEqual(
[event["event"] for event in events],
["batch_pending_disposal", "batch_pending_disposal", "batch_discarded", "batch_discarded"],
)
self.assertEqual([event["zone_id"] for event in events if event["event"] == "batch_discarded"], ["1", "4"])
self.assertEqual(later_events, [])
def test_restore_keeps_active_alarm_batch_after_runtime_restart(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=1200,
trash_confirmation_seconds=120,
zone_ids=("1",),
)
engine = BatchEngine(settings)
engine.restore_from_events(
[
{
"event": "batch_started",
"zone_id": "1",
"batch_id": "batch_000124",
"started_at": self.t0.isoformat(),
"current_count": 1,
"state": "active",
},
{
"event": "time_alarm",
"zone_id": "1",
"batch_id": "batch_000124",
"started_at": self.t0.isoformat(),
"alerted_at": (self.t0 + timedelta(seconds=1200)).isoformat(),
"current_count": 1,
"state": "alerted",
},
],
active_zone_counts={"1": 1},
)
repeated_events = engine.process(obs(self.t0 + timedelta(seconds=1300), {"1": 1}))
removal_events = engine.process(obs(self.t0 + timedelta(seconds=1400), {"1": 0}))
self.assertEqual(repeated_events, [])
self.assertEqual([event["event"] for event in removal_events], ["batch_pending_disposal"])
self.assertEqual(removal_events[0]["batch_id"], "batch_000124")
self.assertEqual(removal_events[0]["dwell_seconds"], 1400)
def test_restore_skips_active_false_positive_when_latest_zone_count_is_empty(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=1200,
trash_confirmation_seconds=120,
zone_ids=("3",),
)
engine = BatchEngine(settings)
engine.restore_from_events(
[
{
"event": "time_alarm",
"zone_id": "3",
"batch_id": "batch_000213",
"started_at": self.t0.isoformat(),
"alerted_at": (self.t0 + timedelta(seconds=1200)).isoformat(),
"current_count": 1,
"state": "alerted",
},
],
active_zone_counts={"3": 0},
)
events = engine.process(obs(self.t0 + timedelta(seconds=1300), {"3": 0}))
self.assertEqual(events, [])
self.assertEqual(engine.active_by_zone, {})
if __name__ == "__main__":
unittest.main()

113
tests/test_main.py Normal file
View File

@@ -0,0 +1,113 @@
from __future__ import annotations
import json
import tempfile
import unittest
from pathlib import Path
from cold_display_guard.main import restore_runtime_state
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)
if __name__ == "__main__":
unittest.main()

248
tests/test_manage_api.py
View File

@@ -28,6 +28,62 @@ class ManageApiTests(unittest.TestCase):
self.assertEqual(merged["zones"][1]["id"], "r1c2")
self.assertEqual(merged["trash"]["roi"][0], [0.8, 0.8])
def test_merge_calibration_replaces_numeric_food_zones_and_keeps_trash_separate(self) -> None:
data = {
"layout": {"zone_count": 2, "zone_ids": ["1", "2"]},
"zones": [
{"id": "1", "polygon": [[0, 0], [0.3, 0], [0.3, 0.3]]},
{"id": "2", "polygon": [[0.3, 0], [0.6, 0], [0.6, 0.3]]},
],
}
merged = merge_calibration(
data,
[
{"id": "1", "label": "区域 1", "polygon": [[0, 0], [0.2, 0], [0.2, 0.2]]},
{"id": "2", "label": "区域 2", "polygon": [[0.2, 0], [0.4, 0], [0.4, 0.2]]},
{"id": "3", "label": "区域 3", "polygon": [[0.4, 0], [0.6, 0], [0.6, 0.2]]},
],
[[0.8, 0.8], [1, 0.8], [1, 1], [0.8, 1]],
)
self.assertEqual(merged["layout"]["zone_count"], 3)
self.assertEqual(merged["layout"]["zone_ids"], ["1", "2", "3"])
self.assertEqual([zone["label"] for zone in merged["zones"]], ["区域 1", "区域 2", "区域 3"])
self.assertEqual(merged["trash"]["roi"][0], [0.8, 0.8])
self.assertNotIn("trash", merged["layout"]["zone_ids"])
def test_merge_calibration_preserves_numeric_zone_count_when_some_zones_are_unmarked(self) -> None:
data = {
"layout": {"zone_count": 3, "zone_ids": ["1", "2", "3"]},
"zones": [
{"id": "1", "label": "区域 1", "polygon": [[0, 0], [0.2, 0], [0.2, 0.2]]},
{"id": "2", "label": "区域 2", "polygon": [[0.2, 0], [0.4, 0], [0.4, 0.2]]},
],
}
merged = merge_calibration(
data,
[{"id": "1", "label": "区域 1", "polygon": [[0, 0], [0.3, 0], [0.3, 0.3]]}],
[[0.8, 0.8], [1, 0.8], [1, 1]],
{"zone_count": 3, "zone_ids": ["1", "2", "3"]},
)
self.assertEqual(merged["layout"]["zone_count"], 3)
self.assertEqual(merged["layout"]["zone_ids"], ["1", "2", "3"])
self.assertEqual([zone["id"] for zone in merged["zones"]], ["1", "2"])
self.assertEqual(merged["zones"][0]["polygon"], [[0.0, 0.0], [0.3, 0.0], [0.3, 0.3]])
self.assertEqual(merged["zones"][1]["polygon"], [[0.2, 0.0], [0.4, 0.0], [0.4, 0.2]])
def test_merge_calibration_rejects_more_than_ten_numeric_food_zones(self) -> None:
zones = [
{"id": str(index), "polygon": [[0, 0], [0.1, 0], [0.1, 0.1]]}
for index in range(1, 12)
]
with self.assertRaisesRegex(ValueError, "1 to 10"):
merge_calibration({"layout": {}}, zones, None)
def test_save_config_document_round_trips_manage_fields(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "config.toml"
@@ -65,8 +121,9 @@ class ManageApiTests(unittest.TestCase):
events_path.write_text(
"\n".join(
[
json.dumps({"event": "batch_started", "ts": "2026-04-27T10:00:00+08:00"}),
json.dumps({"event": "missing_disposal_violation", "ts": "2026-04-27T13:02:00+08:00"}),
json.dumps({"event": "batch_started", "severity": "info", "ts": "2026-04-27T10:00:00+08:00"}),
json.dumps({"event": "time_alarm", "severity": "alarm", "ts": "2026-04-27T12:00:00+08:00"}),
json.dumps({"event": "warning_escalated", "severity": "warning", "ts": "2026-04-27T13:02:00+08:00"}),
]
),
encoding="utf-8",
@@ -74,8 +131,43 @@ class ManageApiTests(unittest.TestCase):
summary = build_summary(ManageContext(config_path=config_path, project_root=root))
self.assertEqual(summary["metrics"]["event_count"], 2)
self.assertEqual(summary["metrics"]["event_count"], 3)
self.assertEqual(summary["metrics"]["alert_count"], 1)
self.assertEqual(summary["metrics"]["warning_count"], 1)
self.assertEqual(summary["metrics"]["violation_count"], 1)
self.assertEqual(summary["metrics"]["latest_alert_time"], "2026-04-27T13:02:00+08:00")
def test_summary_counts_escalated_and_legacy_warnings_without_pending_disposal(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
config_path = root / "config" / "local.toml"
save_config_document(
config_path,
{
"event_sink": {"path": "logs/events.jsonl"},
"layout": {"rows": 1, "cols": 1, "zone_ids": ["r1c1"]},
},
)
events_path = root / "logs" / "events.jsonl"
events_path.parent.mkdir()
events_path.write_text(
"\n".join(
[
json.dumps({"event": "batch_pending_disposal", "severity": "warning", "ts": "2026-04-27T12:01:00+08:00"}),
json.dumps({"event": "mixed_batch_violation", "ts": "2026-04-27T12:02:00+08:00"}),
json.dumps({"event": "warning_escalated", "severity": "warning", "ts": "2026-04-27T12:03:00+08:00"}),
]
),
encoding="utf-8",
)
summary = build_summary(ManageContext(config_path=config_path, project_root=root))
self.assertEqual(summary["metrics"]["event_count"], 3)
self.assertEqual(summary["metrics"]["alert_count"], 0)
self.assertEqual(summary["metrics"]["warning_count"], 2)
self.assertEqual(summary["metrics"]["violation_count"], 2)
self.assertEqual(summary["metrics"]["latest_alert_time"], "2026-04-27T12:03:00+08:00")
def test_summary_reads_runtime_diagnostics(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
@@ -108,6 +200,156 @@ class ManageApiTests(unittest.TestCase):
self.assertEqual(summary["metrics"]["latest_zone_counts"], {"r1c1": 1})
self.assertTrue(summary["metrics"]["baseline_ready"])
def test_summary_uses_stable_runtime_occupancy_when_raw_metrics_flicker(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
config_path = root / "config" / "local.toml"
save_config_document(
config_path,
{
"runtime": {
"diagnostics_path": "logs/runtime_diagnostics.jsonl",
"occupancy_mean_delta": 55.0,
"occupancy_texture_delta": 18.0,
"occupancy_dark_fraction": 0.06,
"occupancy_texture_dark_fraction": 0.04,
},
"event_sink": {"path": "logs/events.jsonl"},
"layout": {"zone_count": 2, "zone_ids": ["1", "2"]},
},
)
diagnostics_path = root / "logs" / "runtime_diagnostics.jsonl"
diagnostics_path.parent.mkdir()
diagnostics_path.write_text(
json.dumps(
{
"ts": "2026-05-29T10:05:26+08:00",
"zone_counts": {"1": 0, "2": 1},
"diagnostics": {
"baseline_ready": True,
"zones": {
"1": {
"mean_delta": 0.0,
"texture_delta": 0.0,
"dark_fraction": 0.0,
"baseline_dark_fraction": 0.0,
"bright_fraction": 0.0,
"occupied": False,
},
"2": {
"mean_delta": 17.077,
"texture_delta": 8.819,
"dark_fraction": 0.0357,
"baseline_dark_fraction": 0.0,
"bright_fraction": 0.0,
"raw_occupied": False,
"occupied": True,
"empty_streak": 1,
},
},
},
}
),
encoding="utf-8",
)
summary = build_summary(ManageContext(config_path=config_path, project_root=root))
self.assertEqual(summary["metrics"]["latest_zone_counts"], {"1": 0, "2": 1})
def test_summary_recomputes_latest_zone_counts_from_runtime_thresholds_when_stable_state_is_absent(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
config_path = root / "config" / "local.toml"
save_config_document(
config_path,
{
"runtime": {
"diagnostics_path": "logs/runtime_diagnostics.jsonl",
"occupancy_mean_delta": 45.0,
"occupancy_texture_delta": 18.0,
},
"event_sink": {"path": "logs/events.jsonl"},
"layout": {"zone_count": 3, "zone_ids": ["1", "2", "3"]},
},
)
diagnostics_path = root / "logs" / "runtime_diagnostics.jsonl"
diagnostics_path.parent.mkdir()
diagnostics_path.write_text(
json.dumps(
{
"ts": "2026-05-27T11:02:23+08:00",
"zone_counts": {"1": 1, "3": 1},
"diagnostics": {
"baseline_ready": True,
"zones": {
"1": {"mean_delta": 70.0, "texture_delta": 27.0},
"3": {"mean_delta": 36.0, "texture_delta": -9.0},
},
},
}
),
encoding="utf-8",
)
summary = build_summary(ManageContext(config_path=config_path, project_root=root))
self.assertEqual(summary["metrics"]["latest_zone_counts"], {"1": 1, "3": 0})
def test_summary_recomputes_latest_zone_counts_with_dark_fraction_rule(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
config_path = root / "config" / "local.toml"
save_config_document(
config_path,
{
"runtime": {
"diagnostics_path": "logs/runtime_diagnostics.jsonl",
"occupancy_mean_delta": 55.0,
"occupancy_texture_delta": 18.0,
"occupancy_dark_fraction": 0.06,
"occupancy_texture_dark_fraction": 0.04,
"occupancy_bright_reflection_fraction": 0.18,
},
"event_sink": {"path": "logs/events.jsonl"},
"layout": {"zone_count": 2, "zone_ids": ["1", "2"]},
},
)
diagnostics_path = root / "logs" / "runtime_diagnostics.jsonl"
diagnostics_path.parent.mkdir()
diagnostics_path.write_text(
json.dumps(
{
"ts": "2026-05-28T09:41:13+08:00",
"zone_counts": {"1": 1, "2": 1},
"diagnostics": {
"baseline_ready": True,
"zones": {
"1": {
"mean_delta": 45.0,
"texture_delta": 20.0,
"dark_fraction": 0.20,
"baseline_dark_fraction": 0.0,
"bright_fraction": 0.0,
},
"2": {
"mean_delta": 16.0,
"texture_delta": 40.0,
"dark_fraction": 0.0769,
"baseline_dark_fraction": 0.0,
"bright_fraction": 0.3077,
},
},
},
}
),
encoding="utf-8",
)
summary = build_summary(ManageContext(config_path=config_path, project_root=root))
self.assertEqual(summary["metrics"]["latest_zone_counts"], {"1": 1, "2": 0})
if __name__ == "__main__":
unittest.main()

206
tests/test_vision.py
View File

@@ -1,13 +1,15 @@
from __future__ import annotations
import unittest
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from cold_display_guard.vision import (
Frame,
Region,
RegionMetrics,
RuntimeVisionSettings,
ZoneOccupancyDetector,
load_runtime_vision_settings,
point_in_polygon,
)
@@ -26,6 +28,16 @@ def patched_frame(width: int, height: int, base: int, patch: tuple[int, int, int
return Frame(width=width, height=height, rgb=bytes(pixels))
def multi_patched_frame(width: int, height: int, base: int, patches: list[tuple[int, int, int, int, int]]) -> Frame:
pixels = bytearray(bytes([base, base, base]) * width * height)
for x1, y1, x2, y2, value in patches:
for y in range(y1, y2):
for x in range(x1, x2):
offset = (y * width + x) * 3
pixels[offset : offset + 3] = bytes([value, value, value])
return Frame(width=width, height=height, rgb=bytes(pixels))
class VisionTests(unittest.TestCase):
def test_point_in_polygon(self) -> None:
polygon = ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0))
@@ -42,6 +54,8 @@ class VisionTests(unittest.TestCase):
sample_stride_pixels=4,
occupancy_mean_delta=10,
occupancy_texture_delta=10,
occupancy_confirm_frames=1,
empty_confirm_frames=1,
),
)
now = datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc)
@@ -67,6 +81,196 @@ class VisionTests(unittest.TestCase):
self.assertEqual(first_deposit, 0)
self.assertEqual(second_deposit, 1)
def test_detector_reports_sustained_trash_motion_below_single_frame_threshold(self) -> None:
trash = Region("trash", ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)))
detector = ZoneOccupancyDetector(
[],
trash_region=trash,
settings=RuntimeVisionSettings(
sample_stride_pixels=4,
trash_motion_delta=18,
trash_sustained_motion_delta=8,
trash_sustained_motion_frames=2,
),
)
now = datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc)
_, first_deposit, _ = detector.observe(solid_frame(32, 32, 30), now)
_, second_deposit, second_diagnostics = detector.observe(solid_frame(32, 32, 39), now)
_, third_deposit, third_diagnostics = detector.observe(solid_frame(32, 32, 48), now)
self.assertEqual(first_deposit, 0)
self.assertEqual(second_deposit, 0)
self.assertEqual(second_diagnostics["trash"]["motion_streak"], 1)
self.assertEqual(third_deposit, 1)
self.assertEqual(third_diagnostics["trash"]["motion_streak"], 2)
def test_detector_allows_quick_sequential_strong_trash_motions(self) -> None:
trash = Region("trash", ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)))
detector = ZoneOccupancyDetector(
[],
trash_region=trash,
settings=RuntimeVisionSettings(sample_stride_pixels=4, trash_motion_delta=18),
)
now = datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc)
_, first_deposit, _ = detector.observe(solid_frame(32, 32, 30), now)
_, second_deposit, _ = detector.observe(solid_frame(32, 32, 90), now + timedelta(seconds=1))
_, third_deposit, third_diagnostics = detector.observe(solid_frame(32, 32, 30), now + timedelta(seconds=7))
self.assertEqual(first_deposit, 0)
self.assertEqual(second_deposit, 1)
self.assertEqual(third_deposit, 1)
self.assertFalse(third_diagnostics["trash"]["in_cooldown"])
def test_detector_requires_consecutive_occupied_frames(self) -> None:
detector = ZoneOccupancyDetector(
[Region("1", ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)))],
trash_region=None,
settings=RuntimeVisionSettings(
baseline_frames=1,
sample_stride_pixels=4,
occupancy_mean_delta=10,
occupancy_texture_delta=10,
occupancy_confirm_frames=2,
empty_confirm_frames=2,
),
)
now = datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc)
detector.observe(solid_frame(32, 32, 30), now)
first_counts, _, first_diagnostics = detector.observe(solid_frame(32, 32, 90), now)
second_counts, _, second_diagnostics = detector.observe(solid_frame(32, 32, 90), now)
first_empty_counts, _, _ = detector.observe(solid_frame(32, 32, 30), now)
second_empty_counts, _, _ = detector.observe(solid_frame(32, 32, 30), now)
self.assertEqual(first_counts, {"1": 0})
self.assertTrue(first_diagnostics["zones"]["1"]["raw_occupied"])
self.assertEqual(first_diagnostics["zones"]["1"]["occupied_streak"], 1)
self.assertEqual(second_counts, {"1": 1})
self.assertTrue(second_diagnostics["zones"]["1"]["occupied"])
self.assertEqual(first_empty_counts, {"1": 1})
self.assertEqual(second_empty_counts, {"1": 0})
def test_runtime_vision_defaults_raise_brightness_reflection_threshold(self) -> None:
settings = load_runtime_vision_settings({})
self.assertEqual(settings.sample_stride_pixels, 4)
self.assertEqual(settings.occupancy_mean_delta, 55.0)
self.assertEqual(settings.occupancy_confirm_frames, 2)
self.assertEqual(settings.empty_confirm_frames, 2)
self.assertEqual(settings.trash_motion_delta, 18.0)
self.assertEqual(settings.trash_sustained_motion_delta, 8.0)
self.assertEqual(settings.trash_sustained_motion_frames, 2)
self.assertEqual(settings.trash_motion_cooldown_seconds, 3)
def test_detector_can_seed_previous_baseline_and_occupancy(self) -> None:
detector = ZoneOccupancyDetector(
[Region("1", ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)))],
trash_region=None,
settings=RuntimeVisionSettings(
baseline_frames=10,
sample_stride_pixels=4,
occupancy_mean_delta=55,
occupancy_texture_delta=18,
occupancy_confirm_frames=2,
empty_confirm_frames=2,
),
)
detector.seed_baseline({"1": RegionMetrics(mean_luma=30.0, texture=0.0, sample_count=1)})
detector.seed_occupancy({"1": 1})
counts, _, diagnostics = detector.observe(solid_frame(32, 32, 90), datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc))
self.assertTrue(diagnostics["baseline_ready"])
self.assertEqual(counts, {"1": 1})
self.assertTrue(diagnostics["zones"]["1"]["occupied"])
def test_detector_reports_compact_dark_object_as_occupied(self) -> None:
detector = ZoneOccupancyDetector(
[Region("1", ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)))],
trash_region=None,
settings=RuntimeVisionSettings(
baseline_frames=1,
sample_stride_pixels=4,
occupancy_mean_delta=55,
occupancy_texture_delta=100,
occupancy_dark_luma_threshold=80,
occupancy_dark_fraction=0.06,
occupancy_confirm_frames=1,
empty_confirm_frames=1,
),
)
now = datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc)
detector.observe(solid_frame(32, 32, 180), now)
counts, _, diagnostics = detector.observe(patched_frame(32, 32, 180, (0, 0, 8, 32, 20)), now)
self.assertEqual(counts, {"1": 1})
self.assertGreaterEqual(diagnostics["zones"]["1"]["dark_fraction"], 0.06)
def test_detector_ignores_bright_reflection_without_dark_object(self) -> None:
detector = ZoneOccupancyDetector(
[Region("1", ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)))],
trash_region=None,
settings=RuntimeVisionSettings(
baseline_frames=1,
sample_stride_pixels=4,
occupancy_mean_delta=55,
occupancy_texture_delta=10,
occupancy_dark_luma_threshold=80,
occupancy_dark_fraction=0.06,
occupancy_texture_dark_fraction=0.04,
occupancy_confirm_frames=1,
empty_confirm_frames=1,
),
)
now = datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc)
detector.observe(solid_frame(32, 32, 160), now)
counts, _, diagnostics = detector.observe(patched_frame(32, 32, 160, (0, 0, 8, 32, 255)), now)
self.assertEqual(counts, {"1": 0})
self.assertGreaterEqual(diagnostics["zones"]["1"]["texture_delta"], 10)
self.assertLess(diagnostics["zones"]["1"]["dark_fraction"], 0.04)
def test_detector_ignores_bright_reflection_with_small_dark_edge(self) -> None:
detector = ZoneOccupancyDetector(
[Region("1", ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)))],
trash_region=None,
settings=RuntimeVisionSettings(
baseline_frames=1,
sample_stride_pixels=4,
occupancy_mean_delta=55,
occupancy_texture_delta=18,
occupancy_dark_luma_threshold=80,
occupancy_dark_fraction=0.06,
occupancy_texture_dark_fraction=0.04,
occupancy_confirm_frames=1,
empty_confirm_frames=1,
),
)
now = datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc)
detector.observe(solid_frame(40, 40, 180), now)
counts, _, diagnostics = detector.observe(
multi_patched_frame(
40,
40,
180,
[
(0, 0, 12, 40, 255),
(12, 0, 16, 32, 20),
],
),
now,
)
zone = diagnostics["zones"]["1"]
self.assertEqual(counts, {"1": 0})
self.assertGreaterEqual(zone["dark_fraction"], 0.06)
self.assertGreaterEqual(zone["bright_fraction"], 0.18)
if __name__ == "__main__":
unittest.main()