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 <noreply@anthropic.com>

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
 
 

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)
+               (frigate_event_id, camera, label, score, snapshot, match_status, ai_description, notified)
-               VALUES (?, ?, ?, ?, ?, ?, 1)""",
+               VALUES (?, ?, ?, ?, ?, ?, ?, 1)""",
-            (event_id, camera, label, score, snapshot, description),
+            (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

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: