commit 6086ed3cc037cf43eb77f686830cf64bd62d19f1 Author: Jeremy Hobbs Date: Sat Mar 7 14:30:57 2026 -0600 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) diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..19cb57d --- /dev/null +++ b/.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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7d7bd2d --- /dev/null +++ b/.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 diff --git a/Dockerfile b/Dockerfile new file mode 100755 index 0000000..65ddc4f --- /dev/null +++ b/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"] diff --git a/app/main.py b/app/main.py new file mode 100755 index 0000000..ba296db --- /dev/null +++ b/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"{camera} · {label.capitalize()} · {score_pct}%
" + f"{description}

" + f'View in Frigate
' + f"{ts} · {date_str}" + ) + + # --- 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() diff --git a/app/matrix_client.py b/app/matrix_client.py new file mode 100755 index 0000000..0a2578a --- /dev/null +++ b/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) diff --git a/app/requirements.txt b/app/requirements.txt new file mode 100755 index 0000000..8991258 --- /dev/null +++ b/app/requirements.txt @@ -0,0 +1,3 @@ +paho-mqtt==2.1.0 +anthropic>=0.40.0 +aiohttp>=3.9.0 diff --git a/app/vision.py b/app/vision.py new file mode 100755 index 0000000..9540dc2 --- /dev/null +++ b/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 diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100755 index 0000000..8dfef6b --- /dev/null +++ b/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"