import base64 import logging from typing import Optional import anthropic from ollama_client import OllamaClient logger = logging.getLogger(__name__) PROMPT = ( "Briefly describe what's happening in this security camera image in 1-2 sentences. " "Focus on the people, vehicles, or animals and their activity. Be concise." ) def analyze_snapshot(client: anthropic.Anthropic, image_bytes: bytes) -> str | None: """Send a snapshot to Claude Haiku for description. Returns None on failure.""" try: b64 = base64.standard_b64encode(image_bytes).decode("utf-8") response = client.messages.create( model="claude-haiku-4-5-20251001", max_tokens=150, messages=[ { "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/jpeg", "data": b64, }, }, {"type": "text", "text": PROMPT}, ], } ], ) description = response.content[0].text.strip() logger.info("Vision analysis complete: %d chars", len(description)) return description 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