administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-02-05 18:55:23 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Make camera recordings mover asynchronous

Browse Source
This commit is contained in:
Nick Mowen 2023-06-14 16:21:17 -06:00
parent ff90db30e6
commit c217c34e3d
1 changed files with 105 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

211
frigate/record/maintainer.py
View File

@ -1,5 +1,6 @@
"""Maintain recording segments in cache.""" """Maintain recording segments in cache."""
import asyncio
import datetime import datetime
import logging import logging
import multiprocessing as mp import multiprocessing as mp
@ -13,6 +14,7 @@ from collections import defaultdict
from multiprocessing.synchronize import Event as MpEvent from multiprocessing.synchronize import Event as MpEvent
from pathlib import Path from pathlib import Path
from typing import Any, Tuple from typing import Any, Tuple
import traceback
import psutil import psutil
@ -42,7 +44,7 @@ class RecordingMaintainer(threading.Thread):
self.recordings_info: dict[str, Any] = defaultdict(list) self.recordings_info: dict[str, Any] = defaultdict(list)
self.end_time_cache: dict[str, Tuple[datetime.datetime, float]] = {} self.end_time_cache: dict[str, Tuple[datetime.datetime, float]] = {}
def move_files(self) -> None: async def move_files(self) -> None:
cache_files = sorted( cache_files = sorted(
[ [
d d
@ -121,115 +123,111 @@ class RecordingMaintainer(threading.Thread):
) )
.order_by(Event.start_time) .order_by(Event.start_time)
) )
for r in recordings:
cache_path = r["cache_path"]
start_time = r["start_time"]
# Just delete files if recordings are turned off await asyncio.gather(
if ( *(self.validate_segment(camera, events, r) for r in recordings)
camera not in self.config.cameras )
or not self.process_info[camera]["record_enabled"].value
): async def validate_segment(
self, camera: str, events: Event, recording: dict[str, any]
) -> None:
cache_path = recording["cache_path"]
start_time = recording["start_time"]
# Just delete files if recordings are turned off
if (
camera not in self.config.cameras
or not self.process_info[camera]["record_enabled"].value
):
Path(cache_path).unlink(missing_ok=True)
self.end_time_cache.pop(cache_path, None)
return
if cache_path in self.end_time_cache:
end_time, duration = self.end_time_cache[cache_path]
else:
ffprobe_cmd = [
"ffprobe",
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
f"{cache_path}",
]
p = sp.run(ffprobe_cmd, capture_output=True)
if p.returncode == 0 and p.stdout.decode():
duration = float(p.stdout.decode().strip())
else:
duration = -1
# ensure duration is within expected length
if 0 < duration < MAX_SEGMENT_DURATION:
end_time = start_time + datetime.timedelta(seconds=duration)
self.end_time_cache[cache_path] = (end_time, duration)
else:
if duration == -1:
logger.warning(
f"Failed to probe corrupt segment {cache_path} : {p.returncode} - {str(p.stderr)}"
)
logger.warning(f"Discarding a corrupt recording segment: {cache_path}")
Path(cache_path).unlink(missing_ok=True)
return
# if cached file's start_time is earlier than the retain days for the camera
if start_time <= (
(
datetime.datetime.now()
- datetime.timedelta(
days=self.config.cameras[camera].record.retain.days
)
)
):
# if the cached segment overlaps with the events:
overlaps = False
for event in events:
# if the event starts in the future, stop checking events
# and remove this segment
if event.start_time > end_time.timestamp():
overlaps = False
Path(cache_path).unlink(missing_ok=True) Path(cache_path).unlink(missing_ok=True)
self.end_time_cache.pop(cache_path, None) self.end_time_cache.pop(cache_path, None)
continue break
if cache_path in self.end_time_cache: # if the event is in progress or ends after the recording starts, keep it
end_time, duration = self.end_time_cache[cache_path] # and stop looking at events
else: if event.end_time is None or event.end_time >= start_time.timestamp():
ffprobe_cmd = [ overlaps = True
"ffprobe", break
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
f"{cache_path}",
]
p = sp.run(ffprobe_cmd, capture_output=True)
if p.returncode == 0 and p.stdout.decode():
duration = float(p.stdout.decode().strip())
else:
duration = -1
# ensure duration is within expected length if overlaps:
if 0 < duration < MAX_SEGMENT_DURATION: record_mode = self.config.cameras[camera].record.events.retain.mode
end_time = start_time + datetime.timedelta(seconds=duration) # move from cache to recordings immediately
self.end_time_cache[cache_path] = (end_time, duration) self.store_segment(
else: camera,
if duration == -1: start_time,
logger.warning( end_time,
f"Failed to probe corrupt segment {cache_path} : {p.returncode} - {str(p.stderr)}" duration,
) cache_path,
record_mode,
logger.warning( )
f"Discarding a corrupt recording segment: {cache_path}" # if it doesn't overlap with an event, go ahead and drop the segment
) # if it ends more than the configured pre_capture for the camera
Path(cache_path).unlink(missing_ok=True) else:
continue pre_capture = self.config.cameras[camera].record.events.pre_capture
most_recently_processed_frame_time = self.recordings_info[camera][-1][0]
# if cached file's start_time is earlier than the retain days for the camera retain_cutoff = most_recently_processed_frame_time - pre_capture
if start_time <= ( if end_time.timestamp() < retain_cutoff:
( Path(cache_path).unlink(missing_ok=True)
datetime.datetime.now() self.end_time_cache.pop(cache_path, None)
- datetime.timedelta( # else retain days includes this segment
days=self.config.cameras[camera].record.retain.days else:
) record_mode = self.config.cameras[camera].record.retain.mode
) self.store_segment(
): camera, start_time, end_time, duration, cache_path, record_mode
# if the cached segment overlaps with the events: )
overlaps = False
for event in events:
# if the event starts in the future, stop checking events
# and remove this segment
if event.start_time > end_time.timestamp():
overlaps = False
Path(cache_path).unlink(missing_ok=True)
self.end_time_cache.pop(cache_path, None)
break
# if the event is in progress or ends after the recording starts, keep it
# and stop looking at events
if (
event.end_time is None
or event.end_time >= start_time.timestamp()
):
overlaps = True
break
if overlaps:
record_mode = self.config.cameras[
camera
].record.events.retain.mode
# move from cache to recordings immediately
self.store_segment(
camera,
start_time,
end_time,
duration,
cache_path,
record_mode,
)
# if it doesn't overlap with an event, go ahead and drop the segment
# if it ends more than the configured pre_capture for the camera
else:
pre_capture = self.config.cameras[
camera
].record.events.pre_capture
most_recently_processed_frame_time = self.recordings_info[
camera
][-1][0]
retain_cutoff = most_recently_processed_frame_time - pre_capture
if end_time.timestamp() < retain_cutoff:
Path(cache_path).unlink(missing_ok=True)
self.end_time_cache.pop(cache_path, None)
# else retain days includes this segment
else:
record_mode = self.config.cameras[camera].record.retain.mode
self.store_segment(
camera, start_time, end_time, duration, cache_path, record_mode
)
def segment_stats( def segment_stats(
self, camera: str, start_time: datetime.datetime, end_time: datetime.datetime self, camera: str, start_time: datetime.datetime, end_time: datetime.datetime
@ -386,12 +384,13 @@ class RecordingMaintainer(threading.Thread):
break break
try: try:
self.move_files() asyncio.run(self.move_files())
except Exception as e: except Exception as e:
logger.error( logger.error(
"Error occurred when attempting to maintain recording cache" "Error occurred when attempting to maintain recording cache"
) )
logger.error(e) logger.error(e)
logger.error(str(traceback.print_exc()))
duration = datetime.datetime.now().timestamp() - run_start duration = datetime.datetime.now().timestamp() - run_start
wait_time = max(0, 5 - duration) wait_time = max(0, 5 - duration)