feat: frigate-notify v2 (modular app, CLIP recognition) + clip-service
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>
This commit is contained in:
@@ -1,3 +1,5 @@
|
||||
"""frigate-notify v2 — FastAPI application with MQTT listener."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
@@ -6,29 +8,28 @@ import signal
|
||||
import sys
|
||||
import time
|
||||
from collections import defaultdict
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
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 vision import analyze_snapshot
|
||||
from config import config
|
||||
from database import close_connection, get_connection, init_db, set_db_path
|
||||
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
|
||||
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(
|
||||
@@ -44,46 +45,12 @@ 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)
|
||||
mqtt_client_instance: mqtt.Client | None = None
|
||||
|
||||
|
||||
# --- Snapshot fetching ---
|
||||
async def fetch_snapshot(event_id: str) -> bytes | None:
|
||||
url = f"{FRIGATE_URL}/api/events/{event_id}/snapshot.jpg"
|
||||
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:
|
||||
@@ -103,7 +70,8 @@ async def fetch_snapshot(event_id: str) -> bytes | None:
|
||||
|
||||
# --- Event processing ---
|
||||
async def process_event(
|
||||
event_data: dict, claude: anthropic.Anthropic, matrix: MatrixClient
|
||||
event_data: dict, claude: anthropic.Anthropic, matrix: MatrixClient,
|
||||
ollama_client: OllamaClient = None,
|
||||
):
|
||||
global events_processed, last_event_time
|
||||
|
||||
@@ -119,27 +87,30 @@ async def process_event(
|
||||
has_snapshot = after.get("has_snapshot", False)
|
||||
|
||||
# --- Filtering ---
|
||||
if label not in ALLOWED_LABELS:
|
||||
if label not in config.ALLOWED_LABELS:
|
||||
return
|
||||
if score < MIN_SCORE:
|
||||
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] < DEDUP_WINDOW_SECONDS:
|
||||
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] < RATE_LIMIT_SECONDS:
|
||||
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 - DEDUP_WINDOW_SECONDS * 2
|
||||
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]
|
||||
|
||||
@@ -150,18 +121,84 @@ async def process_event(
|
||||
# --- Fetch snapshot ---
|
||||
snapshot = await fetch_snapshot(event_id)
|
||||
|
||||
# --- AI analysis ---
|
||||
# --- 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 = analyze_snapshot(claude, 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"{FRIGATE_URL}/review?id={event_id}"
|
||||
frigate_link = f"{config.FRIGATE_URL}/review?id={event_id}"
|
||||
|
||||
plain = (
|
||||
f"{camera} - {label.capitalize()} - {score_pct}%\n"
|
||||
@@ -195,7 +232,8 @@ 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):
|
||||
@@ -213,7 +251,7 @@ def create_mqtt_client(
|
||||
try:
|
||||
payload = json.loads(msg.payload)
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
process_event(payload, claude, matrix), loop
|
||||
process_event(payload, claude, matrix, ollama_client), loop
|
||||
)
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("Invalid JSON on %s", msg.topic)
|
||||
@@ -227,33 +265,99 @@ def create_mqtt_client(
|
||||
return client
|
||||
|
||||
|
||||
# --- Main ---
|
||||
def main():
|
||||
logger.info("Starting frigate-notify")
|
||||
def start_mqtt(loop: asyncio.AbstractEventLoop) -> None:
|
||||
"""Start MQTT client in a background thread."""
|
||||
global mqtt_client_instance
|
||||
|
||||
claude = anthropic.Anthropic()
|
||||
matrix = MatrixClient(MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID)
|
||||
matrix = MatrixClient(
|
||||
config.MATRIX_HOMESERVER,
|
||||
config.MATRIX_ACCESS_TOKEN,
|
||||
config.MATRIX_ROOM_ID,
|
||||
)
|
||||
ollama_client = OllamaClient(config.OLLAMA_URL, config.OLLAMA_MODEL)
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
loop_thread = Thread(target=loop.run_forever, daemon=True)
|
||||
loop_thread.start()
|
||||
mqtt_client_instance = create_mqtt_client(loop, claude, matrix, ollama_client)
|
||||
mqtt_client_instance.connect(config.MQTT_HOST, config.MQTT_PORT)
|
||||
|
||||
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()
|
||||
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__":
|
||||
main()
|
||||
uvicorn.run(
|
||||
"main:app",
|
||||
host=config.SERVER_HOST,
|
||||
port=config.SERVER_PORT,
|
||||
log_level="info",
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user