extract pure chat helpers to chat_util module

frigate/api/chat.py

@@ -3,12 +3,11 @@
 import base64
 import json
 import logging
-import math
 import operator
 import time
 from datetime import datetime
 from functools import reduce
-from typing import Any, Dict, Generator, List, Optional
+from typing import Any, Dict, List, Optional
 
 import cv2
 from fastapi import APIRouter, Body, Depends, Request
@@ -20,6 +19,14 @@ from frigate.api.auth import (
     get_allowed_cameras_for_filter,
     require_camera_access,
 )
+from frigate.api.chat_util import (
+    chunk_content,
+    distance_to_score,
+    format_events_with_local_time,
+    fuse_scores,
+    hydrate_event,
+    parse_iso_to_timestamp,
+)
 from frigate.api.defs.query.events_query_parameters import EventsQueryParams
 from frigate.api.defs.request.chat_body import ChatCompletionRequest
 from frigate.api.defs.response.chat_response import (
@@ -29,7 +36,6 @@ from frigate.api.defs.response.chat_response import (
 )
 from frigate.api.defs.tags import Tags
 from frigate.api.event import events
-from frigate.embeddings.util import ZScoreNormalization
 from frigate.genai.utils import build_assistant_message_for_conversation
 from frigate.jobs.vlm_watch import (
     get_vlm_watch_job,
@@ -43,49 +49,6 @@ logger = logging.getLogger(__name__)
 router = APIRouter(tags=[Tags.chat])
 
 
-def _chunk_content(content: str, chunk_size: int = 80) -> Generator[str, None, None]:
-    """Yield content in word-aware chunks for streaming."""
-    if not content:
-        return
-    words = content.split(" ")
-    current: List[str] = []
-    current_len = 0
-    for w in words:
-        current.append(w)
-        current_len += len(w) + 1
-        if current_len >= chunk_size:
-            yield " ".join(current) + " "
-            current = []
-            current_len = 0
-    if current:
-        yield " ".join(current)
-
-
-def _format_events_with_local_time(
-    events_list: List[Dict[str, Any]],
-) -> List[Dict[str, Any]]:
-    """Add human-readable local start/end times to each event for the LLM."""
-    result = []
-    for evt in events_list:
-        if not isinstance(evt, dict):
-            result.append(evt)
-            continue
-        copy_evt = dict(evt)
-        try:
-            start_ts = evt.get("start_time")
-            end_ts = evt.get("end_time")
-            if start_ts is not None:
-                dt_start = datetime.fromtimestamp(start_ts)
-                copy_evt["start_time_local"] = dt_start.strftime("%Y-%m-%d %I:%M:%S %p")
-            if end_ts is not None:
-                dt_end = datetime.fromtimestamp(end_ts)
-                copy_evt["end_time_local"] = dt_end.strftime("%Y-%m-%d %I:%M:%S %p")
-        except (TypeError, ValueError, OSError):
-            pass
-        result.append(copy_evt)
-    return result
-
-
 class ToolExecuteRequest(BaseModel):
     """Request model for tool execution."""
 
@@ -103,47 +66,6 @@ class VLMMonitorRequest(BaseModel):
     zones: List[str] = []
 
 
-# Similarity fusion weights for find_similar_objects.
-# Visual dominates because the feature's primary use case is "same specific object."
-# If these change, update the test in test_chat_find_similar_objects.py.
-VISUAL_WEIGHT = 0.65
-DESCRIPTION_WEIGHT = 0.35
-
-
-def _distance_to_score(distance: float, stats: ZScoreNormalization) -> float:
-    """Convert a cosine distance to a [0, 1] similarity score.
-
-    Uses the existing ZScoreNormalization stats maintained by EmbeddingsContext
-    to normalize across deployments, then a bounded sigmoid. Lower distance ->
-    higher score. If stats are uninitialized (stddev == 0), returns a neutral
-    0.5 so the fallback ordering by raw distance still dominates.
-    """
-    if stats.stddev == 0:
-        return 0.5
-    z = (distance - stats.mean) / stats.stddev
-    # Sigmoid on -z so that small distance (good) -> high score.
-    return 1.0 / (1.0 + math.exp(z))
-
-
-def _fuse_scores(
-    visual_score: Optional[float],
-    description_score: Optional[float],
-) -> Optional[float]:
-    """Weighted fusion of visual and description similarity scores.
-
-    If one side is missing (e.g., no description embedding for this event),
-    the other side's score is returned alone with no penalty. If both are
-    missing, returns None and the caller should drop the event.
-    """
-    if visual_score is None and description_score is None:
-        return None
-    if visual_score is None:
-        return description_score
-    if description_score is None:
-        return visual_score
-    return VISUAL_WEIGHT * visual_score + DESCRIPTION_WEIGHT * description_score
-
-
 def get_tool_definitions() -> List[Dict[str, Any]]:
     """
     Get OpenAI-compatible tool definitions for Frigate.
@@ -550,39 +472,6 @@ async def _execute_search_objects(
         )
 
 
-def _parse_iso_to_timestamp(value: Optional[str]) -> Optional[float]:
-    """Parse an ISO-8601 string as server-local time -> unix timestamp.
-
-    Mirrors the parsing _execute_search_objects uses so both tools accept the
-    same format from the LLM.
-    """
-    if value is None:
-        return None
-    try:
-        s = value.replace("Z", "").strip()[:19]
-        dt = datetime.strptime(s, "%Y-%m-%dT%H:%M:%S")
-        return time.mktime(dt.timetuple())
-    except (ValueError, AttributeError, TypeError):
-        logger.warning("Invalid timestamp format: %s", value)
-        return None
-
-
-def _hydrate_event(event: Event, score: Optional[float] = None) -> Dict[str, Any]:
-    """Convert an Event row into the dict shape returned by find_similar_objects."""
-    data: Dict[str, Any] = {
-        "id": event.id,
-        "camera": event.camera,
-        "label": event.label,
-        "sub_label": event.sub_label,
-        "start_time": event.start_time,
-        "end_time": event.end_time,
-        "zones": event.zones,
-    }
-    if score is not None:
-        data["score"] = score
-    return data
-
-
 async def _execute_find_similar_objects(
     request: Request,
     arguments: Dict[str, Any],
@@ -625,8 +514,8 @@ async def _execute_find_similar_objects(
         }
 
     # 3. Parse params.
-    after = _parse_iso_to_timestamp(arguments.get("after"))
+    after = parse_iso_to_timestamp(arguments.get("after"))
-    before = _parse_iso_to_timestamp(arguments.get("before"))
+    before = parse_iso_to_timestamp(arguments.get("before"))
 
     cameras = arguments.get("cameras")
     if cameras:
@@ -685,7 +574,7 @@ async def _execute_find_similar_objects(
 
     if not vec_ids:
         return {
-            "anchor": _hydrate_event(anchor),
+            "anchor": hydrate_event(anchor),
             "results": [],
             "similarity_mode": similarity_mode,
             "candidate_truncated": candidate_truncated,
@@ -714,16 +603,16 @@ async def _execute_find_similar_objects(
     scored: List[tuple[str, float]] = []
     for eid in eligible:
         v_score = (
-            _distance_to_score(visual_distances[eid], context.thumb_stats)
+            distance_to_score(visual_distances[eid], context.thumb_stats)
             if eid in visual_distances
             else None
         )
         d_score = (
-            _distance_to_score(description_distances[eid], context.desc_stats)
+            distance_to_score(description_distances[eid], context.desc_stats)
             if eid in description_distances
             else None
         )
-        fused = _fuse_scores(v_score, d_score)
+        fused = fuse_scores(v_score, d_score)
         if fused is None:
             continue
         if min_score is not None and fused < min_score:
@@ -733,10 +622,10 @@ async def _execute_find_similar_objects(
     scored.sort(key=lambda pair: pair[1], reverse=True)
     scored = scored[:limit]
 
-    results = [_hydrate_event(eligible[eid], score=score) for eid, score in scored]
+    results = [hydrate_event(eligible[eid], score=score) for eid, score in scored]
 
     return {
-        "anchor": _hydrate_event(anchor),
+        "anchor": hydrate_event(anchor),
         "results": results,
         "similarity_mode": similarity_mode,
         "candidate_truncated": candidate_truncated,
@@ -1246,7 +1135,7 @@ async def _execute_pending_tools(
                     json.dumps(tool_args),
                 )
             if tool_name == "search_objects" and isinstance(tool_result, list):
-                tool_result = _format_events_with_local_time(tool_result)
+                tool_result = format_events_with_local_time(tool_result)
                 _keys = {
                     "id",
                     "camera",
@@ -1561,7 +1450,7 @@ When a user refers to a specific object they have seen or describe with identify
                                 + b"\n"
                             )
                         # Stream content in word-sized chunks for smooth UX
-                        for part in _chunk_content(final_content):
+                        for part in chunk_content(final_content):
                             yield (
                                 json.dumps({"type": "content", "delta": part}).encode(
                                     "utf-8"

frigate/api/chat_util.py

@@ -0,0 +1,135 @@
+"""Pure, stateless helpers used by the chat tool dispatchers.
+
+These were extracted from frigate/api/chat.py to keep that module focused on
+route handlers, tool dispatchers, and streaming loop internals. Nothing in
+this file touches the FastAPI request, the embeddings context, or the chat
+loop state — all inputs and outputs are plain data.
+"""
+
+import logging
+import math
+import time
+from datetime import datetime
+from typing import Any, Dict, Generator, List, Optional
+
+from frigate.embeddings.util import ZScoreNormalization
+from frigate.models import Event
+
+logger = logging.getLogger(__name__)
+
+
+# Similarity fusion weights for find_similar_objects.
+# Visual dominates because the feature's primary use case is "same specific object."
+# If these change, update the test in test_chat_find_similar_objects.py.
+VISUAL_WEIGHT = 0.65
+DESCRIPTION_WEIGHT = 0.35
+
+
+def chunk_content(content: str, chunk_size: int = 80) -> Generator[str, None, None]:
+    """Yield content in word-aware chunks for streaming."""
+    if not content:
+        return
+    words = content.split(" ")
+    current: List[str] = []
+    current_len = 0
+    for w in words:
+        current.append(w)
+        current_len += len(w) + 1
+        if current_len >= chunk_size:
+            yield " ".join(current) + " "
+            current = []
+            current_len = 0
+    if current:
+        yield " ".join(current)
+
+
+def format_events_with_local_time(
+    events_list: List[Dict[str, Any]],
+) -> List[Dict[str, Any]]:
+    """Add human-readable local start/end times to each event for the LLM."""
+    result = []
+    for evt in events_list:
+        if not isinstance(evt, dict):
+            result.append(evt)
+            continue
+        copy_evt = dict(evt)
+        try:
+            start_ts = evt.get("start_time")
+            end_ts = evt.get("end_time")
+            if start_ts is not None:
+                dt_start = datetime.fromtimestamp(start_ts)
+                copy_evt["start_time_local"] = dt_start.strftime("%Y-%m-%d %I:%M:%S %p")
+            if end_ts is not None:
+                dt_end = datetime.fromtimestamp(end_ts)
+                copy_evt["end_time_local"] = dt_end.strftime("%Y-%m-%d %I:%M:%S %p")
+        except (TypeError, ValueError, OSError):
+            pass
+        result.append(copy_evt)
+    return result
+
+
+def distance_to_score(distance: float, stats: ZScoreNormalization) -> float:
+    """Convert a cosine distance to a [0, 1] similarity score.
+
+    Uses the existing ZScoreNormalization stats maintained by EmbeddingsContext
+    to normalize across deployments, then a bounded sigmoid. Lower distance ->
+    higher score. If stats are uninitialized (stddev == 0), returns a neutral
+    0.5 so the fallback ordering by raw distance still dominates.
+    """
+    if stats.stddev == 0:
+        return 0.5
+    z = (distance - stats.mean) / stats.stddev
+    # Sigmoid on -z so that small distance (good) -> high score.
+    return 1.0 / (1.0 + math.exp(z))
+
+
+def fuse_scores(
+    visual_score: Optional[float],
+    description_score: Optional[float],
+) -> Optional[float]:
+    """Weighted fusion of visual and description similarity scores.
+
+    If one side is missing (e.g., no description embedding for this event),
+    the other side's score is returned alone with no penalty. If both are
+    missing, returns None and the caller should drop the event.
+    """
+    if visual_score is None and description_score is None:
+        return None
+    if visual_score is None:
+        return description_score
+    if description_score is None:
+        return visual_score
+    return VISUAL_WEIGHT * visual_score + DESCRIPTION_WEIGHT * description_score
+
+
+def parse_iso_to_timestamp(value: Optional[str]) -> Optional[float]:
+    """Parse an ISO-8601 string as server-local time -> unix timestamp.
+
+    Mirrors the parsing _execute_search_objects uses so both tools accept the
+    same format from the LLM.
+    """
+    if value is None:
+        return None
+    try:
+        s = value.replace("Z", "").strip()[:19]
+        dt = datetime.strptime(s, "%Y-%m-%dT%H:%M:%S")
+        return time.mktime(dt.timetuple())
+    except (ValueError, AttributeError, TypeError):
+        logger.warning("Invalid timestamp format: %s", value)
+        return None
+
+
+def hydrate_event(event: Event, score: Optional[float] = None) -> Dict[str, Any]:
+    """Convert an Event row into the dict shape returned by find_similar_objects."""
+    data: Dict[str, Any] = {
+        "id": event.id,
+        "camera": event.camera,
+        "label": event.label,
+        "sub_label": event.sub_label,
+        "start_time": event.start_time,
+        "end_time": event.end_time,
+        "zones": event.zones,
+    }
+    if score is not None:
+        data["score"] = score
+    return data

frigate/test/test_chat_find_similar_objects.py

@@ -10,12 +10,14 @@ from unittest.mock import MagicMock
 from playhouse.sqlite_ext import SqliteExtDatabase
 
 from frigate.api.chat import (
+    _execute_find_similar_objects,
+    get_tool_definitions,
+)
+from frigate.api.chat_util import (
     DESCRIPTION_WEIGHT,
     VISUAL_WEIGHT,
-    _distance_to_score,
+    distance_to_score,
-    _execute_find_similar_objects,
+    fuse_scores,
-    _fuse_scores,
-    get_tool_definitions,
 )
 from frigate.embeddings.util import ZScoreNormalization
 from frigate.models import Event
@@ -31,8 +33,8 @@ class TestDistanceToScore(unittest.TestCase):
         # Seed the stats with a small distribution so stddev > 0.
         stats._update([0.1, 0.2, 0.3, 0.4, 0.5])
 
-        close_score = _distance_to_score(0.1, stats)
+        close_score = distance_to_score(0.1, stats)
-        far_score = _distance_to_score(0.5, stats)
+        far_score = distance_to_score(0.5, stats)
 
         self.assertGreater(close_score, far_score)
         self.assertGreaterEqual(close_score, 0.0)
@@ -42,7 +44,7 @@ class TestDistanceToScore(unittest.TestCase):
 
     def test_uninitialized_stats_returns_neutral_score(self):
         stats = ZScoreNormalization()  # n == 0, stddev == 0
-        self.assertEqual(_distance_to_score(0.3, stats), 0.5)
+        self.assertEqual(distance_to_score(0.3, stats), 0.5)
 
 
 class TestFuseScores(unittest.TestCase):
@@ -50,20 +52,20 @@ class TestFuseScores(unittest.TestCase):
         self.assertAlmostEqual(VISUAL_WEIGHT + DESCRIPTION_WEIGHT, 1.0)
 
     def test_fuses_both_sides(self):
-        fused = _fuse_scores(visual_score=0.8, description_score=0.4)
+        fused = fuse_scores(visual_score=0.8, description_score=0.4)
         expected = VISUAL_WEIGHT * 0.8 + DESCRIPTION_WEIGHT * 0.4
         self.assertAlmostEqual(fused, expected)
 
     def test_missing_description_uses_visual_only(self):
-        fused = _fuse_scores(visual_score=0.7, description_score=None)
+        fused = fuse_scores(visual_score=0.7, description_score=None)
         self.assertAlmostEqual(fused, 0.7)
 
     def test_missing_visual_uses_description_only(self):
-        fused = _fuse_scores(visual_score=None, description_score=0.6)
+        fused = fuse_scores(visual_score=None, description_score=0.6)
         self.assertAlmostEqual(fused, 0.6)
 
     def test_both_missing_returns_none(self):
-        self.assertIsNone(_fuse_scores(visual_score=None, description_score=None))
+        self.assertIsNone(fuse_scores(visual_score=None, description_score=None))
 
 
 class TestToolDefinition(unittest.TestCase):