"""Recording APIs.""" import datetime as dt import logging from datetime import datetime, timedelta from functools import reduce from pathlib import Path from urllib.parse import unquote from fastapi import APIRouter, Depends, Request from fastapi import Path as PathParam from fastapi.responses import JSONResponse from peewee import fn, operator from frigate.api.auth import ( allow_any_authenticated, get_allowed_cameras_for_filter, require_camera_access, require_role, ) from frigate.api.defs.query.recordings_query_parameters import ( MediaRecordingsAvailabilityQueryParams, MediaRecordingsSummaryQueryParams, RecordingsDeleteQueryParams, ) from frigate.api.defs.response.generic_response import GenericResponse from frigate.api.defs.tags import Tags from frigate.const import ( MAX_SEGMENT_DURATION, RECORD_DIR, STREAM_TYPE_MAIN, STREAM_TYPE_SUB, ) from frigate.models import Event, Recordings from frigate.util.recording_coverage import ( coverage_spans, known_video_codecs, realized_timelines, recording_overlap, resolve_coverage, stream_media_summary, ) from frigate.util.time import get_dst_transitions logger = logging.getLogger(__name__) router = APIRouter(tags=[Tags.recordings]) @router.get("/recordings/storage", dependencies=[Depends(require_role(["admin"]))]) def get_recordings_storage_usage(request: Request): recording_stats = request.app.stats_emitter.get_latest_stats()["service"][ "storage" ][RECORD_DIR] if not recording_stats: return JSONResponse({}) total_mb = recording_stats["total"] camera_usages: dict[str, dict] = ( request.app.storage_maintainer.calculate_camera_usages() ) for camera_name in camera_usages.keys(): if camera_usages.get(camera_name, {}).get("usage"): camera_usages[camera_name]["usage_percent"] = ( camera_usages.get(camera_name, {}).get("usage", 0) / total_mb ) * 100 return JSONResponse(content=camera_usages) @router.get("/recordings/summary", dependencies=[Depends(allow_any_authenticated())]) async def all_recordings_summary( request: Request, params: MediaRecordingsSummaryQueryParams = Depends(), allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter), ): """Returns true/false by day indicating if recordings exist""" cameras = params.cameras if cameras != "all": requested = set(unquote(cameras).split(",")) filtered = requested.intersection(allowed_cameras) if not filtered: return JSONResponse(content={}) camera_list = list(filtered) else: camera_list = allowed_cameras min_time: float | None = None max_time: float | None = None for camera in camera_list: cam_min = ( Recordings.select(fn.MIN(Recordings.start_time)) .where(Recordings.camera == camera) .scalar() ) if cam_min is None: continue cam_max = ( Recordings.select(fn.MAX(Recordings.start_time)) .where(Recordings.camera == camera) .scalar() ) min_time = cam_min if min_time is None else min(min_time, cam_min) max_time = cam_max if max_time is None else max(max_time, cam_max) if min_time is None or max_time is None: return JSONResponse(content={}) dst_periods = get_dst_transitions(params.timezone, min_time, max_time) days: dict[str, bool] = {} for period_start, period_end, period_offset in dst_periods: first_start = max(min_time, period_start - MAX_SEGMENT_DURATION) first_day = int((first_start + period_offset) // 86400) last_day = int((min(max_time, period_end) + period_offset) // 86400) day_idx = first_day while day_idx <= last_day: day_str = (dt.date(1970, 1, 1) + dt.timedelta(days=day_idx)).isoformat() day_start = day_idx * 86400 - period_offset day_end = day_start + 86400 if day_str in days: day_idx += 1 continue if day_end <= period_end: upper = Recordings.start_time < day_end else: upper = Recordings.start_time <= period_end has_recordings = ( Recordings.select(Recordings.id) .where( (Recordings.camera << camera_list) & (Recordings.end_time >= period_start) & (Recordings.start_time >= day_start) & upper ) .exists() ) if has_recordings: days[day_str] = True day_idx += 1 continue # empty day next_start: float | None = None for camera in camera_list: cam_next = ( Recordings.select(fn.MIN(Recordings.start_time)) .where( Recordings.camera == camera, Recordings.start_time >= day_end, Recordings.start_time <= period_end, ) .scalar() ) if cam_next is not None and ( next_start is None or cam_next < next_start ): next_start = cam_next if next_start is None: break day_idx = max(day_idx + 1, int((next_start + period_offset) // 86400)) return JSONResponse(content=dict(sorted(days.items()))) @router.get( "/{camera_name}/recordings/summary", dependencies=[Depends(require_camera_access)] ) async def recordings_summary(camera_name: str, timezone: str = "UTC"): """Returns hourly summary for recordings of given camera""" time_range_query = ( Recordings.select( fn.MIN(Recordings.start_time).alias("min_time"), fn.MAX(Recordings.start_time).alias("max_time"), ) .where(Recordings.camera == camera_name) .dicts() .get() ) min_time = time_range_query.get("min_time") max_time = time_range_query.get("max_time") days: dict[str, dict] = {} if min_time is None or max_time is None: return JSONResponse(content=list(days.values())) dst_periods = get_dst_transitions(timezone, min_time, max_time) for period_start, period_end, period_offset in dst_periods: hours_offset = int(period_offset / 60 / 60) minutes_offset = int(period_offset / 60 - hours_offset * 60) period_hour_modifier = f"{hours_offset} hour" period_minute_modifier = f"{minutes_offset} minute" hour_expression = fn.strftime( "%Y-%m-%d %H", fn.datetime( Recordings.start_time, "unixepoch", period_hour_modifier, period_minute_modifier, ), ) # sub rows duplicate the camera's motion/object stats, so # aggregating them too would double-count recording_groups = ( Recordings.select( hour_expression.alias("hour"), fn.SUM(Recordings.duration).alias("duration"), fn.SUM(Recordings.motion).alias("motion"), fn.SUM(Recordings.objects).alias("objects"), ) .where( (Recordings.camera == camera_name) & (Recordings.stream_type == STREAM_TYPE_MAIN) & (Recordings.end_time >= period_start) & (Recordings.start_time <= period_end) ) .group_by((Recordings.start_time + period_offset).cast("int") / 3600) .order_by(Recordings.start_time.desc()) .namedtuples() ) # sub recordings can outlive main, so hours covered only by sub # rows are reported too, flagged as sub_only sub_groups = ( Recordings.select( hour_expression.alias("hour"), fn.SUM(Recordings.duration).alias("duration"), ) .where( (Recordings.camera == camera_name) & (Recordings.stream_type == STREAM_TYPE_SUB) & (Recordings.end_time >= period_start) & (Recordings.start_time <= period_end) ) .group_by((Recordings.start_time + period_offset).cast("int") / 3600) .namedtuples() ) event_groups = ( Event.select( fn.strftime( "%Y-%m-%d %H", fn.datetime( Event.start_time, "unixepoch", period_hour_modifier, period_minute_modifier, ), ).alias("hour"), fn.COUNT(Event.id).alias("count"), ) .where(Event.camera == camera_name, Event.has_clip) .where( (Event.start_time >= period_start) & (Event.start_time <= period_end) ) .group_by((Event.start_time + period_offset).cast("int") / 3600) .namedtuples() ) event_map = {g.hour: g.count for g in event_groups} hour_stats = [ ( g.hour, { "motion": g.motion, "objects": g.objects, "duration": round(g.duration), }, ) for g in recording_groups ] main_hours = {group_hour for group_hour, _ in hour_stats} hour_stats.extend( ( g.hour, { "motion": 0, "objects": 0, "duration": round(g.duration), "sub_only": True, }, ) for g in sub_groups if g.hour not in main_hours ) # restore the most-recent-first ordering after merging in sub hours hour_stats.sort(key=lambda entry: entry[0], reverse=True) for group_hour, stats in hour_stats: parts = group_hour.split() hour = parts[1] day = parts[0] events_count = event_map.get(group_hour, 0) hour_data = { "hour": hour, "events": events_count, **stats, } if day in days: # merge counts if already present (edge-case at DST boundary) days[day]["events"] += events_count or 0 days[day]["hours"].append(hour_data) else: days[day] = { "events": events_count or 0, "hours": [hour_data], "day": day, } return JSONResponse(content=list(days.values())) @router.get( "/{camera_name}/recordings/coverage", dependencies=[Depends(require_camera_access)], ) async def recordings_coverage( camera_name: str, after: float, before: float, timelines: bool = False ): """Returns merged recording coverage spans plus codec compatibility. codecs_compatible is false only when more than one known video codec appears across the range's rows, the case where the merged vod route degrades to a single-stream manifest. """ intervals = resolve_coverage(camera_name, after, before) content = { "spans": coverage_spans(intervals), "codecs_compatible": len(known_video_codecs(intervals)) <= 1, "streams": stream_media_summary(intervals), } # pure computation (shared plan_clip, record-time keyframe index), but # opt-in for payload hygiene: day-level requests need only the spans if timelines: content["timelines"] = realized_timelines(intervals) return JSONResponse(content=content) @router.get("/{camera_name}/recordings", dependencies=[Depends(require_camera_access)]) async def recordings( camera_name: str, after: float | None = None, before: float | None = None, ): """Return specific camera recordings between the given 'after'/'end' times. If not provided the last hour will be used""" now = datetime.now() after = after if after is not None else (now - timedelta(hours=1)).timestamp() before = before if before is not None else now.timestamp() recordings = ( Recordings.select( Recordings.id, Recordings.start_time, Recordings.end_time, Recordings.segment_size, Recordings.motion, Recordings.objects, Recordings.motion_heatmap, Recordings.duration, ) .where( Recordings.camera == camera_name, Recordings.stream_type == STREAM_TYPE_MAIN, Recordings.start_time >= after - MAX_SEGMENT_DURATION, Recordings.end_time >= after, Recordings.start_time <= before, ) .order_by(Recordings.start_time) .dicts() .iterator() ) return JSONResponse(content=list(recordings)) @router.get( "/recordings/unavailable", response_model=list[dict], dependencies=[Depends(allow_any_authenticated())], ) async def no_recordings( request: Request, params: MediaRecordingsAvailabilityQueryParams = Depends(), allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter), ): """Get time ranges with no recordings.""" cameras = params.cameras if cameras != "all": requested = set(unquote(cameras).split(",")) camera_list = list(requested.intersection(allowed_cameras)) else: camera_list = list(allowed_cameras) if not camera_list: return JSONResponse(content=[]) before = params.before or datetime.now().timestamp() after = params.after or (datetime.now() - timedelta(hours=1)).timestamp() scale = params.scale recordings: list[tuple[float, float]] = [] for camera in camera_list: recordings.extend( Recordings.select(Recordings.start_time, Recordings.end_time) .where( Recordings.camera == camera, Recordings.start_time >= after - MAX_SEGMENT_DURATION, Recordings.end_time >= after, Recordings.start_time <= before, ) .tuples() .iterator() ) # the merge pass below expects a single start-ordered timeline recordings.sort() # Merge overlapping/adjacent recordings into covered intervals. The query # orders by start_time, so a single pass merges them covered: list[tuple[float, float]] = [] for rec_start, rec_end in recordings: if covered and rec_start <= covered[-1][1]: covered[-1] = (covered[-1][0], max(covered[-1][1], rec_end)) else: covered.append((rec_start, rec_end)) # Iterate through time segments and check if each has any recording no_recording_segments = [] current = after current_gap_start = None idx = 0 covered_count = len(covered) while current < before: segment_end = min(current + scale, before) # Advance past covered intervals that end before this segment begins; # they cannot overlap this or any later segment. while idx < covered_count and covered[idx][1] <= current: idx += 1 # A covered interval overlaps the segment when it starts before the # segment ends (its end is already known to be > current). has_recording = idx < covered_count and covered[idx][0] < segment_end if not has_recording: # This segment has no recordings if current_gap_start is None: current_gap_start = current # Start a new gap else: # This segment has recordings if current_gap_start is not None: # End the current gap and append it no_recording_segments.append( {"start_time": int(current_gap_start), "end_time": int(current)} ) current_gap_start = None current = segment_end # Append the last gap if it exists if current_gap_start is not None: no_recording_segments.append( {"start_time": int(current_gap_start), "end_time": int(before)} ) return JSONResponse(content=no_recording_segments) @router.delete( "/recordings/start/{start}/end/{end}", response_model=GenericResponse, dependencies=[Depends(require_role(["admin"]))], summary="Delete recordings", description="""Deletes recordings within the specified time range. Recordings can be filtered by cameras and kept based on motion, objects, or audio attributes. """, ) async def delete_recordings( start: float = PathParam(..., description="Start timestamp (unix)"), end: float = PathParam(..., description="End timestamp (unix)"), params: RecordingsDeleteQueryParams = Depends(), allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter), ): """Delete recordings in the specified time range.""" if start >= end: return JSONResponse( content={ "success": False, "message": "Start time must be less than end time.", }, status_code=400, ) cameras = params.cameras if cameras != "all": requested = set(cameras.split(",")) filtered = requested.intersection(allowed_cameras) if not filtered: return JSONResponse( content={ "success": False, "message": "No valid cameras found in the request.", }, status_code=400, ) camera_list = list(filtered) else: camera_list = allowed_cameras # Parse keep parameter keep_set = set() if params.keep: keep_set = set(params.keep.split(",")) # Build query to find overlapping recordings clauses = [ recording_overlap(start, end), (Recordings.camera << camera_list), ] keep_clauses = [] if "motion" in keep_set: keep_clauses.append(Recordings.motion.is_null(False) & (Recordings.motion > 0)) if "object" in keep_set: keep_clauses.append( Recordings.objects.is_null(False) & (Recordings.objects > 0) ) if "audio" in keep_set: keep_clauses.append(Recordings.dBFS.is_null(False)) if keep_clauses: keep_condition = reduce(operator.or_, keep_clauses) clauses.append(~keep_condition) recordings_to_delete = ( Recordings.select(Recordings.id, Recordings.path) .where(reduce(operator.and_, clauses)) .dicts() .iterator() ) recording_ids = [] deleted_count = 0 error_count = 0 for recording in recordings_to_delete: recording_ids.append(recording["id"]) try: Path(recording["path"]).unlink(missing_ok=True) deleted_count += 1 except Exception as e: logger.error(f"Failed to delete recording file {recording['path']}: {e}") error_count += 1 if recording_ids: max_deletes = 100000 recording_ids_list = list(recording_ids) for i in range(0, len(recording_ids_list), max_deletes): Recordings.delete().where( Recordings.id << recording_ids_list[i : i + max_deletes] ).execute() message = f"Successfully deleted {deleted_count} recording(s)." if error_count > 0: message += f" {error_count} file deletion error(s) occurred." return JSONResponse( content={"success": True, "message": message}, status_code=200, )