Fix notification debouncer

This commit is contained in:
Ashish D'Souza 2025-09-07 12:59:22 -05:00
parent 468a0b02f3
commit 4fb3d6a99e
2 changed files with 4 additions and 2 deletions

View File

@ -29,7 +29,7 @@ def on_event(payload: dict[str, Any]) -> None:
if __name__ == "__main__":
frigate_config = FrigateConfig()
frigate_config.fetch()
notification_sender = NtfyNotificationSender(quiet_period=dt.timedelta(seconds=3 * 60))
notification_sender = NtfyNotificationSender(quiet_period=dt.timedelta(minutes=3))
mqtt_subscriber = MQTTSubscriber(on_event, username=os.getenv("MQTT_USERNAME"), password=os.getenv("MQTT_PASSWORD"))
mqtt_subscriber.subscribe()

View File

@ -1,6 +1,7 @@
import datetime as dt
import time
from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass
from enum import StrEnum
from functools import cache, cached_property
@ -42,7 +43,8 @@ class NotificationSender[N: Notification](ABC):
def send(self, notification: N) -> None:
now = time.time()
last_notification_time = self._camera_to_last_notification_time.setdefault(notification.camera, 0)
last_notification_time = self._camera_to_last_notification_time.get(notification.camera, 0)
if now - last_notification_time >= self._quiet_period.total_seconds():
# Quiet period has passed since the last notification for this camera
self._camera_to_last_notification_time[notification.camera] = now
self._send(notification)