hobbs-iac/docs/plans/2026-03-04-frigate-video-alerts.md

# Frigate Video Alerts Implementation Plan

> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

**Goal:** Build a Python container that listens for Frigate detection events via MQTT, analyzes snapshots with Claude Vision, and posts rich notifications to a Matrix room.

**Architecture:** A single Python container (`frigate-notify`) on hf-pdocker-01 subscribes to Frigate's MQTT events, filters for person/car/animal detections, fetches snapshots from Frigate's API, sends them to Claude Haiku for natural language description, then posts the snapshot image and description to a dedicated `#video-alerts` Matrix room via a new `@frigate:hobbs.farm` bot.

**Tech Stack:** Python 3.12, paho-mqtt, anthropic SDK, aiohttp, Docker Compose

**Design doc:** `docs/plans/2026-03-04-frigate-video-alerts-design.md`

---

### Task 1: Create Matrix bot account and room

This task runs on hf-pdocker-01 via SSH. No code files involved.

**Step 1: Register @frigate:hobbs.farm bot account**

Use the docker-swarm-ops agent to SSH to hf-pdocker-01 (192.168.86.192) and run:

```bash
# Register the bot via Synapse admin API
# First get an admin token (use @alertbot or register directly)
docker exec $(docker ps -q -f name=matrix_synapse) register_new_matrix_user \
  -u frigate \
  -p "FrigateBot2026!" \
  -a \
  --no-admin \
  -c /data/homeserver.yaml \
  http://localhost:8008
```

**Step 2: Login to get access token**

```bash
curl -s -X POST http://192.168.86.192:8008/_matrix/client/r0/login \
  -H "Content-Type: application/json" \
  -d '{"type":"m.login.password","user":"frigate","password":"FrigateBot2026!"}'
```

Save the `access_token` from the response.

**Step 3: Create #video-alerts room**

```bash
curl -s -X POST "http://192.168.86.192:8008/_matrix/client/r0/createRoom" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <TOKEN>" \
  -d '{
    "room_alias_name": "video-alerts",
    "name": "Video Alerts",
    "topic": "Frigate NVR detection alerts with AI descriptions",
    "visibility": "private",
    "invite": ["@jeremy:hobbs.farm"]
  }'
```

Save the `room_id` from the response.

**Step 4: Send a test message to verify**

```bash
curl -s -X PUT "http://192.168.86.192:8008/_matrix/client/r0/rooms/<ROOM_ID>/send/m.room.message/$(date +%s)" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <TOKEN>" \
  -d '{"msgtype":"m.text","body":"Frigate video alerts bot is online."}'
```

**Step 5: Store credentials in vault**

Use the vault-credential-manager agent to store:
- Name: "Frigate Matrix Bot"
- Username: @frigate:hobbs.farm
- Password: FrigateBot2026!
- Notes: access_token=<TOKEN>, room_id=<ROOM_ID>

**Step 6: Commit** (nothing to commit yet - this is infrastructure setup)

---

### Task 2: Create the Python application files

**Files:**
- Create: `compose/hf-pdocker-01/frigate-notify/app/requirements.txt`
- Create: `compose/hf-pdocker-01/frigate-notify/app/vision.py`
- Create: `compose/hf-pdocker-01/frigate-notify/app/matrix_client.py`
- Create: `compose/hf-pdocker-01/frigate-notify/app/main.py`

**Step 1: Create requirements.txt**

```
compose/hf-pdocker-01/frigate-notify/app/requirements.txt
```

```text
paho-mqtt==2.1.0
anthropic>=0.40.0
aiohttp>=3.9.0
```

**Step 2: Create vision.py - Claude Vision wrapper**

```
compose/hf-pdocker-01/frigate-notify/app/vision.py
```

```python
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
```

**Step 3: Create matrix_client.py - Matrix image upload and messaging**

```
compose/hf-pdocker-01/frigate-notify/app/matrix_client.py
```

```python
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 _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/{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/{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)


async def _sleep(seconds: float):
    import asyncio
    await asyncio.sleep(seconds)
```

**Step 4: Create main.py - MQTT listener and event processor**

```
compose/hf-pdocker-01/frigate-notify/app/main.py
```

```python
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 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):
    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,
            })
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(body.encode())
        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:
    import aiohttp
    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{description}\n\nView in Frigate: {frigate_link}\n{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()
```

**Step 5: Commit**

```bash
git add compose/hf-pdocker-01/frigate-notify/app/
git commit -m "feat: add frigate-notify Python application

MQTT listener for Frigate events with Claude Vision analysis
and Matrix notification delivery."
```

---

### Task 3: Create Dockerfile and docker-compose.yml

**Files:**
- Create: `compose/hf-pdocker-01/frigate-notify/Dockerfile`
- Create: `compose/hf-pdocker-01/frigate-notify/docker-compose.yml`
- Create: `compose/hf-pdocker-01/frigate-notify/.env`

