Add live context tool

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(
         {