diff --git a/frigate/api/media.py b/frigate/api/media.py index 24f5d4f41..d5c0db3e7 100644 --- a/frigate/api/media.py +++ b/frigate/api/media.py @@ -45,6 +45,7 @@ from frigate.models import Event, Previews, Recordings, Regions, ReviewSegment from frigate.output.preview import get_most_recent_preview_frame from frigate.track.object_processing import TrackedObjectProcessor from frigate.util.file import ( + get_event_snapshot_path, get_event_snapshot_bytes, get_event_thumbnail_bytes, load_event_snapshot_image, @@ -1055,8 +1056,10 @@ def clear_region_grid(request: Request, camera_name: str): ) def event_snapshot_clean(request: Request, event_id: str, download: bool = False): webp_bytes = None + event_complete = False try: event = Event.get(Event.id == event_id) + event_complete = event.end_time is not None snapshot_config = request.app.frigate_config.cameras[event.camera].snapshots if not (snapshot_config.enabled and event.has_snapshot): return JSONResponse( @@ -1094,8 +1097,10 @@ def event_snapshot_clean(request: Request, event_id: str, download: bool = False ) if webp_bytes is None: try: - image, is_clean_snapshot = load_event_snapshot_image(event, clean_only=True) - if not is_clean_snapshot or image is None: + image_path, is_clean_snapshot = get_event_snapshot_path( + event, clean_only=True + ) + if not is_clean_snapshot or image_path is None: return JSONResponse( content={ "success": False, @@ -1104,19 +1109,33 @@ def event_snapshot_clean(request: Request, event_id: str, download: bool = False status_code=404, ) - ret, webp_data = cv2.imencode( - ".webp", image, get_image_quality_params("webp", None) - ) - if not ret: - return JSONResponse( - content={ - "success": False, - "message": "Unable to convert snapshot to webp", - }, - status_code=400, - ) + if image_path.endswith(".webp"): + with open(image_path, "rb") as image_file: + webp_bytes = image_file.read() + else: + image = load_event_snapshot_image(event, clean_only=True)[0] + if image is None: + return JSONResponse( + content={ + "success": False, + "message": "Unable to load clean snapshot for event", + }, + status_code=400, + ) - webp_bytes = webp_data.tobytes() + ret, webp_data = cv2.imencode( + ".webp", image, get_image_quality_params("webp", None) + ) + if not ret: + return JSONResponse( + content={ + "success": False, + "message": "Unable to convert snapshot to webp", + }, + status_code=400, + ) + + webp_bytes = webp_data.tobytes() except Exception: logger.error(f"Unable to load clean snapshot for event: {event.id}") return JSONResponse( @@ -1129,7 +1148,7 @@ def event_snapshot_clean(request: Request, event_id: str, download: bool = False headers = { "Content-Type": "image/webp", - "Cache-Control": "private, max-age=31536000", + "Cache-Control": "private, max-age=31536000" if event_complete else "no-cache", } if download: diff --git a/frigate/util/file.py b/frigate/util/file.py index cf26ea066..a6cdefee1 100644 --- a/frigate/util/file.py +++ b/frigate/util/file.py @@ -55,38 +55,43 @@ def _event_snapshot_is_clean(event: Event) -> bool: return bool(event.data and event.data.get("snapshot_clean")) -def load_event_snapshot_image( +def get_event_snapshot_path( event: Event, *, clean_only: bool = False -) -> tuple[ndarray | None, bool]: +) -> 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 not os.path.exists(image_path): - continue - - image = _load_snapshot_image(image_path) - if image is None: - logger.warning("Unable to load clean snapshot from %s", image_path) - continue - - return image, True + 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 - image = _load_snapshot_image(snapshot_path) - if image is None: - logger.warning("Unable to load snapshot from %s", snapshot_path) - return None, False - is_clean_snapshot = _event_snapshot_is_clean(event) if clean_only and not is_clean_snapshot: return None, False + return snapshot_path, is_clean_snapshot + + +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 = _load_snapshot_image(image_path) + if image is None: + logger.warning("Unable to load snapshot from %s", image_path) + return None, False + return image, is_clean_snapshot