From 2df137ee2a101138f4070eed3641e877b2546f71 Mon Sep 17 00:00:00 2001 From: Pluto Date: Sat, 7 Mar 2026 17:17:49 -0600 Subject: [PATCH] feat: add recognition engine with CLIP matching and Ollama vision LLM - OllamaClient: analyze_image, compare_images, is_available - Recognition pipeline: CLIP embedding -> cosine similarity -> threshold routing (known >0.85, unknown <0.60, ambiguous -> Ollama fallback) - Vision dispatcher: get_description() supports haiku/ollama/both backends - 9 embedding serialization tests Co-Authored-By: Claude Opus 4.6 --- app/ollama_client.py | 132 ++++++++++++++++++++++ app/recognition.py | 232 ++++++++++++++++++++++++++++++++++++++ app/vision.py | 67 +++++++++++ tests/test_recognition.py | 82 ++++++++++++++ 4 files changed, 513 insertions(+) create mode 100644 app/ollama_client.py create mode 100644 app/recognition.py create mode 100644 tests/test_recognition.py diff --git a/app/ollama_client.py b/app/ollama_client.py new file mode 100644 index 0000000..4d68c68 --- /dev/null +++ b/app/ollama_client.py @@ -0,0 +1,132 @@ +"""Ollama vision LLM client for frigate-notify v2.""" + +import base64 +import logging + +import aiohttp + +logger = logging.getLogger(__name__) + + +class OllamaClient: + """Client for Ollama vision models (e.g., llava).""" + + def __init__(self, base_url: str, model: str): + self.base_url = base_url.rstrip("/") + self.model = model + + async def analyze_image(self, image_bytes: bytes, prompt: str) -> str | None: + """Send an image to Ollama for analysis. + + Args: + image_bytes: Raw image bytes (JPEG). + prompt: Text prompt for the vision model. + + Returns: + Response text or None on failure. + """ + b64 = base64.b64encode(image_bytes).decode("utf-8") + payload = { + "model": self.model, + "prompt": prompt, + "images": [b64], + "stream": False, + "options": {"num_predict": 200}, + } + try: + async with aiohttp.ClientSession() as session: + async with session.post( + f"{self.base_url}/api/generate", + json=payload, + timeout=aiohttp.ClientTimeout(total=30), + ) as resp: + if resp.status == 200: + data = await resp.json() + text = data.get("response", "").strip() + logger.info( + "Ollama analysis complete: %d chars", len(text) + ) + return text if text else None + body = await resp.text() + logger.error( + "Ollama analyze_image failed %s: %s", resp.status, body + ) + except Exception: + logger.exception("Ollama analyze_image request failed") + return None + + async def compare_images( + self, + image_bytes: bytes, + reference_bytes: bytes, + object_name: str, + ) -> bool | None: + """Compare two images to determine if they show the same object/person. + + Args: + image_bytes: The new event image bytes. + reference_bytes: The reference (known) image bytes. + object_name: Name of the known object for context. + + Returns: + True if same, False if different, None on failure/inconclusive. + """ + b64_new = base64.b64encode(image_bytes).decode("utf-8") + b64_ref = base64.b64encode(reference_bytes).decode("utf-8") + prompt = ( + f"Look at these two images. The first image is from a security camera. " + f"The second image is a reference photo of '{object_name}'. " + f"Are these the same person/vehicle/animal? " + f"Answer with ONLY 'yes' or 'no'." + ) + payload = { + "model": self.model, + "prompt": prompt, + "images": [b64_new, b64_ref], + "stream": False, + "options": {"num_predict": 200}, + } + try: + async with aiohttp.ClientSession() as session: + async with session.post( + f"{self.base_url}/api/generate", + json=payload, + timeout=aiohttp.ClientTimeout(total=30), + ) as resp: + if resp.status == 200: + data = await resp.json() + text = data.get("response", "").strip().lower() + logger.info("Ollama compare result: %s", text) + if "yes" in text.split()[:3]: + return True + if "no" in text.split()[:3]: + return False + logger.warning( + "Ollama compare inconclusive: %s", text + ) + return None + body = await resp.text() + logger.error( + "Ollama compare_images failed %s: %s", + resp.status, + body, + ) + except Exception: + logger.exception("Ollama compare_images request failed") + return None + + async def is_available(self) -> bool: + """Check if Ollama is reachable and responding. + + Returns: + True if Ollama responds with HTTP 200, False otherwise. + """ + try: + async with aiohttp.ClientSession() as session: + async with session.get( + f"{self.base_url}/api/tags", + timeout=aiohttp.ClientTimeout(total=3), + ) as resp: + return resp.status == 200 + except Exception: + return False diff --git a/app/recognition.py b/app/recognition.py new file mode 100644 index 0000000..0099c26 --- /dev/null +++ b/app/recognition.py @@ -0,0 +1,232 @@ +"""Recognition pipeline using CLIP embeddings with Ollama vision fallback.""" + +import base64 +import logging +import struct +from typing import Optional + +import aiohttp +import numpy as np + +from config import config +from database import get_connection +from ollama_client import OllamaClient + +logger = logging.getLogger(__name__) + + +def embedding_to_bytes(embedding: list[float]) -> bytes: + """Pack a float32 embedding list into bytes for SQLite storage. + + Args: + embedding: List of float values (typically 768-dimensional). + + Returns: + Packed bytes (len(embedding) * 4 bytes). + """ + return struct.pack(f"{len(embedding)}f", *embedding) + + +def bytes_to_embedding(data: bytes) -> np.ndarray: + """Unpack bytes back into a float32 numpy array. + + Args: + data: Packed bytes from embedding_to_bytes(). + + Returns: + Numpy array of float32 values. + """ + count = len(data) // 4 + values = struct.unpack(f"{count}f", data) + return np.array(values, dtype=np.float32) + + +async def get_clip_embedding(image_bytes: bytes) -> list[float] | None: + """Get CLIP embedding for an image from the clip-service sidecar. + + Args: + image_bytes: Raw image bytes (JPEG). + + Returns: + List of float embedding values, or None on failure. + """ + b64 = base64.b64encode(image_bytes).decode("utf-8") + try: + async with aiohttp.ClientSession() as session: + async with session.post( + f"{config.CLIP_SERVICE_URL}/embed", + json={"image": b64}, + timeout=aiohttp.ClientTimeout(total=10), + ) as resp: + if resp.status == 200: + data = await resp.json() + embedding = data.get("embedding") + if embedding: + logger.info( + "CLIP embedding received: %d dimensions", + len(embedding), + ) + return embedding + logger.warning("CLIP response missing embedding field") + else: + body = await resp.text() + logger.error( + "CLIP embed failed %s: %s", resp.status, body + ) + except Exception: + logger.exception("CLIP embedding request failed") + return None + + +async def find_best_match( + query_embedding: list[float], +) -> tuple[int | None, float, str | None]: + """Find the best matching known object for a query embedding. + + Loads all embeddings from the database, computes cosine similarity, + groups by known_object_id and takes the max score per object. + + Args: + query_embedding: The query embedding to match against. + + Returns: + Tuple of (best_object_id, best_score, best_name). + Returns (None, 0.0, None) if no embeddings exist. + """ + conn = get_connection() + cursor = conn.execute( + """ + SELECT e.id, e.known_object_id, e.embedding, ko.name + FROM embeddings e + JOIN known_objects ko ON e.known_object_id = ko.id + """ + ) + rows = cursor.fetchall() + + if not rows: + return (None, 0.0, None) + + query_vec = np.array(query_embedding, dtype=np.float32) + query_norm = np.linalg.norm(query_vec) + if query_norm == 0: + return (None, 0.0, None) + query_vec = query_vec / query_norm + + # Group scores by known_object_id, keeping max per object + best_per_object: dict[int, tuple[float, str]] = {} + + for row in rows: + stored_vec = bytes_to_embedding(row["embedding"]) + stored_norm = np.linalg.norm(stored_vec) + if stored_norm == 0: + continue + stored_vec = stored_vec / stored_norm + + similarity = float(np.dot(query_vec, stored_vec)) + obj_id = row["known_object_id"] + obj_name = row["name"] + + if obj_id not in best_per_object or similarity > best_per_object[obj_id][0]: + best_per_object[obj_id] = (similarity, obj_name) + + if not best_per_object: + return (None, 0.0, None) + + best_id = max(best_per_object, key=lambda k: best_per_object[k][0]) + best_score, best_name = best_per_object[best_id] + + return (best_id, best_score, best_name) + + +async def recognize_event( + image_bytes: bytes, + ollama_client: Optional[OllamaClient] = None, +) -> tuple[str, int | None, float, str | None]: + """Full recognition pipeline for a Frigate event image. + + Steps: + 1. Get CLIP embedding for the image. + 2. Find the best matching known object. + 3. High confidence (>= threshold_high): return "known". + 4. Low confidence (< threshold_low): return "unknown". + 5. Ambiguous range: use Ollama vision LLM to confirm/deny. + 6. If LLM unavailable or inconclusive: return "ambiguous". + 7. If no embedding could be obtained: return "unreviewed". + + Args: + image_bytes: Raw image bytes (JPEG). + ollama_client: Optional OllamaClient for ambiguous case fallback. + + Returns: + Tuple of (match_status, object_id, score, object_name). + match_status is one of: "known", "unknown", "ambiguous", "unreviewed". + """ + # Step 1: Get CLIP embedding + embedding = await get_clip_embedding(image_bytes) + if embedding is None: + logger.warning("Could not get CLIP embedding, returning unreviewed") + return ("unreviewed", None, 0.0, None) + + # Step 2: Find best match + best_id, best_score, best_name = await find_best_match(embedding) + + # No embeddings in DB at all + if best_id is None: + return ("unknown", None, 0.0, None) + + high_threshold = config.RECOGNITION_THRESHOLD_HIGH + low_threshold = config.RECOGNITION_THRESHOLD_LOW + + # Step 3: High confidence match + if best_score >= high_threshold: + logger.info( + "High confidence match: %s (%.3f)", best_name, best_score + ) + return ("known", best_id, best_score, best_name) + + # Step 4: Low confidence - clearly not a match + if best_score < low_threshold: + logger.info("Low confidence, no match (%.3f)", best_score) + return ("unknown", None, best_score, None) + + # Step 5: Ambiguous range - try Ollama vision LLM + if ollama_client is not None: + logger.info( + "Ambiguous score %.3f for %s, trying Ollama comparison", + best_score, + best_name, + ) + # Get a reference thumbnail from the embeddings table + conn = get_connection() + cursor = conn.execute( + """ + SELECT thumbnail FROM embeddings + WHERE known_object_id = ? AND thumbnail IS NOT NULL + LIMIT 1 + """, + (best_id,), + ) + ref_row = cursor.fetchone() + if ref_row and ref_row["thumbnail"] is not None: + result = await ollama_client.compare_images( + image_bytes, ref_row["thumbnail"], best_name + ) + if result is True: + logger.info( + "Ollama confirmed match: %s (%.3f)", + best_name, + best_score, + ) + return ("known", best_id, best_score, best_name) + elif result is False: + logger.info( + "Ollama denied match: %s (%.3f)", best_name, best_score + ) + return ("unknown", None, best_score, None) + # result is None: inconclusive, fall through to ambiguous + + # Step 6: LLM unavailable or inconclusive + logger.info( + "Ambiguous match: %s (%.3f)", best_name, best_score + ) + return ("ambiguous", best_id, best_score, best_name) diff --git a/app/vision.py b/app/vision.py index 9540dc2..5d200fc 100755 --- a/app/vision.py +++ b/app/vision.py @@ -1,7 +1,11 @@ import base64 import logging +from typing import Optional + import anthropic +from ollama_client import OllamaClient + logger = logging.getLogger(__name__) PROMPT = ( @@ -40,3 +44,66 @@ def analyze_snapshot(client: anthropic.Anthropic, image_bytes: bytes) -> str | N except Exception: logger.exception("Claude Vision API failed") return None + + +async def analyze_snapshot_ollama( + ollama_client: OllamaClient, image_bytes: bytes +) -> str | None: + """Send a snapshot to Ollama vision model for description. + + Args: + ollama_client: Configured OllamaClient instance. + image_bytes: Raw image bytes (JPEG). + + Returns: + Description string or None on failure. + """ + return await ollama_client.analyze_image(image_bytes, PROMPT) + + +async def get_description( + image_bytes: bytes, + vision_backend: str, + claude_client: Optional[anthropic.Anthropic] = None, + ollama_client: Optional[OllamaClient] = None, +) -> str | None: + """Get a scene description using the configured vision backend. + + Args: + image_bytes: Raw image bytes (JPEG). + vision_backend: One of "haiku", "ollama", or "both". + "both" tries Ollama first, falls back to Haiku. + claude_client: Anthropic client (required for "haiku" and "both"). + ollama_client: OllamaClient instance (required for "ollama" and "both"). + + Returns: + Description string or None if all backends fail. + """ + if vision_backend == "ollama": + if ollama_client is None: + logger.error("Ollama backend selected but no client provided") + return None + return await analyze_snapshot_ollama(ollama_client, image_bytes) + + if vision_backend == "haiku": + if claude_client is None: + logger.error("Haiku backend selected but no client provided") + return None + return analyze_snapshot(claude_client, image_bytes) + + if vision_backend == "both": + # Try Ollama first, fall back to Haiku + if ollama_client is not None: + result = await analyze_snapshot_ollama(ollama_client, image_bytes) + if result is not None: + return result + logger.info("Ollama failed, falling back to Haiku") + + if claude_client is not None: + return analyze_snapshot(claude_client, image_bytes) + + logger.error("Both backend selected but no clients provided") + return None + + logger.error("Unknown vision backend: %s", vision_backend) + return None diff --git a/tests/test_recognition.py b/tests/test_recognition.py new file mode 100644 index 0000000..3437bf4 --- /dev/null +++ b/tests/test_recognition.py @@ -0,0 +1,82 @@ +"""Tests for the recognition module — embedding serialization helpers.""" + +import numpy as np +import pytest + +from recognition import bytes_to_embedding, embedding_to_bytes + + +class TestEmbeddingToBytes: + """Tests for embedding_to_bytes().""" + + def test_roundtrip_preserves_values(self): + """Packing and unpacking should return the same values.""" + original = [0.1, 0.2, 0.3, -0.5, 1.0, 0.0] + packed = embedding_to_bytes(original) + recovered = bytes_to_embedding(packed) + np.testing.assert_allclose(recovered, original, rtol=1e-6) + + def test_correct_byte_length_768(self): + """A 768-dim embedding should produce 768 * 4 = 3072 bytes.""" + embedding = [0.0] * 768 + packed = embedding_to_bytes(embedding) + assert len(packed) == 768 * 4 + + def test_correct_byte_length_arbitrary(self): + """Byte length should always be len(embedding) * 4.""" + for dim in [1, 10, 512, 1024]: + embedding = [1.0] * dim + packed = embedding_to_bytes(embedding) + assert len(packed) == dim * 4 + + def test_empty_embedding(self): + """An empty embedding should produce 0 bytes.""" + packed = embedding_to_bytes([]) + assert len(packed) == 0 + recovered = bytes_to_embedding(packed) + assert len(recovered) == 0 + + +class TestBytesToEmbedding: + """Tests for bytes_to_embedding().""" + + def test_correct_shape(self): + """Should return a 1-D array with the correct number of elements.""" + embedding = [0.5] * 768 + packed = embedding_to_bytes(embedding) + recovered = bytes_to_embedding(packed) + assert recovered.shape == (768,) + + def test_correct_dtype(self): + """Should return float32 numpy array.""" + embedding = [0.1, 0.2, 0.3] + packed = embedding_to_bytes(embedding) + recovered = bytes_to_embedding(packed) + assert recovered.dtype == np.float32 + + def test_negative_values(self): + """Negative values should survive roundtrip.""" + original = [-1.0, -0.5, 0.0, 0.5, 1.0] + packed = embedding_to_bytes(original) + recovered = bytes_to_embedding(packed) + np.testing.assert_allclose(recovered, original, rtol=1e-6) + + def test_large_values(self): + """Large float values should survive roundtrip within float32 precision.""" + original = [1e10, -1e10, 3.14159, 2.71828] + packed = embedding_to_bytes(original) + recovered = bytes_to_embedding(packed) + np.testing.assert_allclose(recovered, original, rtol=1e-5) + + def test_realistic_clip_embedding(self): + """Simulate a realistic CLIP embedding (768-dim, normalized).""" + rng = np.random.default_rng(42) + vec = rng.standard_normal(768).astype(np.float32) + vec = vec / np.linalg.norm(vec) # normalize + + packed = embedding_to_bytes(vec.tolist()) + recovered = bytes_to_embedding(packed) + + assert recovered.shape == (768,) + assert recovered.dtype == np.float32 + np.testing.assert_allclose(recovered, vec, rtol=1e-6)