Initial commit: frigate-notify application

AI-powered video alert service that bridges Frigate NVR events to Matrix.
Listens to MQTT, uses Claude Haiku vision to analyze snapshots, and posts
annotated alerts to a Matrix room.

- app/main.py: MQTT subscriber and event orchestration loop
- app/vision.py: Claude Haiku vision analysis wrapper
- app/matrix_client.py: Matrix HTTP client for sending messages/images
- app/requirements.txt: Python dependencies
- Dockerfile: container image build definition
- docker-compose.yml: deployment configuration (host network, NAS mounts)
- .env.example: required environment variable template (copy to .env)

.env.example

@@ -0,0 +1,11 @@
+# Anthropic API key for Claude Vision (Haiku model)
+# Get yours at: https://console.anthropic.com/
+ANTHROPIC_API_KEY=sk-ant-...
+
+# Matrix bot access token
+# Create a dedicated Matrix bot account and generate an access token
+MATRIX_ACCESS_TOKEN=syt_...
+
+# Matrix room ID where video alerts will be posted
+# Format: !roomid:homeserver.example.com
+MATRIX_ROOM_ID=!roomid:homeserver.example.com

.gitignore

@@ -0,0 +1,27 @@
+# Environment file with real secrets - NEVER commit this
+.env
+
+# Python
+__pycache__/
+*.py[cod]
+*.pyo
+*.pyd
+.Python
+*.egg-info/
+dist/
+build/
+*.egg
+
+# Docker
+*.log
+
+# Editor
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS
+.DS_Store
+Thumbs.db

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/*
+
+COPY app/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app/ .
+
+EXPOSE 5199
+
+CMD ["python", "-u", "main.py"]

app/main.py

@@ -0,0 +1,259 @@
+import asyncio
+import json
+import logging
+import os
+import signal
+import sys
+import time
+from collections import defaultdict
+from http.server import HTTPServer, BaseHTTPRequestHandler
+from threading import Thread
+
+import aiohttp
+import anthropic
+import paho.mqtt.client as mqtt
+
+from vision import analyze_snapshot
+from matrix_client import MatrixClient
+
+# --- Configuration ---
+MQTT_HOST = os.environ.get("MQTT_HOST", "192.168.86.192")
+MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
+FRIGATE_URL = os.environ.get("FRIGATE_URL", "http://192.168.86.192:5000")
+MATRIX_HOMESERVER = os.environ.get("MATRIX_HOMESERVER", "http://192.168.86.192:8008")
+MATRIX_ACCESS_TOKEN = os.environ["MATRIX_ACCESS_TOKEN"]
+MATRIX_ROOM_ID = os.environ["MATRIX_ROOM_ID"]
+HEALTH_PORT = int(os.environ.get("HEALTH_PORT", "5199"))
+
+ALLOWED_LABELS = {"person", "car", "dog", "cat", "bird"}
+MIN_SCORE = 0.7
+RATE_LIMIT_SECONDS = 60
+DEDUP_WINDOW_SECONDS = 30
+
+# --- Logging ---
+logging.basicConfig(
+    level=logging.INFO,
+    format='{"time":"%(asctime)s","level":"%(levelname)s","module":"%(module)s","message":"%(message)s"}',
+    stream=sys.stdout,
+)
+logger = logging.getLogger(__name__)
+
+# --- State ---
+recent_events: dict[str, float] = {}
+last_notify_per_camera: defaultdict[str, float] = defaultdict(float)
+events_processed = 0
+last_event_time: str | None = None
+mqtt_connected = False
+
+
+# --- Health endpoint ---
+class HealthHandler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def do_GET(self):
+        if self.path == "/health":
+            body = json.dumps(
+                {
+                    "status": "ok",
+                    "mqtt_connected": mqtt_connected,
+                    "events_processed": events_processed,
+                    "last_event": last_event_time,
+                }
+            )
+            encoded = body.encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(encoded)))
+            self.end_headers()
+            self.wfile.write(encoded)
+        else:
+            self.send_response(404)
+            self.end_headers()
+
+    def log_message(self, format, *args):
+        pass  # suppress access logs
+
+
+def start_health_server():
+    server = HTTPServer(("0.0.0.0", HEALTH_PORT), HealthHandler)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    logger.info("Health server started on port %d", HEALTH_PORT)
+
+
+# --- Snapshot fetching ---
+async def fetch_snapshot(event_id: str) -> bytes | None:
+    url = f"{FRIGATE_URL}/api/events/{event_id}/snapshot.jpg"
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url) as resp:
+                if resp.status == 200:
+                    data = await resp.read()
+                    logger.info(
+                        "Fetched snapshot for %s: %d bytes", event_id, len(data)
+                    )
+                    return data
+                logger.warning(
+                    "Snapshot fetch failed for %s: HTTP %d", event_id, resp.status
+                )
+    except Exception:
+        logger.exception("Snapshot fetch error for %s", event_id)
+    return None
+
+
+# --- Event processing ---
+async def process_event(
+    event_data: dict, claude: anthropic.Anthropic, matrix: MatrixClient
+):
+    global events_processed, last_event_time
+
+    event_type = event_data.get("type")
+    if event_type != "end":
+        return
+
+    after = event_data.get("after", {})
+    event_id = after.get("id", "")
+    camera = after.get("camera", "unknown")
+    label = after.get("label", "")
+    score = after.get("top_score", after.get("score", 0))
+    has_snapshot = after.get("has_snapshot", False)
+
+    # --- Filtering ---
+    if label not in ALLOWED_LABELS:
+        return
+    if score < MIN_SCORE:
+        return
+    if not has_snapshot:
+        return
+
+    # Dedup
+    now = time.time()
+    if event_id in recent_events and now - recent_events[event_id] < DEDUP_WINDOW_SECONDS:
+        return
+    recent_events[event_id] = now
+
+    # Rate limit per camera
+    if now - last_notify_per_camera[camera] < RATE_LIMIT_SECONDS:
+        logger.info("Rate limited: %s on %s", label, camera)
+        return
+    last_notify_per_camera[camera] = now
+
+    # Cleanup old dedup entries
+    cutoff = now - DEDUP_WINDOW_SECONDS * 2
+    for eid in [k for k, v in recent_events.items() if v < cutoff]:
+        del recent_events[eid]
+
+    logger.info(
+        "Processing event: %s on %s (%s, score=%.2f)", event_id, camera, label, score
+    )
+
+    # --- Fetch snapshot ---
+    snapshot = await fetch_snapshot(event_id)
+
+    # --- AI analysis ---
+    description = None
+    if snapshot:
+        description = analyze_snapshot(claude, snapshot)
+    if not description:
+        description = f"{label.capitalize()} detected on {camera}."
+
+    # --- Format message ---
+    score_pct = int(score * 100)
+    ts = time.strftime("%I:%M:%S %p", time.localtime())
+    date_str = time.strftime("%b %d, %Y", time.localtime())
+    frigate_link = f"{FRIGATE_URL}/review?id={event_id}"
+
+    plain = (
+        f"{camera} - {label.capitalize()} - {score_pct}%\n"
+        f"{description}\n\n"
+        f"View in Frigate: {frigate_link}\n"
+        f"{ts} - {date_str}"
+    )
+    html = (
+        f"<b>{camera}</b> &middot; {label.capitalize()} &middot; {score_pct}%<br/>"
+        f"{description}<br/><br/>"
+        f'<a href="{frigate_link}">View in Frigate</a><br/>'
+        f"<small>{ts} &middot; {date_str}</small>"
+    )
+
+    # --- Send to Matrix ---
+    if snapshot:
+        filename = f"{camera}_{event_id}.jpg"
+        mxc_uri = await matrix.upload_image(snapshot, filename)
+        if mxc_uri:
+            await matrix.send_image(mxc_uri, filename, len(snapshot))
+
+    await matrix.send_text(plain, html)
+
+    events_processed += 1
+    last_event_time = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+    logger.info("Notification sent for %s on %s", label, camera)
+
+
+# --- MQTT ---
+def create_mqtt_client(
+    loop: asyncio.AbstractEventLoop,
+    claude: anthropic.Anthropic,
+    matrix: MatrixClient,
+):
+    global mqtt_connected
+
+    def on_connect(client, userdata, flags, reason_code, properties):
+        global mqtt_connected
+        mqtt_connected = True
+        client.subscribe("frigate/events")
+        logger.info("MQTT connected, subscribed to frigate/events")
+
+    def on_disconnect(client, userdata, flags, reason_code, properties):
+        global mqtt_connected
+        mqtt_connected = False
+        logger.warning("MQTT disconnected: %s", reason_code)
+
+    def on_message(client, userdata, msg):
+        try:
+            payload = json.loads(msg.payload)
+            asyncio.run_coroutine_threadsafe(
+                process_event(payload, claude, matrix), loop
+            )
+        except json.JSONDecodeError:
+            logger.warning("Invalid JSON on %s", msg.topic)
+        except Exception:
+            logger.exception("Error processing MQTT message")
+
+    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
+    client.on_connect = on_connect
+    client.on_disconnect = on_disconnect
+    client.on_message = on_message
+    return client
+
+
+# --- Main ---
+def main():
+    logger.info("Starting frigate-notify")
+
+    claude = anthropic.Anthropic()
+    matrix = MatrixClient(MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID)
+
+    loop = asyncio.new_event_loop()
+    loop_thread = Thread(target=loop.run_forever, daemon=True)
+    loop_thread.start()
+
+    start_health_server()
+
+    mqtt_client = create_mqtt_client(loop, claude, matrix)
+    mqtt_client.connect(MQTT_HOST, MQTT_PORT)
+
+    def shutdown(signum, frame):
+        logger.info("Shutting down...")
+        mqtt_client.disconnect()
+        loop.call_soon_threadsafe(loop.stop)
+        sys.exit(0)
+
+    signal.signal(signal.SIGTERM, shutdown)
+    signal.signal(signal.SIGINT, shutdown)
+
+    mqtt_client.loop_forever()
+
+
+if __name__ == "__main__":
+    main()

app/matrix_client.py

@@ -0,0 +1,84 @@
+import asyncio
+import logging
+import time
+import aiohttp
+
+logger = logging.getLogger(__name__)
+
+RETRY_DELAYS = [1, 2, 4]
+
+
+class MatrixClient:
+    def __init__(self, homeserver: str, token: str, room_id: str):
+        self.homeserver = homeserver.rstrip("/")
+        self.token = token
+        self.room_id = room_id
+        self.headers = {"Authorization": f"Bearer {token}"}
+
+    async def _request(self, method: str, url: str, **kwargs) -> dict | None:
+        for attempt, delay in enumerate(RETRY_DELAYS):
+            try:
+                async with aiohttp.ClientSession() as session:
+                    async with session.request(
+                        method, url, headers=self.headers, **kwargs
+                    ) as resp:
+                        if resp.status < 300:
+                            return await resp.json()
+                        body = await resp.text()
+                        logger.warning(
+                            "Matrix API %s (attempt %d): %s",
+                            resp.status,
+                            attempt + 1,
+                            body,
+                        )
+            except Exception:
+                logger.exception("Matrix request failed (attempt %d)", attempt + 1)
+            if attempt < len(RETRY_DELAYS) - 1:
+                await asyncio.sleep(delay)
+        logger.error("All Matrix retries exhausted for %s", url)
+        return None
+
+    async def upload_image(self, image_bytes: bytes, filename: str) -> str | None:
+        url = f"{self.homeserver}/_matrix/media/v3/upload?filename={filename}"
+        headers = {**self.headers, "Content-Type": "image/jpeg"}
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(url, headers=headers, data=image_bytes) as resp:
+                    if resp.status < 300:
+                        data = await resp.json()
+                        mxc_uri = data.get("content_uri")
+                        logger.info("Uploaded image: %s", mxc_uri)
+                        return mxc_uri
+                    body = await resp.text()
+                    logger.error("Image upload failed %s: %s", resp.status, body)
+        except Exception:
+            logger.exception("Image upload failed")
+        return None
+
+    async def send_image(self, mxc_uri: str, filename: str, size: int) -> dict | None:
+        txn_id = str(int(time.time() * 1000))
+        url = (
+            f"{self.homeserver}/_matrix/client/r0/rooms/"
+            f"{self.room_id}/send/m.room.message/{txn_id}"
+        )
+        body = {
+            "msgtype": "m.image",
+            "body": filename,
+            "url": mxc_uri,
+            "info": {"mimetype": "image/jpeg", "size": size},
+        }
+        return await self._request("PUT", url, json=body)
+
+    async def send_text(self, plain: str, html: str) -> dict | None:
+        txn_id = str(int(time.time() * 1000) + 1)
+        url = (
+            f"{self.homeserver}/_matrix/client/r0/rooms/"
+            f"{self.room_id}/send/m.room.message/{txn_id}"
+        )
+        body = {
+            "msgtype": "m.text",
+            "body": plain,
+            "format": "org.matrix.custom.html",
+            "formatted_body": html,
+        }
+        return await self._request("PUT", url, json=body)

app/requirements.txt

@@ -0,0 +1,3 @@
+paho-mqtt==2.1.0
+anthropic>=0.40.0
+aiohttp>=3.9.0

app/vision.py

@@ -0,0 +1,42 @@
+import base64
+import logging
+import anthropic
+
+logger = logging.getLogger(__name__)
+
+PROMPT = (
+    "Briefly describe what's happening in this security camera image in 1-2 sentences. "
+    "Focus on the people, vehicles, or animals and their activity. Be concise."
+)
+
+
+def analyze_snapshot(client: anthropic.Anthropic, image_bytes: bytes) -> str | None:
+    """Send a snapshot to Claude Haiku for description. Returns None on failure."""
+    try:
+        b64 = base64.standard_b64encode(image_bytes).decode("utf-8")
+        response = client.messages.create(
+            model="claude-haiku-4-5-20251001",
+            max_tokens=150,
+            messages=[
+                {
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "image",
+                            "source": {
+                                "type": "base64",
+                                "media_type": "image/jpeg",
+                                "data": b64,
+                            },
+                        },
+                        {"type": "text", "text": PROMPT},
+                    ],
+                }
+            ],
+        )
+        description = response.content[0].text.strip()
+        logger.info("Vision analysis complete: %d chars", len(description))
+        return description
+    except Exception:
+        logger.exception("Claude Vision API failed")
+        return None

docker-compose.yml

@@ -0,0 +1,30 @@
+services:
+  frigate-notify:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    container_name: frigate-notify
+    restart: unless-stopped
+    network_mode: host
+    environment:
+      - TZ=America/Chicago
+      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
+      - MATRIX_ACCESS_TOKEN=${MATRIX_ACCESS_TOKEN}
+      - MATRIX_ROOM_ID=${MATRIX_ROOM_ID}
+      - FRIGATE_URL=http://192.168.86.192:5000
+      - MQTT_HOST=192.168.86.192
+      - MQTT_PORT=1883
+      - MATRIX_HOMESERVER=http://192.168.86.192:8008
+    mem_limit: 256M
+    memswap_limit: 256M
+    healthcheck:
+      test: ["CMD-SHELL", "curl -sf http://localhost:5199/health || exit 1"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 30s
+    logging:
+      driver: "json-file"
+      options:
+        max-size: "50m"
+        max-file: "3"