Improve Notifications (#16453)

* backend

* frontend

* add notification config at camera level

* camera level notifications in dispatcher

* initial onconnect

* frontend

* backend for suspended notifications

* frontend

* use base communicator

* initialize all cameras in suspended array and use 0 for unsuspended

* remove switch and use select for suspending in frontend

* use timestamp instead of datetime

* frontend tweaks

* mqtt docs

* fix button width

* use grid for layout

* use thread and queue for processing notifications with 10s timeout

* clean up

* move async code to main class

* tweaks

* docs

* remove warning message

frigate/comms/base_communicator.py

@@ -0,0 +1,21 @@
+from abc import ABC, abstractmethod
+from typing import Any, Callable
+
+
+class Communicator(ABC):
+    """pub/sub model via specific protocol."""
+
+    @abstractmethod
+    def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
+        """Send data via specific protocol."""
+        pass
+
+    @abstractmethod
+    def subscribe(self, receiver: Callable) -> None:
+        """Pass receiver so communicators can pass commands."""
+        pass
+
+    @abstractmethod
+    def stop(self) -> None:
+        """Stop the communicator."""
+        pass

frigate/comms/dispatcher.py

@@ -3,17 +3,19 @@
 import datetime
 import json
 import logging
-from abc import ABC, abstractmethod
 from typing import Any, Callable, Optional
 
 from frigate.camera import PTZMetrics
 from frigate.camera.activity_manager import CameraActivityManager
+from frigate.comms.base_communicator import Communicator
 from frigate.comms.config_updater import ConfigPublisher
+from frigate.comms.webpush import WebPushClient
 from frigate.config import BirdseyeModeEnum, FrigateConfig
 from frigate.const import (
     CLEAR_ONGOING_REVIEW_SEGMENTS,
     INSERT_MANY_RECORDINGS,
     INSERT_PREVIEW,
+    NOTIFICATION_TEST,
     REQUEST_REGION_GRID,
     UPDATE_CAMERA_ACTIVITY,
     UPDATE_EMBEDDINGS_REINDEX_PROGRESS,
@@ -30,25 +32,6 @@ from frigate.util.services import restart_frigate
 logger = logging.getLogger(__name__)
 
 
-class Communicator(ABC):
-    """pub/sub model via specific protocol."""
-
-    @abstractmethod
-    def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
-        """Send data via specific protocol."""
-        pass
-
-    @abstractmethod
-    def subscribe(self, receiver: Callable) -> None:
-        """Pass receiver so communicators can pass commands."""
-        pass
-
-    @abstractmethod
-    def stop(self) -> None:
-        """Stop the communicator."""
-        pass
-
-
 class Dispatcher:
     """Handle communication between Frigate and communicators."""
 
@@ -77,18 +60,23 @@ class Dispatcher:
             "motion": self._on_motion_command,
             "motion_contour_area": self._on_motion_contour_area_command,
             "motion_threshold": self._on_motion_threshold_command,
+            "notifications": self._on_camera_notification_command,
             "recordings": self._on_recordings_command,
             "snapshots": self._on_snapshots_command,
             "birdseye": self._on_birdseye_command,
             "birdseye_mode": self._on_birdseye_mode_command,
         }
         self._global_settings_handlers: dict[str, Callable] = {
-            "notifications": self._on_notification_command,
+            "notifications": self._on_global_notification_command,
         }
 
         for comm in self.comms:
             comm.subscribe(self._receive)
 
+        self.web_push_client = next(
+            (comm for comm in communicators if isinstance(comm, WebPushClient)), None
+        )
+
     def _receive(self, topic: str, payload: str) -> Optional[Any]:
         """Handle receiving of payload from communicators."""
 
@@ -180,6 +168,13 @@ class Dispatcher:
                     "snapshots": self.config.cameras[camera].snapshots.enabled,
                     "record": self.config.cameras[camera].record.enabled,
                     "audio": self.config.cameras[camera].audio.enabled,
+                    "notifications": self.config.cameras[camera].notifications.enabled,
+                    "notifications_suspended": int(
+                        self.web_push_client.suspended_cameras.get(camera, 0)
+                    )
+                    if self.web_push_client
+                    and camera in self.web_push_client.suspended_cameras
+                    else 0,
                     "autotracking": self.config.cameras[
                         camera
                     ].onvif.autotracking.enabled,
@@ -192,6 +187,9 @@ class Dispatcher:
                 json.dumps(self.embeddings_reindex.copy()),
             )
 
