diff --git a/frigate/api/chat.py b/frigate/api/chat.py index eeff3ab6d..ada7c91c3 100644 --- a/frigate/api/chat.py +++ b/frigate/api/chat.py @@ -1,10 +1,12 @@ """Chat and LLM tool calling APIs.""" +import base64 import json import logging from datetime import datetime, timezone from typing import Any, Dict, List +import cv2 from fastapi import APIRouter, Body, Depends, Request from fastapi.responses import JSONResponse from pydantic import BaseModel @@ -87,6 +89,28 @@ def get_tool_definitions() -> List[Dict[str, Any]]: "required": [], }, }, + { + "type": "function", + "function": { + "name": "get_live_context", + "description": ( + "Get the current live view and detection information for a camera. " + "Returns the current camera frame as a base64-encoded image along with " + "information about objects currently being tracked/detected on the camera. " + "Use this to answer questions about what is happening right now on a specific camera." + ), + "parameters": { + "type": "object", + "properties": { + "camera": { + "type": "string", + "description": "Camera name to get live context for.", + }, + }, + "required": ["camera"], + }, + }, + }, ] @@ -207,6 +231,69 @@ async def execute_tool( ) +async def _execute_get_live_context( + request: Request, + camera: str, + allowed_cameras: List[str], +) -> Dict[str, Any]: + if camera not in allowed_cameras: + return { + "error": f"Camera '{camera}' not found or access denied", + } + + if camera not in request.app.frigate_config.cameras: + return { + "error": f"Camera '{camera}' not found", + } + + try: + frame_processor = request.app.detected_frames_processor + camera_state = frame_processor.camera_states.get(camera) + + if camera_state is None: + return { + "error": f"Camera '{camera}' state not available", + } + + frame = frame_processor.get_current_frame(camera, {}) + if frame is None: + return { + "error": f"Unable to get current frame for camera '{camera}'", + } + + _, img_encoded = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 85]) + image_base64 = base64.b64encode(img_encoded.tobytes()).decode("utf-8") + image_data_url = f"data:image/jpeg;base64,{image_base64}" + + tracked_objects_dict = {} + with camera_state.current_frame_lock: + tracked_objects = camera_state.tracked_objects.copy() + frame_time = camera_state.current_frame_time + + for obj_id, tracked_obj in tracked_objects.items(): + obj_dict = tracked_obj.to_dict() + if obj_dict.get("frame_time") == frame_time: + tracked_objects_dict[obj_id] = { + "label": obj_dict.get("label"), + "zones": obj_dict.get("current_zones", []), + "sub_label": obj_dict.get("sub_label"), + "stationary": obj_dict.get("stationary", False), + } + + return { + "camera": camera, + "timestamp": frame_time, + "image": image_data_url, + "detections": list(tracked_objects_dict.values()), + } + + except Exception as e: + logger.error(f"Error executing get_live_context: {e}", exc_info=True) + return { + "error": f"Error getting live context: {str(e)}", + } + + async def _execute_tool_internal( tool_name: str, arguments: Dict[str, Any], @@ -231,6 +318,11 @@ async def _execute_tool_internal( except (json.JSONDecodeError, AttributeError) as e: logger.warning(f"Failed to extract tool result: {e}") return {"error": "Failed to parse tool result"} + elif tool_name == "get_live_context": + camera = arguments.get("camera") + if not camera: + return {"error": "Camera parameter is required"} + return await _execute_get_live_context(request, camera, allowed_cameras) else: return {"error": f"Unknown tool: {tool_name}"} @@ -277,13 +369,35 @@ async def chat_completion( current_datetime = datetime.now(timezone.utc) current_date_str = current_datetime.strftime("%Y-%m-%d") current_time_str = current_datetime.strftime("%H:%M:%S %Z") + + cameras_info = [] + config = request.app.frigate_config + for camera_id in allowed_cameras: + if camera_id not in config.cameras: + continue + camera_config = config.cameras[camera_id] + friendly_name = ( + camera_config.friendly_name + if camera_config.friendly_name + else camera_id.replace("_", " ").title() + ) + cameras_info.append(f" - {friendly_name} (ID: {camera_id})") + + cameras_section = "" + if cameras_info: + cameras_section = ( + "\n\nAvailable cameras:\n" + + "\n".join(cameras_info) + + "\n\nWhen users refer to cameras by their friendly name (e.g., 'Back Deck Camera'), use the corresponding camera ID (e.g., 'back_deck_cam') in tool calls." + ) + system_prompt = f"""You are a helpful assistant for Frigate, a security camera NVR system. You help users answer questions about their cameras, detected objects, and events. Current date and time: {current_date_str} at {current_time_str} (UTC) When users ask questions about "today", "yesterday", "this week", etc., use the current date above as reference. When searching for objects or events, use ISO 8601 format for dates (e.g., {current_date_str}T00:00:00Z for the start of today). -Always be accurate with time calculations based on the current date provided.""" +Always be accurate with time calculations based on the current date provided.{cameras_section}""" conversation.append( {