Rearchitected FrigateEventNotifier

notify/Dockerfile

@@ -1,7 +1,7 @@
 FROM python:3.11
 WORKDIR /code
 
-ENTRYPOINT ["python3", "mqtt.py"]
+ENTRYPOINT ["python3", "frigate_event_notifier.py"]
 
 RUN pip3 install --upgrade pip
 

notify/src/frigate_event_notifier.py

@@ -0,0 +1,118 @@
+import os
+import json
+from time import sleep
+from datetime import datetime, timedelta
+
+import requests
+import paho.mqtt.client as mqtt
+
+
+class FrigateEventNotifier:
+    def __init__(self, mqtt_username, mqtt_password, quiet_period=60):
+        self.mqtt_client = mqtt.Client()
+        self.mqtt_client.username_pw_set(mqtt_username, password=mqtt_password)
+        self.mqtt_client.on_connect = self._on_connect
+        self.mqtt_client.on_message = self._on_message
+
+        frigate_api_response = requests.get('http://frigate:5000/api/config')
+        frigate_api_response.raise_for_status()
+        frigate_config = json.loads(frigate_api_response.content)
+        self.camera_zones = {camera_name: {required_zone: {object_label for object_label in camera_config['zones'][required_zone]['objects']} for required_zone in camera_config['record']['events']['required_zones']} for camera_name, camera_config in frigate_config['cameras'].items()}
+
+        self.quiet_period = quiet_period
+        self.last_notification_time = {}
+
+    def send_notification(self, event_id, camera, object_label, score, priority=3):
+        now = datetime.now()
+        if now - self.last_notification_time.get(camera, datetime.min) >= timedelta(seconds=self.quiet_period):
+            # Quiet period has passed since the last notification for this camera
+            self.last_notification_time[camera] = now
+            camera_location = ' '.join(word.capitalize() for word in camera.split('_'))
+
+            ntfy_api_response = requests.post('https://ntfy.homelab.net', json={
+                'topic': 'frigate_notifications',
+                'title': 'Frigate',
+                'message': f'{object_label.capitalize()} at {camera_location} ({score:.0%})',
+                'priority': priority,
+                'click': f'https://frigate.homelab.net/cameras/{camera}',
+                'icon': 'https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/frigate.png',
+                'attach': f'https://frigate.homelab.net/api/events/{event_id}/thumbnail.jpg?format=android',
+                'actions': [
+                    {
+                        'action': 'http',
+                        'label': 'Disable (30m)',
+                        'url': f'https://frigate.homelab.net/webcontrol/camera/{camera}/detect',
+                        'method': 'POST',
+                        'headers': {
+                            'Content-Type': 'application/json'
+                        },
+                        'body': json.dumps({
+                            'value': False,
+                            'duration': 30
+                        }),
+                        'clear': True
+                    }
+                ]
+            })
+            ntfy_api_response.raise_for_status()
+
+            ntfy_api_response = requests.post('https://ntfy.homelab.net', json={
+                'topic': 'frigate_notifications_dad',
+                'title': 'Frigate',
+                'message': f'{object_label.capitalize()} at {camera_location} ({score:.0%})',
+                'priority': priority,
+                'click': f'https://frigate.homelab.net/cameras/{camera}',
+                'icon': 'https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/frigate.png',
+                'attach': f'https://frigate.homelab.net/api/events/{event_id}/thumbnail.jpg?format=android',
+                'actions': [
+                    {
+                        'action': 'http',
+                        'label': 'DBL (30m)',
+                        'url': f'https://frigate.homelab.net/webcontrol/camera/{camera}/detect',
+                        'method': 'POST',
+                        'headers': {
+                            'Content-Type': 'application/json'
+                        },
+                        'body': json.dumps({
+                            'value': False,
+                            'duration': 30
+                        })
+                    }
+                ]
+            })
+            ntfy_api_response.raise_for_status()
+
+    def start(self):
+        self.mqtt_client.connect(host='mqtt', port=1883)
+        self.mqtt_client.loop_forever()
+
+    def _on_connect(self, client, userdata, flags, rc):
+        print(f'Connected with return code {rc}')
+        client.subscribe('frigate/events')
+
+    def _on_message(self, client, userdata, message):
+        payload = json.loads(message.payload.decode())
+        camera = payload['after']['camera']
+        object_label = payload['after']['label']
+
+        if not self.camera_zones[camera]:
+            # No required zones, send notification on receipt of new event
+            if payload['type'] == 'new':
+                event_id = payload['after']['id']
+                score = payload['after']['top_score']
+
+                self.send_notification(event_id, camera, object_label, score)
+        else:
+            new_zones = set(payload['after']['entered_zones']) - set(payload['before']['entered_zones'])
+            for zone in new_zones:
+                if zone in self.camera_zones[camera] and (not self.camera_zones[camera][zone] or object_label in self.camera_zones[camera][zone]):
+                    event_id = payload['after']['id']
+                    score = payload['after']['top_score']
+
+                    self.send_notification(event_id, camera, object_label, score)
+                    break
+
+
+if __name__ == '__main__':
+    frigate_event_notifier = FrigateEventNotifier(os.environ['MQTT_USERNAME'], os.environ['MQTT_PASSWORD'])
+    frigate_event_notifier.start()

