frigate-notify/app/main.py

"""frigate-notify v2 — FastAPI application with MQTT listener."""

import asyncio
import json
import logging
import os
import signal
import sys
import time
from collections import defaultdict
from contextlib import asynccontextmanager
from pathlib import Path
from threading import Thread

import aiohttp
import anthropic
import paho.mqtt.client as mqtt
import uvicorn
from fastapi import FastAPI
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

from config import config
from database import close_connection, get_connection, init_db, set_db_path
from matrix_client import MatrixClient
from routes.events import router as events_router
from routes.known_objects import router as known_objects_router
from routes.settings import router as settings_router
from ollama_client import OllamaClient
from recognition import recognize_event, get_clip_embedding, embedding_to_bytes
from vision import get_description
from cleanup import cleanup_old_events

# --- 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
mqtt_client_instance: mqtt.Client | None = None


# --- Snapshot fetching ---
async def fetch_snapshot(event_id: str) -> bytes | None:
    url = f"{config.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,
    ollama_client: OllamaClient = None,
):
    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 config.ALLOWED_LABELS:
        return
    if score < config.MIN_SCORE:
        return
    if not has_snapshot:
        return

    # Dedup
    now = time.time()
    if (
        event_id in recent_events
        and now - recent_events[event_id] < config.DEDUP_WINDOW_SECONDS
    ):
        return
    recent_events[event_id] = now

    # Rate limit per camera
    if now - last_notify_per_camera[camera] < config.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 - config.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)

    # --- Recognition ---
    match_status = "unreviewed"
    matched_id = None
    match_conf = 0.0
    matched_name = None

    # Check if recognition is enabled
    recognition_enabled = False
    try:
        conn = get_connection()
        row = conn.execute("SELECT value FROM settings WHERE key='recognition_enabled'").fetchone()
        if row:
            recognition_enabled = row["value"].lower() == "true"
    except Exception:
        pass

    if recognition_enabled and snapshot:
        match_status, matched_id, match_conf, matched_name = await recognize_event(
            snapshot, ollama_client
        )
        # Store embedding with event
        embedding = await get_clip_embedding(snapshot)
        if embedding:
            embedding_blob = embedding_to_bytes(embedding)
            try:
                conn = get_connection()
                conn.execute(
                    "UPDATE events SET embedding=?, match_status=?, matched_object_id=?, match_confidence=? WHERE frigate_event_id=?",
                    (embedding_blob, match_status, matched_id, match_conf, event_id),
                )
                conn.commit()
            except Exception:
                logger.exception("Failed to update event recognition")

        # If known, check action - skip notification for quiet/suppress
        if match_status == "known" and matched_id:
            conn = get_connection()
            obj = conn.execute("SELECT action FROM known_objects WHERE id=?", (matched_id,)).fetchone()
            if obj and obj["action"] in ("quiet", "suppress"):
                logger.info("Known object '%s' (%s), skipping notification", matched_name, obj["action"])
                events_processed += 1
                last_event_time = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                return

    # --- AI description ---
    vision_backend = "haiku"
    try:
        conn = get_connection()
        row = conn.execute("SELECT value FROM settings WHERE key='vision_backend'").fetchone()
        if row:
            vision_backend = row["value"]
    except Exception:
        pass

    description = None
    if snapshot:
        description = await get_description(snapshot, vision_backend, claude, ollama_client)
    if not description:
        description = f"{label.capitalize()} detected on {camera}."

    # --- Store event in database ---
    try:
        conn = get_connection()
        conn.execute(
            """INSERT OR IGNORE INTO events
               (frigate_event_id, camera, label, score, snapshot, match_status, ai_description, notified)
               VALUES (?, ?, ?, ?, ?, ?, ?, 1)""",
            (event_id, camera, label, score, snapshot, match_status, description),
        )
        conn.commit()
    except Exception:
        logger.exception("Failed to store event %s in database", event_id)

    # --- 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"{config.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,
    ollama_client: OllamaClient = None,
) -> mqtt.Client:
    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, ollama_client), 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


def start_mqtt(loop: asyncio.AbstractEventLoop) -> None:
    """Start MQTT client in a background thread."""
    global mqtt_client_instance

    claude = anthropic.Anthropic()
    matrix = MatrixClient(
        config.MATRIX_HOMESERVER,
        config.MATRIX_ACCESS_TOKEN,
        config.MATRIX_ROOM_ID,
    )
    ollama_client = OllamaClient(config.OLLAMA_URL, config.OLLAMA_MODEL)

    mqtt_client_instance = create_mqtt_client(loop, claude, matrix, ollama_client)
    mqtt_client_instance.connect(config.MQTT_HOST, config.MQTT_PORT)

    thread = Thread(target=mqtt_client_instance.loop_forever, daemon=True)
    thread.start()
    logger.info("MQTT client started in background thread")


# --- FastAPI lifespan ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup and shutdown logic for the FastAPI application."""
    logger.info("Starting frigate-notify v2")

    # Initialize database
    set_db_path(config.DB_PATH)
    init_db()

    # Start MQTT listener in background
    loop = asyncio.get_event_loop()
    start_mqtt(loop)

    # Start periodic event cleanup
    asyncio.create_task(cleanup_old_events())

    yield

    # Shutdown
    logger.info("Shutting down frigate-notify v2")
    if mqtt_client_instance:
        mqtt_client_instance.disconnect()
    close_connection()


# --- FastAPI app ---
app = FastAPI(
    title="frigate-notify",
    version="2.0.0",
    lifespan=lifespan,
)

# Include API routers
app.include_router(events_router)
app.include_router(known_objects_router)
app.include_router(settings_router)

# Mount static files directory
static_dir = Path(__file__).parent / "static"
static_dir.mkdir(exist_ok=True)
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {
        "status": "ok",
        "version": "2.0.0",
        "mqtt_connected": mqtt_connected,
        "events_processed": events_processed,
        "last_event": last_event_time,
    }


@app.get("/")
async def index():
    """Serve the web UI index page."""
    index_path = static_dir / "index.html"
    if index_path.exists():
        return FileResponse(str(index_path))
    return JSONResponse(
        {"message": "frigate-notify v2", "docs": "/docs"},
        status_code=200,
    )


# --- Entrypoint ---
if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host=config.SERVER_HOST,
        port=config.SERVER_PORT,
        log_level="info",
    )