+        def handle_notification_test():
+            self.publish("notification_test", "Test notification")
+
         # Dictionary mapping topic to handlers
         topic_handlers = {
             INSERT_MANY_RECORDINGS: handle_insert_many_recordings,
@@ -203,13 +201,14 @@ class Dispatcher:
             UPDATE_EVENT_DESCRIPTION: handle_update_event_description,
             UPDATE_MODEL_STATE: handle_update_model_state,
             UPDATE_EMBEDDINGS_REINDEX_PROGRESS: handle_update_embeddings_reindex_progress,
+            NOTIFICATION_TEST: handle_notification_test,
             "restart": handle_restart,
             "embeddingsReindexProgress": handle_embeddings_reindex_progress,
             "modelState": handle_model_state,
             "onConnect": handle_on_connect,
         }
 
-        if topic.endswith("set") or topic.endswith("ptz"):
+        if topic.endswith("set") or topic.endswith("ptz") or topic.endswith("suspend"):
             try:
                 parts = topic.split("/")
                 if len(parts) == 3 and topic.endswith("set"):
@@ -224,6 +223,11 @@ class Dispatcher:
                     # example /cam_name/ptz payload=MOVE_UP|MOVE_DOWN|STOP...
                     camera_name = parts[-2]
                     handle_camera_command("ptz", camera_name, "", payload)
