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

fix: harden v1.2 trajectory disposal matching

Browse Source
This commit is contained in:
Yoilun
2026-05-29 16:26:15 +08:00
parent 90aa5dd704
commit 100b949f1f
11 changed files with 248 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
tests/test_engine.py
View File

@@ -250,6 +250,33 @@ class BatchEngineTests(unittest.TestCase):
self.assertEqual([(event["event"], event["zone_id"]) for event in events], [("batch_discarded", "4")])
self.assertEqual([batch.zone_id for batch in engine.pending_disposal], ["1"])
def test_extra_trash_deposits_still_fallback_after_matching_disposal_evidence(self) -> None:
settings = EngineSettings(
camera_id="test_cam",
max_dwell_seconds=1200,
trash_confirmation_seconds=120,
zone_ids=("1", "2"),
)
engine = BatchEngine(settings)
engine.process(obs(self.t0, {"1": 1, "2": 1}))
engine.process(obs(self.t0 + timedelta(seconds=1200), {"1": 1, "2": 1}))
engine.process(obs(self.t0 + timedelta(seconds=1300), {"1": 0, "2": 0}))
events = engine.process(
obs(
self.t0 + timedelta(seconds=1310),
{"1": 0, "2": 0},
trash=2,
disposal_evidence=[disposal_evidence("1")],
)
)
self.assertEqual(
[(event["event"], event["zone_id"]) for event in events],
[("batch_discarded", "1"), ("batch_discarded", "2")],
)
self.assertEqual(engine.pending_disposal, [])
def test_same_observation_removal_and_disposal_evidence_discards_newly_pending_batch(self) -> None:
settings = EngineSettings(
camera_id="test_cam",

34
tests/test_vision.py
View File

@@ -474,7 +474,39 @@ class VisionTests(unittest.TestCase):
rejected.extend(diagnostics["rejected"])
self.assertEqual(all_evidence, [])
self.assertTrue(any(item["reason"] == "missing_source_motion" for item in rejected))
self.assertTrue(any(item["reason"] == "motion_started_outside_source" for item in rejected))
def test_motion_before_source_motion_cannot_seed_later_trash_evidence(self) -> None:
source = Region("source", ((0.05, 0.35), (0.25, 0.35), (0.25, 0.65), (0.05, 0.65)))
trash = Region("trash", ((0.72, 0.35), (0.95, 0.35), (0.95, 0.65), (0.72, 0.65)))
tracker = TrajectoryTracker(
[source],
trash,
RuntimeVisionSettings(trajectory_sample_interval_seconds=0.0, trajectory_min_points=3),
)
now = datetime(2026, 4, 28, 10, 0, tzinfo=timezone.utc)
tracker.observe(solid_frame(80, 80, 40), now, {"source": 1})
all_evidence = []
rejected = []
frames = [
frame_with_motion_patch(80, 80, (34, 36)),
frame_with_motion_patch(80, 80, (16, 36)),
frame_with_motion_patch(80, 80, (34, 36)),
frame_with_motion_patch(80, 80, (50, 36)),
frame_with_motion_patch(80, 80, (66, 36)),
]
for index, frame in enumerate(frames):
evidence, diagnostics = tracker.observe(
frame,
now + timedelta(seconds=index + 1),
{"source": 0},
)
all_evidence.extend(evidence)
rejected.extend(diagnostics["rejected"])
self.assertEqual(all_evidence, [])
self.assertTrue(any(item["reason"] == "motion_started_outside_source" for item in rejected))
def test_trajectory_diagnostics_include_per_candidate_events(self) -> None:
source = Region("source", ((0.05, 0.35), (0.25, 0.35), (0.25, 0.65), (0.05, 0.65)))