jeremy
dddc6734d0
Splits the single main.py into config/database/models/routes/vision/recognition/ ollama_client/cleanup modules, adds a web UI, and introduces clip-service for CLIP (ViT-L-14) embeddings. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
364 lines
11 KiB
Python
364 lines
11 KiB
Python
"""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> · {label.capitalize()} · {score_pct}%<br/>"
|
|
f"{description}<br/><br/>"
|
|
f'<a href="{frigate_link}">View in Frigate</a><br/>'
|
|
f"<small>{ts} · {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",
|
|
)
|