notify/src/mqtt.py

@@ -1,56 +0,0 @@
-import os
-import json
-from time import sleep
-
-import requests
-import paho.mqtt.client as mqtt
-
-from ntfy import send_rate_limited_notification
-
-
-class FrigateEventListener:
-    def __init__(self, username, password):
-        self.client = mqtt.Client()
-        self.client.username_pw_set(username, password=password)
-        self.client.on_connect = self._on_connect
-        self.client.on_message = self._on_message
-
-        frigate_api_response = requests.get('http://frigate:5000/api/config')
-        frigate_api_response.raise_for_status()
-        frigate_config = json.loads(frigate_api_response.content)
-        self.camera_zones = {camera_name: {required_zone: {object_label for object_label in camera_config['zones'][required_zone]['objects']} for required_zone in camera_config['record']['events']['required_zones']} for camera_name, camera_config in frigate_config['cameras'].items()}
-
-    def start(self):
-        self.client.connect(host='mqtt', port=1883)
-        self.client.loop_forever()
-
-    def _on_connect(self, client, userdata, flags, rc):
-        print(f'Connected with return code {rc}')
-        client.subscribe('frigate/events')
-
-    def _on_message(self, client, userdata, message):
-        payload = json.loads(message.payload.decode())
-        camera = payload['after']['camera']
-        object_label = payload['after']['label']
-
-        if not self.camera_zones[camera]:
-            # No required zones, send notification on receipt of new event
-            if payload['type'] == 'new':
-                event_id = payload['after']['id']
-                score = payload['after']['top_score']
-
-                send_rate_limited_notification(event_id, camera, object_label, score)
-        else:
-            new_zones = set(payload['after']['entered_zones']) - set(payload['before']['entered_zones'])
-            for zone in new_zones:
-                if zone in self.camera_zones[camera] and (not self.camera_zones[camera][zone] or object_label in self.camera_zones[camera][zone]):
-                    event_id = payload['after']['id']
-                    score = payload['after']['top_score']
-
-                    send_rate_limited_notification(event_id, camera, object_label, score)
-                    break
-
-
-if __name__ == '__main__':
-    frigate_event_listener = FrigateEventListener(os.environ['MQTT_USERNAME'], os.environ['MQTT_PASSWORD'])
-    frigate_event_listener.start()

notify/src/ntfy.py

@@ -1,72 +0,0 @@
-import json
-from datetime import datetime, timedelta
-
-import requests
-
-last_notification_time = {}
-
-
-def send_rate_limited_notification(event_id, camera, object_label, score, quiet_period=1):
-    global last_notification_time
-
-    now = datetime.now()
-    if camera not in last_notification_time or now - last_notification_time[camera] >= timedelta(minutes=quiet_period):
-        last_notification_time[camera] = now
-        send_notification(event_id, camera, object_label, score, priority=3)
-
-
-def send_notification(event_id, camera, object_label, score, priority=3):
-    camera_location = ' '.join(word.capitalize() for word in camera.split('_'))
-
-    response = requests.post('https://ntfy.homelab.net', json={
-        'topic': 'frigate_notifications',
-        'title': 'Frigate',
-        'message': f'{object_label.capitalize()} at {camera_location} ({score:.0%})',
-        'priority': priority,
-        'click': f'https://frigate.homelab.net/cameras/{camera}',
-        'icon': 'https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/frigate.png',
-        'attach': f'https://frigate.homelab.net/api/events/{event_id}/thumbnail.jpg?format=android',
-        'actions': [
-            {
-                'action': 'http',
-                'label': 'Disable (30m)',
-                'url': f'https://frigate.homelab.net/webcontrol/camera/{camera}/detect',
-                'method': 'POST',
-                'headers': {
-                    'Content-Type': 'application/json'
-                },
-                'body': json.dumps({
-                    'value': False,
-                    'duration': 30
-                }),
-                'clear': True
-            }
-        ]
-    })
-
-    response2 = requests.post('https://ntfy.homelab.net', json={
-        'topic': 'frigate_notifications_dad',
-        'title': 'Frigate',
-        'message': f'{object_label.capitalize()} at {camera_location} ({score:.0%})',
-        'priority': priority,
-        'click': f'https://frigate.homelab.net/cameras/{camera}',
-        'icon': 'https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/frigate.png',
-        'attach': f'https://frigate.homelab.net/api/events/{event_id}/thumbnail.jpg?format=android',
-        'actions': [
-            {
-                'action': 'http',
-                'label': 'DBL (30m)',
-                'url': f'https://frigate.homelab.net/webcontrol/camera/{camera}/detect',
-                'method': 'POST',
-                'headers': {
-                    'Content-Type': 'application/json'
-                },
-                'body': json.dumps({
-                    'value': False,
-                    'duration': 30
-                })
-            }
-        ]
-    })
-
-    return response.status_code == 200 and response2.status_code == 200