+                elif len(parts) == 3 and topic.endswith("suspend"):
+                    # example /cam_name/notifications/suspend payload=duration
+                    camera_name = parts[-3]
+                    command = parts[-2]
+                    self._on_camera_notification_suspend(camera_name, payload)
             except IndexError:
                 logger.error(
                     f"Received invalid {topic.split('/')[-1]} command: {topic}"
@@ -365,16 +369,18 @@ class Dispatcher:
         self.config_updater.publish(f"config/motion/{camera_name}", motion_settings)
         self.publish(f"{camera_name}/motion_threshold/state", payload, retain=True)
 
-    def _on_notification_command(self, payload: str) -> None:
-        """Callback for notification topic."""
+    def _on_global_notification_command(self, payload: str) -> None:
+        """Callback for global notification topic."""
         if payload != "ON" and payload != "OFF":
-            f"Received unsupported value for notification: {payload}"
+            f"Received unsupported value for all notification: {payload}"
             return
 
         notification_settings = self.config.notifications
-        logger.info(f"Setting notifications: {payload}")
+        logger.info(f"Setting all notifications: {payload}")
         notification_settings.enabled = payload == "ON"  # type: ignore[union-attr]
-        self.config_updater.publish("config/notifications", notification_settings)
+        self.config_updater.publish(
+            "config/notifications", {"_global_notifications": notification_settings}
+        )
         self.publish("notifications/state", payload, retain=True)
 
     def _on_audio_command(self, camera_name: str, payload: str) -> None:
@@ -491,3 +497,71 @@ class Dispatcher:
 
         self.config_updater.publish(f"config/birdseye/{camera_name}", birdseye_settings)
         self.publish(f"{camera_name}/birdseye_mode/state", payload, retain=True)
+
+    def _on_camera_notification_command(self, camera_name: str, payload: str) -> None:
+        """Callback for camera level notifications topic."""
+        notification_settings = self.config.cameras[camera_name].notifications
+
+        if payload == "ON":
+            if not self.config.cameras[camera_name].notifications.enabled_in_config:
+                logger.error(
+                    "Notifications must be enabled in the config to be turned on via MQTT."
+                )
+                return
+
+            if not notification_settings.enabled:
+                logger.info(f"Turning on notifications for {camera_name}")
+                notification_settings.enabled = True
+            if (
+                self.web_push_client
+                and camera_name in self.web_push_client.suspended_cameras
+            ):
+                self.web_push_client.suspended_cameras[camera_name] = 0
+        elif payload == "OFF":
+            if notification_settings.enabled:
+                logger.info(f"Turning off notifications for {camera_name}")
+                notification_settings.enabled = False
+            if (
+                self.web_push_client
+                and camera_name in self.web_push_client.suspended_cameras
+            ):
+                self.web_push_client.suspended_cameras[camera_name] = 0
+
+        self.config_updater.publish(
+            "config/notifications", {camera_name: notification_settings}
+        )
+        self.publish(f"{camera_name}/notifications/state", payload, retain=True)
+        self.publish(f"{camera_name}/notifications/suspended", "0", retain=True)
+
+    def _on_camera_notification_suspend(self, camera_name: str, payload: str) -> None:
+        """Callback for camera level notifications suspend topic."""
+        try:
+            duration = int(payload)
+        except ValueError:
+            logger.error(f"Invalid suspension duration: {payload}")
+            return
+
+        if self.web_push_client is None:
+            logger.error("WebPushClient not available for suspension")
+            return
+
+        notification_settings = self.config.cameras[camera_name].notifications
+
+        if not notification_settings.enabled:
+            logger.error(f"Notifications are not enabled for {camera_name}")
+            return
+
+        if duration != 0:
+            self.web_push_client.suspend_notifications(camera_name, duration)
+        else:
+            self.web_push_client.unsuspend_notifications(camera_name)
+
+        self.publish(
+            f"{camera_name}/notifications/suspended",
+            str(
+                int(self.web_push_client.suspended_cameras.get(camera_name, 0))
+                if camera_name in self.web_push_client.suspended_cameras
+                else 0
+            ),
+            retain=True,
+        )

frigate/comms/inter_process.py

@@ -7,7 +7,7 @@ from typing import Callable
 
 import zmq
 
-from frigate.comms.dispatcher import Communicator
+from frigate.comms.base_communicator import Communicator
 
 SOCKET_REP_REQ = "ipc:///tmp/cache/comms"
 

frigate/comms/mqtt.py

@@ -5,7 +5,7 @@ from typing import Any, Callable
 import paho.mqtt.client as mqtt
 from paho.mqtt.enums import CallbackAPIVersion
 
-from frigate.comms.dispatcher import Communicator
+from frigate.comms.base_communicator import Communicator
 from frigate.config import FrigateConfig
 
 logger = logging.getLogger(__name__)

frigate/comms/webpush.py

@@ -4,13 +4,17 @@ import datetime
 import json
 import logging
 import os
+import queue
+import threading
+from dataclasses import dataclass
+from multiprocessing.synchronize import Event as MpEvent
 from typing import Any, Callable
 
 from py_vapid import Vapid01
 from pywebpush import WebPusher
 
+from frigate.comms.base_communicator import Communicator
 from frigate.comms.config_updater import ConfigSubscriber
-from frigate.comms.dispatcher import Communicator
 from frigate.config import FrigateConfig
 from frigate.const import CONFIG_DIR
 from frigate.models import User
@@ -18,15 +22,36 @@ from frigate.models import User
 logger = logging.getLogger(__name__)
 
 
+@dataclass
+class PushNotification:
+    user: str
+    payload: dict[str, Any]
+    title: str
+    message: str
+    direct_url: str = ""
+    image: str = ""
+    notification_type: str = "alert"
+    ttl: int = 0
+
+
 class WebPushClient(Communicator):  # type: ignore[misc]
     """Frigate wrapper for webpush client."""
 
-    def __init__(self, config: FrigateConfig) -> None:
+    def __init__(self, config: FrigateConfig, stop_event: MpEvent) -> None:
         self.config = config
         self.claim_headers: dict[str, dict[str, str]] = {}
         self.refresh: int = 0
         self.web_pushers: dict[str, list[WebPusher]] = {}
         self.expired_subs: dict[str, list[str]] = {}
+        self.suspended_cameras: dict[str, int] = {
+            c.name: 0 for c in self.config.cameras.values()
+        }
+        self.notification_queue: queue.Queue[PushNotification] = queue.Queue()
+        self.notification_thread = threading.Thread(
+            target=self._process_notifications, daemon=True
+        )
+        self.notification_thread.start()
+        self.stop_event = stop_event
 
         if not self.config.notifications.email:
             logger.warning("Email must be provided for push notifications to be sent.")
@@ -103,30 +128,144 @@ class WebPushClient(Communicator):  # type: ignore[misc]
 
         self.expired_subs = {}
 
+    def suspend_notifications(self, camera: str, minutes: int) -> None:
+        """Suspend notifications for a specific camera."""
+        suspend_until = int(
+            (datetime.datetime.now() + datetime.timedelta(minutes=minutes)).timestamp()
+        )
+        self.suspended_cameras[camera] = suspend_until
+        logger.info(
+            f"Notifications for {camera} suspended until {datetime.datetime.fromtimestamp(suspend_until).strftime('%Y-%m-%d %H:%M:%S')}"
+        )
+
+    def unsuspend_notifications(self, camera: str) -> None:
+        """Unsuspend notifications for a specific camera."""
+        self.suspended_cameras[camera] = 0
+        logger.info(f"Notifications for {camera} unsuspended")
+
+    def is_camera_suspended(self, camera: str) -> bool:
+        return datetime.datetime.now().timestamp() <= self.suspended_cameras[camera]
+
     def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
         """Wrapper for publishing when client is in valid state."""
         # check for updated notification config
         _, updated_notification_config = self.config_subscriber.check_for_update()
 
         if updated_notification_config:
-            self.config.notifications = updated_notification_config
+            for key, value in updated_notification_config.items():
+                if key == "_global_notifications":
+                    self.config.notifications = value
 
-        if not self.config.notifications.enabled:
-            return
+                elif key in self.config.cameras:
+                    self.config.cameras[key].notifications = value
 
         if topic == "reviews":
-            self.send_alert(json.loads(payload))
+            decoded = json.loads(payload)
+            camera = decoded["before"]["camera"]
+            if not self.config.cameras[camera].notifications.enabled:
+                return
+            if self.is_camera_suspended(camera):
+                logger.debug(f"Notifications for {camera} are currently suspended.")
+                return
+            self.send_alert(decoded)
+        elif topic == "notification_test":
+            if not self.config.notifications.enabled:
+                return
+            self.send_notification_test()
 
-    def send_alert(self, payload: dict[str, any]) -> None:
+    def send_push_notification(
+        self,
+        user: str,
+        payload: dict[str, Any],
+        title: str,
+        message: str,
+        direct_url: str = "",
+        image: str = "",
+        notification_type: str = "alert",
+        ttl: int = 0,
+    ) -> None:
+        notification = PushNotification(
+            user=user,
+            payload=payload,
+            title=title,
+            message=message,
+            direct_url=direct_url,
+            image=image,
+            notification_type=notification_type,
+            ttl=ttl,
+        )
+        self.notification_queue.put(notification)
+
+    def _process_notifications(self) -> None:
+        while not self.stop_event.is_set():
+            try:
+                notification = self.notification_queue.get(timeout=1.0)
+                self.check_registrations()
+
+                for pusher in self.web_pushers[notification.user]:
+                    endpoint = pusher.subscription_info["endpoint"]
+                    headers = self.claim_headers[
+                        endpoint[: endpoint.index("/", 10)]
+                    ].copy()
+                    headers["urgency"] = "high"
+
+                    resp = pusher.send(
+                        headers=headers,
+                        ttl=notification.ttl,
+                        data=json.dumps(
+                            {
+                                "title": notification.title,
+                                "message": notification.message,
+                                "direct_url": notification.direct_url,
+                                "image": notification.image,
+                                "id": notification.payload.get("after", {}).get(
+                                    "id", ""
+                                ),
+                                "type": notification.notification_type,
+                            }
+                        ),
+                        timeout=10,
+                    )
+
+                    if resp.status_code in (404, 410):
+                        self.expired_subs.setdefault(notification.user, []).append(
+                            endpoint
+                        )
+                    elif resp.status_code != 201:
+                        logger.warning(
+                            f"Failed to send notification to {notification.user} :: {resp.status_code}"
+                        )
+
+            except queue.Empty:
+                continue
+            except Exception as e:
+                logger.error(f"Error processing notification: {str(e)}")
+
+    def send_notification_test(self) -> None:
         if not self.config.notifications.email:
             return
 
         self.check_registrations()
 
-        # Only notify for alerts
-        if payload["after"]["severity"] != "alert":
+        for user in self.web_pushers:
+            self.send_push_notification(
+                user=user,
+                payload={},
+                title="Test Notification",
+                message="This is a test notification from Frigate.",
+                direct_url="/",
+                notification_type="test",
+            )
+
+    def send_alert(self, payload: dict[str, Any]) -> None:
+        if (
+            not self.config.notifications.email
+            or payload["after"]["severity"] != "alert"
+        ):
             return
 
+        self.check_registrations()
+
         state = payload["type"]
 
         # Don't notify if message is an update and important fields don't have an update
@@ -155,49 +294,21 @@ class WebPushClient(Communicator):  # type: ignore[misc]
 
         # if event is ongoing open to live view otherwise open to recordings view
         direct_url = f"/review?id={reviewId}" if state == "end" else f"/#{camera}"
+        ttl = 3600 if state == "end" else 0
 
-        for user, pushers in self.web_pushers.items():
-            for pusher in pushers:
-                endpoint = pusher.subscription_info["endpoint"]
-
-                # set headers for notification behavior
-                headers = self.claim_headers[
-                    endpoint[0 : endpoint.index("/", 10)]
-                ].copy()
-                headers["urgency"] = "high"
-                ttl = 3600 if state == "end" else 0
-
-                # send message
-                resp = pusher.send(
-                    headers=headers,
-                    ttl=ttl,
-                    data=json.dumps(
-                        {
-                            "title": title,
-                            "message": message,
-                            "direct_url": direct_url,
-                            "image": image,
-                            "id": reviewId,
-                            "type": "alert",
-                        }
-                    ),
-                )
-
-                if resp.status_code == 201:
-                    pass
-                elif resp.status_code == 404 or resp.status_code == 410:
-                    # subscription is not found or has been unsubscribed
-                    if not self.expired_subs.get(user):
-                        self.expired_subs[user] = []
-
-                    self.expired_subs[user].append(pusher.subscription_info["endpoint"])
-                    # the subscription no longer exists and should be removed
-                else:
-                    logger.warning(
-                        f"Failed to send notification to {user} :: {resp.headers}"
-                    )
+        for user in self.web_pushers:
+            self.send_push_notification(
+                user=user,
+                payload=payload,
+                title=title,
+                message=message,
+                direct_url=direct_url,
+                image=image,
+                ttl=ttl,
+            )
 
         self.cleanup_registrations()
 
     def stop(self) -> None:
-        pass
+        logger.info("Closing notification queue")
+        self.notification_thread.join()

frigate/comms/ws.py

@@ -15,7 +15,7 @@ from ws4py.server.wsgirefserver import (
 from ws4py.server.wsgiutils import WebSocketWSGIApplication
 from ws4py.websocket import WebSocket as WebSocket_
 
-from frigate.comms.dispatcher import Communicator
+from frigate.comms.base_communicator import Communicator
 from frigate.config import FrigateConfig
 
 logger = logging.getLogger(__name__)