**Step 1: Create Dockerfile**

```
compose/hf-pdocker-01/frigate-notify/Dockerfile
```

```dockerfile
FROM python:3.12-slim

WORKDIR /app

COPY app/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ .

EXPOSE 5199

CMD ["python", "-u", "main.py"]
```

**Step 2: Create docker-compose.yml**

```
compose/hf-pdocker-01/frigate-notify/docker-compose.yml
```

```yaml
services:
  frigate-notify:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: frigate-notify
    restart: unless-stopped
    network_mode: host
    environment:
      - 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"
```

**Step 3: Create .env file**

```
compose/hf-pdocker-01/frigate-notify/.env
```

```
ANTHROPIC_API_KEY=<get from existing netclaw .env or vault>
MATRIX_ACCESS_TOKEN=<from Task 1 step 2>
MATRIX_ROOM_ID=<from Task 1 step 3>
```

Note: Get the ANTHROPIC_API_KEY from the existing netclaw deployment:
```bash
ssh glyptic@192.168.86.100 "cat /mnt/docker-data/compose/hf-pdocker-02/netclaw/.env | grep ANTHROPIC"
```

**Step 4: Commit**

```bash
git add compose/hf-pdocker-01/frigate-notify/Dockerfile
git add compose/hf-pdocker-01/frigate-notify/docker-compose.yml
# Do NOT commit .env (contains secrets)
git commit -m "feat: add frigate-notify Docker configuration

Standalone Compose service on hf-pdocker-01 with host networking,
256M memory limit, health check on port 5199."
```

---

### Task 4: Build and deploy on hf-pdocker-01

This task uses the docker-swarm-ops agent to SSH to hf-pdocker-01.

**Step 1: Copy files to hf-pdocker-01**

SCP the entire `compose/hf-pdocker-01/frigate-notify/` directory to `/mnt/docker-data/compose/hf-pdocker-01/frigate-notify/` on hf-pdocker-01.

**Step 2: Build the container**

```bash
cd /mnt/docker-data/compose/hf-pdocker-01/frigate-notify
docker compose build
```

**Step 3: Start the container**

```bash
docker compose up -d
```

**Step 4: Verify container is running**

```bash
docker ps --filter name=frigate-notify --format "table {{.Names}}\t{{.Status}}"
```

Expected: `frigate-notify   Up X seconds (health: starting)`

**Step 5: Check logs for successful startup**

```bash
docker logs frigate-notify --tail 20
```

Expected: Health server started, MQTT connected, subscribed to frigate/events.

**Step 6: Verify health endpoint**

```bash
curl -s http://localhost:5199/health | python3 -m json.tool
```

Expected: `{"status": "ok", "mqtt_connected": true, "events_processed": 0, "last_event": null}`

---

### Task 5: Add monitoring and verify end-to-end

**Files:**
- Modify: `stacks/monitoring/prometheus.yml` - add health probe target

**Step 1: Add blackbox probe target**

In `stacks/monitoring/prometheus.yml`, add to the `http-pdocker01` targets list:

```yaml
- http://192.168.86.192:5199/health   # Frigate Notify
```

**Step 2: Deploy updated prometheus config**

SCP to hf-pdocker-01, copy to NAS, hot-reload Prometheus:

```bash
sudo cp /data/monitoring/prometheus.yml /mnt/docker-data/configs/prometheus/prometheus.yml
# Reload inside container
docker exec $(docker ps -q -f name=monitoring_prometheus) wget -qO- --post-data='' http://localhost:9090/-/reload
```

**Step 3: Trigger a test detection**

Walk in front of a camera, or wait for an organic detection event. Check logs:

```bash
docker logs frigate-notify --tail 30 -f
```

Expected: Event processing log, snapshot fetch, vision analysis, Matrix notification sent.

**Step 4: Verify Matrix message received**

Check `#video-alerts:hobbs.farm` in Matrix for the snapshot image and AI-generated description.

**Step 5: Store credentials in vault**

Use vault-credential-manager agent to store the Frigate bot credentials.

**Step 6: Commit monitoring update**

```bash
git add stacks/monitoring/prometheus.yml
git commit -m "feat: add frigate-notify to Prometheus blackbox probes"
```

---

### Task 6: Update documentation and memory

**Step 1: Update MEMORY.md**

Add entry for frigate-notify service:
- Container location, port, purpose
- Matrix bot identity and room
- Key config notes

**Step 2: Update CLAUDE.md service distribution**

Add frigate-notify to hf-pdocker-01 Compose services list.

**Step 3: Update alert-manager agent**

Add frigate-notify to the service list and probe targets documentation.

**Step 4: Commit documentation updates**

```bash
git add -A
git commit -m "docs: add frigate-notify to infrastructure documentation"
```