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>
233 lines
7.5 KiB
Python
233 lines
7.5 KiB
Python
"""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)
|