Change clean snapshots from png to webp format (#20484)

* save clean webp instead of png

* send clean webp to plus with fallback for old events

* manual event webp

* event cleanup

* api def

* convert png to webp if exists

* update reference config

* change quality

docs/docs/configuration/reference.md

@@ -574,7 +574,7 @@ record:
 snapshots:
   # Optional: Enable writing jpg snapshot to /media/frigate/clips (default: shown below)
   enabled: False
-  # Optional: save a clean PNG copy of the snapshot image (default: shown below)
+  # Optional: save a clean copy of the snapshot image (default: shown below)
   clean_copy: True
   # Optional: print a timestamp on the snapshots (default: shown below)
   timestamp: False

docs/static/frigate-api.yaml

@@ -4077,7 +4077,7 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
-  /events/{event_id}/snapshot-clean.png:
+  /events/{event_id}/snapshot-clean.webp:
     get:
       tags:
         - Media

frigate/api/event.py

@@ -939,12 +939,12 @@ async def send_to_plus(request: Request, event_id: str, body: SubmitPlusBody = N
         include_annotation = None
 
     if event.end_time is None:
-        logger.error(f"Unable to load clean png for in-progress event: {event.id}")
+        logger.error(f"Unable to load clean snapshot for in-progress event: {event.id}")
         return JSONResponse(
             content=(
                 {
                     "success": False,
-                    "message": "Unable to load clean png for in-progress event",
+                    "message": "Unable to load clean snapshot for in-progress event",
                 }
             ),
             status_code=400,
@@ -957,24 +957,44 @@ async def send_to_plus(request: Request, event_id: str, body: SubmitPlusBody = N
             content=({"success": False, "message": message}), status_code=400
         )
 
-    # load clean.png
+    # load clean.webp or clean.png (legacy)
     try:
-        filename = f"{event.camera}-{event.id}-clean.png"
-        image = cv2.imread(os.path.join(CLIPS_DIR, filename))
+        filename_webp = f"{event.camera}-{event.id}-clean.webp"
+        filename_png = f"{event.camera}-{event.id}-clean.png"
+
+        image_path = None
+        if os.path.exists(os.path.join(CLIPS_DIR, filename_webp)):
+            image_path = os.path.join(CLIPS_DIR, filename_webp)
+        elif os.path.exists(os.path.join(CLIPS_DIR, filename_png)):
+            image_path = os.path.join(CLIPS_DIR, filename_png)
+
+        if image_path is None:
+            logger.error(f"Unable to find clean snapshot for event: {event.id}")
+            return JSONResponse(
+                content=(
+                    {
+                        "success": False,
+                        "message": "Unable to find clean snapshot for event",
+                    }
+                ),
+                status_code=400,
+            )
+
+        image = cv2.imread(image_path)
     except Exception:
-        logger.error(f"Unable to load clean png for event: {event.id}")
+        logger.error(f"Unable to load clean snapshot for event: {event.id}")
         return JSONResponse(
             content=(
-                {"success": False, "message": "Unable to load clean png for event"}
+                {"success": False, "message": "Unable to load clean snapshot for event"}
             ),
             status_code=400,
         )
 
     if image is None or image.size == 0:
-        logger.error(f"Unable to load clean png for event: {event.id}")
+        logger.error(f"Unable to load clean snapshot for event: {event.id}")
         return JSONResponse(
             content=(
-                {"success": False, "message": "Unable to load clean png for event"}
+                {"success": False, "message": "Unable to load clean snapshot for event"}
             ),
             status_code=400,
         )
@@ -1422,6 +1442,7 @@ async def delete_single_event(event_id: str, request: Request) -> dict:
         snapshot_paths = [
             Path(f"{os.path.join(CLIPS_DIR, media_name)}.jpg"),
             Path(f"{os.path.join(CLIPS_DIR, media_name)}-clean.png"),
+            Path(f"{os.path.join(CLIPS_DIR, media_name)}-clean.webp"),
         ]
         for media in snapshot_paths:
             media.unlink(missing_ok=True)

frigate/api/media.py

@@ -1165,9 +1165,9 @@ def grid_snapshot(
         )
 
 
-@router.get("/events/{event_id}/snapshot-clean.png")
+@router.get("/events/{event_id}/snapshot-clean.webp")
 def event_snapshot_clean(request: Request, event_id: str, download: bool = False):
-    png_bytes = None
+    webp_bytes = None
     try:
         event = Event.get(Event.id == event_id)
         snapshot_config = request.app.frigate_config.cameras[event.camera].snapshots
@@ -1189,7 +1189,7 @@ def event_snapshot_clean(request: Request, event_id: str, download: bool = False
                     if event_id in camera_state.tracked_objects:
                         tracked_obj = camera_state.tracked_objects.get(event_id)
                         if tracked_obj is not None:
-                            png_bytes = tracked_obj.get_clean_png()
+                            webp_bytes = tracked_obj.get_clean_webp()
                             break
             except Exception:
                 return JSONResponse(
@@ -1205,12 +1205,56 @@ def event_snapshot_clean(request: Request, event_id: str, download: bool = False
         return JSONResponse(
             content={"success": False, "message": "Event not found"}, status_code=404
         )
-    if png_bytes is None:
+    if webp_bytes is None:
         try:
-            clean_snapshot_path = os.path.join(
+            # webp
+            clean_snapshot_path_webp = os.path.join(
+                CLIPS_DIR, f"{event.camera}-{event.id}-clean.webp"
+            )
+            # png (legacy)
+            clean_snapshot_path_png = os.path.join(
                 CLIPS_DIR, f"{event.camera}-{event.id}-clean.png"
             )
-            if not os.path.exists(clean_snapshot_path):
+
+            if os.path.exists(clean_snapshot_path_webp):
+                with open(clean_snapshot_path_webp, "rb") as image_file:
+                    webp_bytes = image_file.read()
+            elif os.path.exists(clean_snapshot_path_png):
+                # convert png to webp and save for future use
+                png_image = cv2.imread(clean_snapshot_path_png, cv2.IMREAD_UNCHANGED)
+                if png_image is None:
+                    return JSONResponse(
+                        content={
+                            "success": False,
+                            "message": "Invalid png snapshot",
+                        },
+                        status_code=400,
+                    )
+
+                ret, webp_data = cv2.imencode(
+                    ".webp", png_image, [int(cv2.IMWRITE_WEBP_QUALITY), 60]
+                )
+                if not ret:
+                    return JSONResponse(
+                        content={
+                            "success": False,
+                            "message": "Unable to convert png to webp",
+                        },
+                        status_code=400,
+                    )
+
+                webp_bytes = webp_data.tobytes()
+
+                # save the converted webp for future requests
+                try:
+                    with open(clean_snapshot_path_webp, "wb") as f:
+                        f.write(webp_bytes)
+                except Exception as e:
+                    logger.warning(
+                        f"Failed to save converted webp for event {event.id}: {e}"
+                    )
+                    # continue since we now have the data to return
+            else:
                 return JSONResponse(
                     content={
                         "success": False,
@@ -1218,33 +1262,29 @@ def event_snapshot_clean(request: Request, event_id: str, download: bool = False
                     },
                     status_code=404,
                 )
-            with open(
-                os.path.join(CLIPS_DIR, f"{event.camera}-{event.id}-clean.png"), "rb"
-            ) as image_file:
-                png_bytes = image_file.read()
         except Exception:
-            logger.error(f"Unable to load clean png for event: {event.id}")
+            logger.error(f"Unable to load clean snapshot for event: {event.id}")
             return JSONResponse(
                 content={
                     "success": False,
-                    "message": "Unable to load clean png for event",
+                    "message": "Unable to load clean snapshot for event",
                 },
                 status_code=400,
             )
 
     headers = {
-        "Content-Type": "image/png",
+        "Content-Type": "image/webp",
         "Cache-Control": "private, max-age=31536000",
     }
 
     if download:
         headers["Content-Disposition"] = (
-            f"attachment; filename=snapshot-{event_id}-clean.png"
+            f"attachment; filename=snapshot-{event_id}-clean.webp"
         )
 
     return Response(
-        png_bytes,
-        media_type="image/png",
+        webp_bytes,
+        media_type="image/webp",
         headers=headers,
     )
 

frigate/camera/state.py

@@ -530,17 +530,19 @@ class CameraState:
 
         # write clean snapshot if enabled
         if self.camera_config.snapshots.clean_copy:
-            ret, png = cv2.imencode(".png", img_frame)
+            ret, webp = cv2.imencode(
+                ".webp", img_frame, [int(cv2.IMWRITE_WEBP_QUALITY), 80]
+            )
 
             if ret:
                 with open(
                     os.path.join(
                         CLIPS_DIR,
-                        f"{self.camera_config.name}-{event_id}-clean.png",
+                        f"{self.camera_config.name}-{event_id}-clean.webp",
                     ),
                     "wb",
                 ) as p:
-                    p.write(png.tobytes())
+                    p.write(webp.tobytes())
 
         # write jpg snapshot with optional annotations
         if draw.get("boxes") and isinstance(draw.get("boxes"), list):

frigate/events/cleanup.py

@@ -229,6 +229,11 @@ class EventCleanup(threading.Thread):
             try:
                 media_path.unlink(missing_ok=True)
                 if file_extension == "jpg":
+                    media_path = Path(
+                        f"{os.path.join(CLIPS_DIR, media_name)}-clean.webp"
+                    )
+                    media_path.unlink(missing_ok=True)
+                    # Also delete clean.png (legacy) for backward compatibility
                     media_path = Path(
                         f"{os.path.join(CLIPS_DIR, media_name)}-clean.png"
                     )

frigate/track/tracked_object.py

@@ -432,7 +432,7 @@ class TrackedObject:
             _, img = cv2.imencode(f".{ext}", np.zeros((175, 175, 3), np.uint8))
             return img.tobytes()
 
-    def get_clean_png(self) -> bytes | None:
+    def get_clean_webp(self) -> bytes | None:
         if self.thumbnail_data is None:
             return None
 
@@ -443,13 +443,15 @@ class TrackedObject:
             )
         except KeyError:
             logger.warning(
-                f"Unable to create clean png because frame {self.thumbnail_data['frame_time']} is not in the cache"
+                f"Unable to create clean webp because frame {self.thumbnail_data['frame_time']} is not in the cache"
             )
             return None
 
-        ret, png = cv2.imencode(".png", best_frame)
+        ret, webp = cv2.imencode(
+            ".webp", best_frame, [int(cv2.IMWRITE_WEBP_QUALITY), 60]
+        )
         if ret:
-            return png.tobytes()
+            return webp.tobytes()
         else:
             return None
 
@@ -583,8 +585,8 @@ class TrackedObject:
 
         # write clean snapshot if enabled
         if snapshot_config.clean_copy:
-            png_bytes = self.get_clean_png()
-            if png_bytes is None:
+            webp_bytes = self.get_clean_webp()
+            if webp_bytes is None:
                 logger.warning(
                     f"Unable to save clean snapshot for {self.obj_data['id']}."
                 )
@@ -592,11 +594,11 @@ class TrackedObject:
                 with open(
                     os.path.join(
                         CLIPS_DIR,
-                        f"{self.camera_config.name}-{self.obj_data['id']}-clean.png",
+                        f"{self.camera_config.name}-{self.obj_data['id']}-clean.webp",
                     ),
                     "wb",
                 ) as p:
-                    p.write(png_bytes)
+                    p.write(webp_bytes)
 
     def write_thumbnail_to_disk(self) -> None:
         if not self.camera_config.name:

frigate/util/path.py

@@ -42,6 +42,9 @@ def delete_event_snapshot(event: Event) -> bool:
 
     try:
         media_path.unlink(missing_ok=True)
+        media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}-clean.webp")
+        media_path.unlink(missing_ok=True)
+        # also delete clean.png (legacy) for backward compatibility
         media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}-clean.png")
         media_path.unlink(missing_ok=True)
         return True