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

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()