mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-09-26 17:58:59 +03:00
Josh HawkinsandNicolas Mowen
7bc32fd4c9
* refactor notices * show startup message for enrichments in health pane * tweaks
130 lines
4.7 KiB
Python
130 lines
4.7 KiB
Python
"""Tests for the skipped detection rate and the notice it can raise."""
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from frigate.stats import emitter
|
|
from frigate.stats.emitter import SkippedDetectionsTracker
|
|
from frigate.stats.util import skipped_percent
|
|
|
|
|
|
class TestSkippedPercent(unittest.TestCase):
|
|
def test_share_of_all_frames(self):
|
|
self.assertEqual(skipped_percent(0.5, 5.0, True), 10.0)
|
|
|
|
def test_offline_camera_reports_zero(self):
|
|
self.assertEqual(skipped_percent(2.0, 0.0, True), 0.0)
|
|
|
|
def test_disabled_camera_reports_zero(self):
|
|
self.assertEqual(skipped_percent(2.0, 5.0, False), 0.0)
|
|
|
|
|
|
class TestSkippedDetectionsTracker(unittest.TestCase):
|
|
def _cameras(self, pct: float) -> dict:
|
|
return {"front_door": {"skipped_pct": pct}}
|
|
|
|
def test_below_threshold_never_qualifies(self):
|
|
tracker = SkippedDetectionsTracker()
|
|
|
|
for tick in range(10):
|
|
self.assertEqual(tracker.update(self._cameras(4.9), tick * 15.0), [])
|
|
|
|
def test_qualifies_once_after_the_hold(self):
|
|
tracker = SkippedDetectionsTracker()
|
|
|
|
results = [
|
|
tracker.update(self._cameras(10.0), tick * 15.0) for tick in range(8)
|
|
]
|
|
|
|
# ticks at 0, 15, 30, and 45 s are inside the hold; 60 s qualifies
|
|
self.assertEqual(results[:4], [[], [], [], []])
|
|
self.assertEqual(results[4], ["front_door"])
|
|
self.assertEqual(results[5:], [[], [], []])
|
|
|
|
def test_a_dip_restarts_the_hold(self):
|
|
tracker = SkippedDetectionsTracker()
|
|
|
|
for now, pct in ((0.0, 10.0), (15.0, 10.0), (30.0, 10.0), (45.0, 2.0)):
|
|
self.assertEqual(tracker.update(self._cameras(pct), now), [])
|
|
|
|
self.assertEqual(tracker.update(self._cameras(10.0), 60.0), [])
|
|
self.assertEqual(tracker.update(self._cameras(10.0), 120.0), ["front_door"])
|
|
|
|
def test_recovery_then_relapse_is_a_new_episode(self):
|
|
tracker = SkippedDetectionsTracker()
|
|
tracker.update(self._cameras(10.0), 0.0)
|
|
self.assertEqual(tracker.update(self._cameras(10.0), 60.0), ["front_door"])
|
|
|
|
tracker.update(self._cameras(0.0), 75.0)
|
|
tracker.update(self._cameras(10.0), 90.0)
|
|
|
|
self.assertEqual(tracker.update(self._cameras(10.0), 150.0), ["front_door"])
|
|
|
|
def test_cameras_are_tracked_separately(self):
|
|
tracker = SkippedDetectionsTracker()
|
|
tracker.update({"a": {"skipped_pct": 10.0}, "b": {"skipped_pct": 0.0}}, 0.0)
|
|
|
|
qualified = tracker.update(
|
|
{"a": {"skipped_pct": 10.0}, "b": {"skipped_pct": 10.0}}, 60.0
|
|
)
|
|
|
|
self.assertEqual(qualified, ["a"])
|
|
|
|
def test_a_removed_camera_starts_a_new_episode(self):
|
|
tracker = SkippedDetectionsTracker()
|
|
tracker.update(self._cameras(10.0), 0.0)
|
|
tracker.update({}, 15.0)
|
|
tracker.update(self._cameras(10.0), 30.0)
|
|
|
|
self.assertEqual(tracker.update(self._cameras(10.0), 60.0), [])
|
|
self.assertEqual(tracker.update(self._cameras(10.0), 90.0), ["front_door"])
|
|
|
|
|
|
class TestEmitterNotices(unittest.TestCase):
|
|
def setUp(self):
|
|
raise_patch = patch.object(emitter, "raise_notice")
|
|
resolve_patch = patch.object(emitter, "resolve_notice")
|
|
flush_patch = patch.object(emitter, "flush_notices")
|
|
self.raise_notice = raise_patch.start()
|
|
resolve_patch.start()
|
|
self.flush_notices = flush_patch.start()
|
|
self.addCleanup(raise_patch.stop)
|
|
self.addCleanup(resolve_patch.stop)
|
|
self.addCleanup(flush_patch.stop)
|
|
|
|
def _emitter(self) -> emitter.StatsEmitter:
|
|
stats_emitter = emitter.StatsEmitter.__new__(emitter.StatsEmitter)
|
|
stats_emitter.skipped_detections = SkippedDetectionsTracker()
|
|
stats_emitter._shm_checked = False
|
|
stats_emitter._shm_params = None
|
|
return stats_emitter
|
|
|
|
def _stats(self, uptime: int, pct: float) -> dict:
|
|
return {
|
|
"service": {"uptime": uptime, "storage": {"/dev/shm": {}}},
|
|
"cameras": {"front_door": {"skipped_pct": pct}},
|
|
}
|
|
|
|
def test_qualified_camera_raises_a_notice(self):
|
|
stats_emitter = self._emitter()
|
|
|
|
stats_emitter._update_notices(self._stats(300, 12.5), 0.0)
|
|
stats_emitter._update_notices(self._stats(360, 12.5), 60.0)
|
|
|
|
self.raise_notice.assert_called_once_with(
|
|
"skipped_detections", scope="front_door", params={"pct": 12.5}
|
|
)
|
|
|
|
def test_startup_window_is_ignored(self):
|
|
stats_emitter = self._emitter()
|
|
|
|
stats_emitter._update_notices(self._stats(30, 50.0), 0.0)
|
|
stats_emitter._update_notices(self._stats(90, 50.0), 60.0)
|
|
|
|
self.raise_notice.assert_not_called()
|
|
|
|
def test_every_tick_flushes_held_repeats(self):
|
|
self._emitter()._update_notices(self._stats(30, 0.0), 0.0)
|
|
|
|
self.flush_notices.assert_called_once_with()
|