Enable event snapshot API to honour query params after event ends (#22375)

* Enable event snapshot API to honour query params

* fix unused imports

* Fixes

* Run ruff check --fix

* Web changes

* Further config and web fixes

* Further docs tweak

* Fix missing quality default in MediaEventsSnapshotQueryParams

* Manual events: don't save annotated jpeg; store frame time

* Remove unnecessary grayscale helper

* Add caveat to docs on snapshot_frame_time pre-0.18

* JPG snapshot should not be treated as clean

* Ensure tracked details uses uncropped, bbox'd snapshot

* Ensure all UI pages / menu actions use uncropped, bbox'd

* web lint

* Add missed config helper text

* Expect  SnapshotsConfig not Any

* docs: Remove pre-0.18 note

* Specify timestamp=0 in the UI

* Move tests out of http media

* Correct missed settings.json wording

Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>

* Revert to default None for quality

* Correct camera snapshot config wording

Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>

* Fix quality=0 handling

Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>

* Fix quality=0 handling #2

Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>

* ReRun generate_config_translations

---------

Co-authored-by: leccelecce <example@example.com>
Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>

frigate/util/camera_cleanup.py

@@ -133,6 +133,18 @@ def cleanup_camera_files(
         except Exception as e:
             logger.error("Failed to remove snapshot %s: %s", snapshot, e)
 
+    for snapshot in glob.glob(os.path.join(CLIPS_DIR, f"{camera_name}-*-clean.webp")):
+        try:
+            os.remove(snapshot)
+        except Exception as e:
+            logger.error("Failed to remove snapshot %s: %s", snapshot, e)
+
+    for snapshot in glob.glob(os.path.join(CLIPS_DIR, f"{camera_name}-*-clean.png")):
+        try:
+            os.remove(snapshot)
+        except Exception as e:
+            logger.error("Failed to remove snapshot %s: %s", snapshot, e)
+
     # Remove review thumbnail files
     for thumb in glob.glob(
         os.path.join(CLIPS_DIR, "review", f"thumb-{camera_name}-*.webp")

frigate/util/config.py

@@ -586,6 +586,23 @@ def migrate_018_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]
 
         new_config["cameras"][name] = camera_config
 
+    # Remove deprecated clean_copy from global snapshots config
+    if new_config.get("snapshots", {}).get("clean_copy") is not None:
+        del new_config["snapshots"]["clean_copy"]
+        if not new_config["snapshots"]:
+            del new_config["snapshots"]
+
+    # Remove deprecated clean_copy from camera snapshots configs
+    for name, camera in new_config.get("cameras", {}).items():
+        camera_config: dict[str, dict[str, Any]] = camera.copy()
+
+        if camera_config.get("snapshots", {}).get("clean_copy") is not None:
+            del camera_config["snapshots"]["clean_copy"]
+            if not camera_config["snapshots"]:
+                del camera_config["snapshots"]
+
+        new_config["cameras"][name] = camera_config
+
     new_config["version"] = "0.18-0"
     return new_config
 

frigate/util/file.py

@@ -5,14 +5,16 @@ import fcntl
 import logging
 import os
 import time
+from datetime import datetime
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 import cv2
 from numpy import ndarray
 
 from frigate.const import CLIPS_DIR, THUMB_DIR
 from frigate.models import Event
+from frigate.util.image import get_snapshot_bytes, relative_box_to_absolute
 
 logger = logging.getLogger(__name__)
 
@@ -30,9 +32,207 @@ def get_event_thumbnail_bytes(event: Event) -> bytes | None:
             return None
 
 
-def get_event_snapshot(event: Event) -> ndarray:
-    media_name = f"{event.camera}-{event.id}"
-    return cv2.imread(f"{os.path.join(CLIPS_DIR, media_name)}.jpg")
+def get_event_snapshot(event: Event) -> ndarray | None:
+    image, _ = load_event_snapshot_image(event)
+    return image
+
+
+def get_event_snapshot_path(
+    event: Event, *, clean_only: bool = False
+) -> tuple[str | None, bool]:
+    clean_snapshot_paths = [
+        os.path.join(CLIPS_DIR, f"{event.camera}-{event.id}-clean.webp"),
+        os.path.join(CLIPS_DIR, f"{event.camera}-{event.id}-clean.png"),
+    ]
+
+    for image_path in clean_snapshot_paths:
+        if os.path.exists(image_path):
+            return image_path, True
+
+    snapshot_path = os.path.join(CLIPS_DIR, f"{event.camera}-{event.id}.jpg")
+    if not os.path.exists(snapshot_path):
+        return None, False
+
+    # Legacy JPG snapshots may already include overlays, so they should never
+    # be treated as clean input for additional rendering.
+    if clean_only:
+        return None, False
+
+    return snapshot_path, False
+
+
+def load_event_snapshot_image(
+    event: Event, *, clean_only: bool = False
+) -> tuple[ndarray | None, bool]:
+    image_path, is_clean_snapshot = get_event_snapshot_path(
+        event, clean_only=clean_only
+    )
+    if image_path is None:
+        return None, False
+
+    image = cv2.imread(image_path)
+    if image is None:
+        logger.warning("Unable to load snapshot from %s", image_path)
+        return None, False
+
+    return image, is_clean_snapshot
+
+
+def _get_event_snapshot_overlay_boxes(
+    frame_shape: tuple[int, ...], event: Event
+) -> list[dict[str, Any]]:
+    overlay_boxes: list[dict[str, Any]] = []
+    draw_data = event.data.get("draw") if event.data else {}
+    draw_boxes = draw_data.get("boxes", []) if isinstance(draw_data, dict) else []
+
+    for draw_box in draw_boxes:
+        box = relative_box_to_absolute(frame_shape, draw_box.get("box"))
+        if box is None:
+            continue
+
+        draw_color = draw_box.get("color", (255, 0, 0))
+        color = (
+            tuple(draw_color) if isinstance(draw_color, (list, tuple)) else (255, 0, 0)
+        )
+        overlay_boxes.append(
+            {
+                "box": box,
+                "label": event.label,
+                "score": draw_box.get("score"),
+                "color": color,
+            }
+        )
+
+    return overlay_boxes
+
+
+def get_event_snapshot_bytes(
+    event: Event,
+    *,
+    ext: str,
+    timestamp: bool = False,
+    bounding_box: bool = False,
+    crop: bool = False,
+    height: int | None = None,
+    quality: int | None = None,
+    timestamp_style: Any | None = None,
+    colormap: dict[str, tuple[int, int, int]] | None = None,
+) -> tuple[bytes | None, float]:
+    best_frame, is_clean_snapshot = load_event_snapshot_image(event)
+    if best_frame is None:
+        return None, 0
+
+    frame_time = _get_event_snapshot_frame_time(event)
+    box = relative_box_to_absolute(
+        best_frame.shape,
+        event.data.get("box") if event.data else None,
+    )
+    overlay_boxes = _get_event_snapshot_overlay_boxes(best_frame.shape, event)
+
+    if (bounding_box or crop or timestamp) and not is_clean_snapshot:
+        logger.warning(
+            "Unable to fully honor snapshot query parameters for completed event %s because the clean snapshot is unavailable.",
+            event.id,
+        )
+
+    return get_snapshot_bytes(
+        best_frame,
+        frame_time,
+        ext=ext,
+        timestamp=timestamp and is_clean_snapshot,
+        bounding_box=bounding_box and is_clean_snapshot,
+        crop=crop and is_clean_snapshot,
+        height=height,
+        quality=quality,
+        label=event.label,
+        box=box,
+        score=_get_event_snapshot_score(event),
+        area=_get_event_snapshot_area(event),
+        attributes=_get_event_snapshot_attributes(
+            best_frame.shape,
+            event.data.get("attributes") if event.data else None,
+        ),
+        color=(colormap or {}).get(event.label, (255, 255, 255)),
+        overlay_boxes=overlay_boxes,
+        timestamp_style=timestamp_style,
+        estimated_speed=_get_event_snapshot_estimated_speed(event),
+    )
+
+
+def _as_timestamp(value: Any) -> float:
+    if isinstance(value, datetime):
+        return value.timestamp()
+
+    return float(value)
+
+
+def _get_event_snapshot_frame_time(event: Event) -> float:
+    if event.data:
+        snapshot_frame_time = event.data.get("snapshot_frame_time")
+        if snapshot_frame_time is not None:
+            return _as_timestamp(snapshot_frame_time)
+
+        frame_time = event.data.get("frame_time")
+        if frame_time is not None:
+            return _as_timestamp(frame_time)
+
+    return _as_timestamp(event.start_time)
+
+
+def _get_event_snapshot_attributes(
+    frame_shape: tuple[int, ...], attributes: list[dict[str, Any]] | None
+) -> list[dict[str, Any]]:
+    absolute_attributes: list[dict[str, Any]] = []
+
+    for attribute in attributes or []:
+        box = relative_box_to_absolute(frame_shape, attribute.get("box"))
+        if box is None:
+            continue
+
+        absolute_attributes.append(
+            {
+                "box": box,
+                "label": attribute.get("label", "attribute"),
+                "score": attribute.get("score", 0),
+            }
+        )
+
+    return absolute_attributes
+
+
+def _get_event_snapshot_score(event: Event) -> float:
+    if event.data:
+        score = event.data.get("score")
+        if score is not None:
+            return score
+
+        top_score = event.data.get("top_score")
+        if top_score is not None:
+            return top_score
+
+    return event.top_score or event.score or 0
+
+
+def _get_event_snapshot_area(event: Event) -> int | None:
+    if event.data:
+        area = event.data.get("snapshot_area")
+        if area is not None:
+            return int(area)
+
+    return None
+
+
+def _get_event_snapshot_estimated_speed(event: Event) -> float:
+    if event.data:
+        estimated_speed = event.data.get("snapshot_estimated_speed")
+        if estimated_speed is not None:
+            return float(estimated_speed)
+
+        average_speed = event.data.get("average_estimated_speed")
+        if average_speed is not None:
+            return float(average_speed)
+
+    return 0
 
 
 ### Deletion

frigate/util/image.py

@@ -270,6 +270,229 @@ def draw_box_with_label(
     )
 
 
+def get_image_quality_params(ext: str, quality: Optional[int]) -> list[int]:
+    if ext in ("jpg", "jpeg"):
+        return [int(cv2.IMWRITE_JPEG_QUALITY), quality if quality is not None else 70]
+
+    if ext == "webp":
+        return [int(cv2.IMWRITE_WEBP_QUALITY), quality if quality is not None else 60]
+
+    return []
+
+
+def relative_box_to_absolute(
+    frame_shape: tuple[int, ...], box: list[float] | tuple[float, ...] | None
+) -> tuple[int, int, int, int] | None:
+    if box is None or len(box) != 4:
+        return None
+
+    frame_height = frame_shape[0]
+    frame_width = frame_shape[1]
+    x_min = int(box[0] * frame_width)
+    y_min = int(box[1] * frame_height)
+    x_max = x_min + int(box[2] * frame_width)
+    y_max = y_min + int(box[3] * frame_height)
+
+    x_min = max(0, min(frame_width - 1, x_min))
+    y_min = max(0, min(frame_height - 1, y_min))
+    x_max = max(x_min + 1, min(frame_width - 1, x_max))
+    y_max = max(y_min + 1, min(frame_height - 1, y_max))
+
+    return (x_min, y_min, x_max, y_max)
+
+
+def _format_snapshot_label(
+    score: float | None,
+    area: int | None,
+    box: tuple[int, int, int, int] | None,
+    estimated_speed: float = 0,
+) -> str:
+    score_value = score or 0
+    score_text = (
+        f"{int(score_value * 100)}%" if score_value <= 1 else f"{int(score_value)}%"
+    )
+
+    if area is None and box is not None:
+        area = int((box[2] - box[0]) * (box[3] - box[1]))
+
+    label = f"{score_text} {int(area or 0)}"
+    if estimated_speed:
+        label = f"{label} {estimated_speed:.1f}"
+
+    return label
+
+
+def draw_snapshot_bounding_boxes(
+    frame: np.ndarray,
+    label: str,
+    box: tuple[int, int, int, int] | None,
+    score: float | None,
+    area: int | None,
+    attributes: list[dict[str, Any]] | None,
+    color: tuple[int, int, int],
+    estimated_speed: float = 0,
+) -> None:
+    if box is None:
+        return
+
+    draw_box_with_label(
+        frame,
+        box[0],
+        box[1],
+        box[2],
+        box[3],
+        label,
+        _format_snapshot_label(score, area, box, estimated_speed),
+        thickness=2,
+        color=color,
+    )
+
+    for attribute in attributes or []:
+        attribute_box = attribute.get("box")
+        if attribute_box is None:
+            continue
+
+        box_area = int(
+            (attribute_box[2] - attribute_box[0])
+            * (attribute_box[3] - attribute_box[1])
+        )
+        draw_box_with_label(
+            frame,
+            attribute_box[0],
+            attribute_box[1],
+            attribute_box[2],
+            attribute_box[3],
+            attribute.get("label", "attribute"),
+            f"{attribute.get('score', 0):.0%} {box_area}",
+            thickness=2,
+            color=color,
+        )
+
+
+def _get_snapshot_overlay_box_label(
+    score: float | int | None, box: tuple[int, int, int, int]
+) -> str:
+    area = int((box[2] - box[0]) * (box[3] - box[1]))
+
+    if score is None:
+        return f"- {area}"
+
+    score_value = float(score)
+    score_text = (
+        f"{int(score_value * 100)}%" if score_value <= 1 else f"{int(score_value)}%"
+    )
+    return f"{score_text} {area}"
+
+
+def draw_snapshot_overlay_boxes(
+    frame: np.ndarray,
+    overlay_boxes: list[dict[str, Any]] | None,
+    default_label: str,
+    default_color: tuple[int, int, int],
+) -> None:
+    for overlay_box in overlay_boxes or []:
+        box = overlay_box.get("box")
+        if box is None:
+            continue
+
+        box_color = overlay_box.get("color", default_color)
+        color = (
+            tuple(box_color) if isinstance(box_color, (list, tuple)) else default_color
+        )
+        draw_box_with_label(
+            frame,
+            box[0],
+            box[1],
+            box[2],
+            box[3],
+            overlay_box.get("label", default_label),
+            _get_snapshot_overlay_box_label(overlay_box.get("score"), box),
+            thickness=2,
+            color=color,
+        )
+
+
+def get_snapshot_bytes(
+    frame: np.ndarray,
+    frame_time: float,
+    ext: str,
+    *,
+    timestamp: bool = False,
+    bounding_box: bool = False,
+    crop: bool = False,
+    height: int | None = None,
+    quality: int | None = None,
+    label: str,
+    box: tuple[int, int, int, int] | None,
+    score: float | None,
+    area: int | None,
+    attributes: list[dict[str, Any]] | None,
+    color: tuple[int, int, int],
+    overlay_boxes: list[dict[str, Any]] | None = None,
+    timestamp_style: Any | None = None,
+    estimated_speed: float = 0,
+) -> tuple[bytes | None, float]:
+    best_frame = frame.copy()
+    crop_box = box
+
+    if crop_box is None and overlay_boxes and len(overlay_boxes) == 1:
+        crop_box = overlay_boxes[0].get("box")
+
+    if bounding_box and box:
+        draw_snapshot_bounding_boxes(
+            best_frame,
+            label,
+            box,
+            score,
+            area,
+            attributes,
+            color,
+            estimated_speed,
+        )
+
+    if bounding_box and overlay_boxes:
+        draw_snapshot_overlay_boxes(best_frame, overlay_boxes, label, color)
+
+    if crop and crop_box:
+        region = calculate_region(
+            best_frame.shape,
+            crop_box[0],
+            crop_box[1],
+            crop_box[2],
+            crop_box[3],
+            300,
+            multiplier=1.1,
+        )
+        best_frame = best_frame[region[1] : region[3], region[0] : region[2]]
+
+    if height:
+        width = int(height * best_frame.shape[1] / best_frame.shape[0])
+        best_frame = cv2.resize(
+            best_frame, dsize=(width, height), interpolation=cv2.INTER_AREA
+        )
+
+    if timestamp and timestamp_style is not None:
+        colors = timestamp_style.color
+        draw_timestamp(
+            best_frame,
+            frame_time,
+            timestamp_style.format,
+            font_effect=timestamp_style.effect,
+            font_thickness=timestamp_style.thickness,
+            font_color=(colors.blue, colors.green, colors.red),
+            position=timestamp_style.position,
+        )
+
+    ret, img = cv2.imencode(
+        f".{ext}", best_frame, get_image_quality_params(ext, quality)
+    )
+
+    if ret:
+        return img.tobytes(), frame_time
+
+    return None, frame_time
+
+
 def grab_cv2_contours(cnts):
     # if the length the contours tuple returned by cv2.findContours
     # is '2' then we are using either OpenCV v2.4, v4-beta, or

frigate/util/media.py

@@ -246,8 +246,8 @@ def sync_recordings(
 def sync_event_snapshots(dry_run: bool = False, force: bool = False) -> SyncResult:
     """Sync event snapshots - delete files not referenced by any event.
 
-    Event snapshots are stored at: CLIPS_DIR/{camera}-{event_id}.jpg
-    Also checks for clean variants: {camera}-{event_id}-clean.webp and -clean.png
+    Event snapshots are stored at: CLIPS_DIR/{camera}-{event_id}-clean.webp
+    Also checks legacy variants: {camera}-{event_id}.jpg and -clean.png
     """
     result = SyncResult(media_type="event_snapshots")