From f5f688277f4df99af47765cbd1249c7584ebf768 Mon Sep 17 00:00:00 2001 From: Pluto Date: Sat, 7 Mar 2026 17:23:54 -0600 Subject: [PATCH] feat: integrate recognition engine into MQTT event pipeline - Recognition pipeline runs between snapshot fetch and notification - Checks recognition_enabled setting before running CLIP matching - Known objects with quiet/suppress action skip Matrix notifications - Vision description uses configurable backend (haiku/ollama/both) - docker-compose.yml adds CLIP/Ollama env vars and data volume Co-Authored-By: Claude Opus 4.6 --- app/database.py | 3 +- app/main.py | 80 ++++++++++++++++++++++++++++++++++++++++------ docker-compose.yml | 6 ++++ 3 files changed, 79 insertions(+), 10 deletions(-) diff --git a/app/database.py b/app/database.py index d170c37..a768746 100644 --- a/app/database.py +++ b/app/database.py @@ -13,8 +13,9 @@ _db_path: str = "/data/frigate-notify.db" def set_db_path(path: str) -> None: - """Set the database path. Must be called before any other database operations.""" + """Set the database path. Closes any existing connection first.""" global _db_path + close_connection() _db_path = path diff --git a/app/main.py b/app/main.py index 5ee8612..34e6172 100755 --- a/app/main.py +++ b/app/main.py @@ -26,7 +26,10 @@ 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 vision import analyze_snapshot +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( @@ -67,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 @@ -117,10 +121,63 @@ 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}." @@ -129,9 +186,9 @@ async def process_event( conn = get_connection() conn.execute( """INSERT OR IGNORE INTO events - (frigate_event_id, camera, label, score, snapshot, ai_description, notified) - VALUES (?, ?, ?, ?, ?, ?, 1)""", - (event_id, camera, label, score, snapshot, description), + (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: @@ -175,6 +232,7 @@ def create_mqtt_client( loop: asyncio.AbstractEventLoop, claude: anthropic.Anthropic, matrix: MatrixClient, + ollama_client: OllamaClient = None, ) -> mqtt.Client: global mqtt_connected @@ -193,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) @@ -217,8 +275,9 @@ def start_mqtt(loop: asyncio.AbstractEventLoop) -> None: 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) + 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) @@ -240,6 +299,9 @@ async def lifespan(app: FastAPI): loop = asyncio.get_event_loop() start_mqtt(loop) + # Start periodic event cleanup + asyncio.create_task(cleanup_old_events()) + yield # Shutdown diff --git a/docker-compose.yml b/docker-compose.yml index 8dfef6b..c2d3b56 100755 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,6 +15,12 @@ services: - MQTT_HOST=192.168.86.192 - MQTT_PORT=1883 - MATRIX_HOMESERVER=http://192.168.86.192:8008 + - CLIP_SERVICE_URL=http://192.168.86.196:5200 + - OLLAMA_URL=http://192.168.86.196:11434 + - OLLAMA_MODEL=llava:13b + - DB_PATH=/data/frigate-notify.db + volumes: + - /mnt/docker-data/configs/frigate-notify:/data mem_limit: 256M memswap_limit: 256M healthcheck: