chris/frigate
1
0
Fork 0
forked from Github/frigate
Code Pull Requests Packages Activity

formatting cleanup

Browse Source
This commit is contained in:
Blake Blackshear
2021-02-17 07:23:32 -06:00
parent b8f72a5bcb
commit 39ff49e054
23 changed files with 2621 additions and 1736 deletions
Download Patch File Download Diff File Expand all files Collapse all files

335
frigate/events.py
View File

@@ -20,10 +20,13 @@ from peewee import fn
logger = logging.getLogger(__name__)
class EventProcessor(threading.Thread):
def __init__(self, config, camera_processes, event_queue, event_processed_queue, stop_event):
def __init__(
self, config, camera_processes, event_queue, event_processed_queue, stop_event
):
threading.Thread.__init__(self)
self.name = 'event_processor'
self.name = "event_processor"
self.config = config
self.camera_processes = camera_processes
self.cached_clips = {}
@@ -33,31 +36,35 @@ class EventProcessor(threading.Thread):
self.stop_event = stop_event
def should_create_clip(self, camera, event_data):
if event_data['false_positive']:
if event_data["false_positive"]:
return False
# if there are required zones and there is no overlap
required_zones = self.config.cameras[camera].clips.required_zones
if len(required_zones) > 0 and not set(event_data['entered_zones']) & set(required_zones):
logger.debug(f"Not creating clip for {event_data['id']} because it did not enter required zones")
if len(required_zones) > 0 and not set(event_data["entered_zones"]) & set(
required_zones
):
logger.debug(
f"Not creating clip for {event_data['id']} because it did not enter required zones"
)
return False
return True
def refresh_cache(self):
cached_files = os.listdir(CACHE_DIR)
files_in_use = []
for process in psutil.process_iter():
try:
if process.name() != 'ffmpeg':
if process.name() != "ffmpeg":
continue
flist = process.open_files()
if flist:
for nt in flist:
if nt.path.startswith(CACHE_DIR):
files_in_use.append(nt.path.split('/')[-1])
files_in_use.append(nt.path.split("/")[-1])
except:
continue
@@ -65,119 +72,154 @@ class EventProcessor(threading.Thread):
if f in files_in_use or f in self.cached_clips:
continue
camera = '-'.join(f.split('-')[:-1])
start_time = datetime.datetime.strptime(f.split('-')[-1].split('.')[0], '%Y%m%d%H%M%S')
ffprobe_cmd = " ".join([
'ffprobe',
'-v',
'error',
'-show_entries',
'format=duration',
'-of',
'default=noprint_wrappers=1:nokey=1',
f"{os.path.join(CACHE_DIR,f)}"
])
camera = "-".join(f.split("-")[:-1])
start_time = datetime.datetime.strptime(
f.split("-")[-1].split(".")[0], "%Y%m%d%H%M%S"
)
ffprobe_cmd = " ".join(
[
"ffprobe",
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
f"{os.path.join(CACHE_DIR,f)}",
]
)
p = sp.Popen(ffprobe_cmd, stdout=sp.PIPE, shell=True)
(output, err) = p.communicate()
p_status = p.wait()
if p_status == 0:
duration = float(output.decode('utf-8').strip())
duration = float(output.decode("utf-8").strip())
else:
logger.info(f"bad file: {f}")
os.remove(os.path.join(CACHE_DIR,f))
os.remove(os.path.join(CACHE_DIR, f))
continue
self.cached_clips[f] = {
'path': f,
'camera': camera,
'start_time': start_time.timestamp(),
'duration': duration
"path": f,
"camera": camera,
"start_time": start_time.timestamp(),
"duration": duration,
}
if len(self.events_in_process) > 0:
earliest_event = min(self.events_in_process.values(), key=lambda x:x['start_time'])['start_time']
earliest_event = min(
self.events_in_process.values(), key=lambda x: x["start_time"]
)["start_time"]
else:
earliest_event = datetime.datetime.now().timestamp()
# if the earliest event exceeds the max seconds, cap it
max_seconds = self.config.clips.max_seconds
if datetime.datetime.now().timestamp()-earliest_event > max_seconds:
earliest_event = datetime.datetime.now().timestamp()-max_seconds
if datetime.datetime.now().timestamp() - earliest_event > max_seconds:
earliest_event = datetime.datetime.now().timestamp() - max_seconds
for f, data in list(self.cached_clips.items()):
if earliest_event-90 > data['start_time']+data['duration']:
if earliest_event - 90 > data["start_time"] + data["duration"]:
del self.cached_clips[f]
logger.debug(f"Cleaning up cached file {f}")
os.remove(os.path.join(CACHE_DIR,f))
os.remove(os.path.join(CACHE_DIR, f))
# if we are still using more than 90% of the cache, proactively cleanup
cache_usage = shutil.disk_usage("/tmp/cache")
if cache_usage.used/cache_usage.total > .9 and cache_usage.free < 200000000 and len(self.cached_clips) > 0:
if (
cache_usage.used / cache_usage.total > 0.9
and cache_usage.free < 200000000
and len(self.cached_clips) > 0
):
logger.warning("More than 90% of the cache is used.")
logger.warning("Consider increasing space available at /tmp/cache or reducing max_seconds in your clips config.")
logger.warning(
"Consider increasing space available at /tmp/cache or reducing max_seconds in your clips config."
)
logger.warning("Proactively cleaning up the cache...")
while cache_usage.used/cache_usage.total > .9:
oldest_clip = min(self.cached_clips.values(), key=lambda x:x['start_time'])
del self.cached_clips[oldest_clip['path']]
os.remove(os.path.join(CACHE_DIR,oldest_clip['path']))
while cache_usage.used / cache_usage.total > 0.9:
oldest_clip = min(
self.cached_clips.values(), key=lambda x: x["start_time"]
)
del self.cached_clips[oldest_clip["path"]]
os.remove(os.path.join(CACHE_DIR, oldest_clip["path"]))
cache_usage = shutil.disk_usage("/tmp/cache")
def create_clip(self, camera, event_data, pre_capture, post_capture):
# get all clips from the camera with the event sorted
sorted_clips = sorted([c for c in self.cached_clips.values() if c['camera'] == camera], key = lambda i: i['start_time'])
sorted_clips = sorted(
[c for c in self.cached_clips.values() if c["camera"] == camera],
key=lambda i: i["start_time"],
)
# if there are no clips in the cache or we are still waiting on a needed file check every 5 seconds
wait_count = 0
while len(sorted_clips) == 0 or sorted_clips[-1]['start_time'] + sorted_clips[-1]['duration'] < event_data['end_time']+post_capture:
while (
len(sorted_clips) == 0
or sorted_clips[-1]["start_time"] + sorted_clips[-1]["duration"]
< event_data["end_time"] + post_capture
):
if wait_count > 4:
logger.warning(f"Unable to create clip for {camera} and event {event_data['id']}. There were no cache files for this event.")
logger.warning(
f"Unable to create clip for {camera} and event {event_data['id']}. There were no cache files for this event."
)
return False
logger.debug(f"No cache clips for {camera}. Waiting...")
time.sleep(5)
self.refresh_cache()
# get all clips from the camera with the event sorted
sorted_clips = sorted([c for c in self.cached_clips.values() if c['camera'] == camera], key = lambda i: i['start_time'])
sorted_clips = sorted(
[c for c in self.cached_clips.values() if c["camera"] == camera],
key=lambda i: i["start_time"],
)
wait_count += 1
playlist_start = event_data['start_time']-pre_capture
playlist_end = event_data['end_time']+post_capture
playlist_start = event_data["start_time"] - pre_capture
playlist_end = event_data["end_time"] + post_capture
playlist_lines = []
for clip in sorted_clips:
# clip ends before playlist start time, skip
if clip['start_time']+clip['duration'] < playlist_start:
if clip["start_time"] + clip["duration"] < playlist_start:
continue
# clip starts after playlist ends, finish
if clip['start_time'] > playlist_end:
if clip["start_time"] > playlist_end:
break
playlist_lines.append(f"file '{os.path.join(CACHE_DIR,clip['path'])}'")
# if this is the starting clip, add an inpoint
if clip['start_time'] < playlist_start:
playlist_lines.append(f"inpoint {int(playlist_start-clip['start_time'])}")
if clip["start_time"] < playlist_start:
playlist_lines.append(
f"inpoint {int(playlist_start-clip['start_time'])}"
)
# if this is the ending clip, add an outpoint
if clip['start_time']+clip['duration'] > playlist_end:
playlist_lines.append(f"outpoint {int(playlist_end-clip['start_time'])}")
if clip["start_time"] + clip["duration"] > playlist_end:
playlist_lines.append(
f"outpoint {int(playlist_end-clip['start_time'])}"
)
clip_name = f"{camera}-{event_data['id']}"
ffmpeg_cmd = [
'ffmpeg',
'-y',
'-protocol_whitelist',
'pipe,file',
'-f',
'concat',
'-safe',
'0',
'-i',
'-',
'-c',
'copy',
'-movflags',
'+faststart',
f"{os.path.join(CLIPS_DIR, clip_name)}.mp4"
"ffmpeg",
"-y",
"-protocol_whitelist",
"pipe,file",
"-f",
"concat",
"-safe",
"0",
"-i",
"-",
"-c",
"copy",
"-movflags",
"+faststart",
f"{os.path.join(CLIPS_DIR, clip_name)}.mp4",
]
p = sp.run(ffmpeg_cmd, input="\n".join(playlist_lines), encoding='ascii', capture_output=True)
p = sp.run(
ffmpeg_cmd,
input="\n".join(playlist_lines),
encoding="ascii",
capture_output=True,
)
if p.returncode != 0:
logger.error(p.stderr)
return False
@@ -199,68 +241,80 @@ class EventProcessor(threading.Thread):
logger.debug(f"Event received: {event_type} {camera} {event_data['id']}")
self.refresh_cache()
if event_type == 'start':
self.events_in_process[event_data['id']] = event_data
if event_type == "start":
self.events_in_process[event_data["id"]] = event_data
if event_type == 'end':
if event_type == "end":
clips_config = self.config.cameras[camera].clips
clip_created = False
if self.should_create_clip(camera, event_data):
if clips_config.enabled and (clips_config.objects is None or event_data['label'] in clips_config.objects):
clip_created = self.create_clip(camera, event_data, clips_config.pre_capture, clips_config.post_capture)
if clip_created or event_data['has_snapshot']:
if clips_config.enabled and (
clips_config.objects is None
or event_data["label"] in clips_config.objects
):
clip_created = self.create_clip(
camera,
event_data,
clips_config.pre_capture,
clips_config.post_capture,
)
if clip_created or event_data["has_snapshot"]:
Event.create(
id=event_data['id'],
label=event_data['label'],
id=event_data["id"],
label=event_data["label"],
camera=camera,
start_time=event_data['start_time'],
end_time=event_data['end_time'],
top_score=event_data['top_score'],
false_positive=event_data['false_positive'],
zones=list(event_data['entered_zones']),
thumbnail=event_data['thumbnail'],
start_time=event_data["start_time"],
end_time=event_data["end_time"],
top_score=event_data["top_score"],
false_positive=event_data["false_positive"],
zones=list(event_data["entered_zones"]),
thumbnail=event_data["thumbnail"],
has_clip=clip_created,
has_snapshot=event_data['has_snapshot'],
has_snapshot=event_data["has_snapshot"],
)
del self.events_in_process[event_data['id']]
self.event_processed_queue.put((event_data['id'], camera))
del self.events_in_process[event_data["id"]]
self.event_processed_queue.put((event_data["id"], camera))
class EventCleanup(threading.Thread):
def __init__(self, config: FrigateConfig, stop_event):
threading.Thread.__init__(self)
self.name = 'event_cleanup'
self.name = "event_cleanup"
self.config = config
self.stop_event = stop_event
self.camera_keys = list(self.config.cameras.keys())
def expire(self, media):
## Expire events from unlisted cameras based on the global config
if media == 'clips':
if media == "clips":
retain_config = self.config.clips.retain
file_extension = 'mp4'
update_params = {'has_clip': False}
file_extension = "mp4"
update_params = {"has_clip": False}
else:
retain_config = self.config.snapshots.retain
file_extension = 'jpg'
update_params = {'has_snapshot': False}
distinct_labels = (Event.select(Event.label)
.where(Event.camera.not_in(self.camera_keys))
.distinct())
file_extension = "jpg"
update_params = {"has_snapshot": False}
distinct_labels = (
Event.select(Event.label)
.where(Event.camera.not_in(self.camera_keys))
.distinct()
)
# loop over object types in db
for l in distinct_labels:
# get expiration time for this label
expire_days = retain_config.objects.get(l.label, retain_config.default)
expire_after = (datetime.datetime.now() - datetime.timedelta(days=expire_days)).timestamp()
expire_after = (
datetime.datetime.now() - datetime.timedelta(days=expire_days)
).timestamp()
# grab all events after specific time
expired_events = (
Event.select()
.where(Event.camera.not_in(self.camera_keys),
Event.start_time < expire_after,
Event.label == l.label)
expired_events = Event.select().where(
Event.camera.not_in(self.camera_keys),
Event.start_time < expire_after,
Event.label == l.label,
)
# delete the media from disk
for event in expired_events:
@@ -268,56 +322,57 @@ class EventCleanup(threading.Thread):
media = Path(f"{os.path.join(CLIPS_DIR, media_name)}.{file_extension}")
media.unlink(missing_ok=True)
# update the clips attribute for the db entry
update_query = (
Event.update(update_params)
.where(Event.camera.not_in(self.camera_keys),
Event.start_time < expire_after,
Event.label == l.label)
update_query = Event.update(update_params).where(
Event.camera.not_in(self.camera_keys),
Event.start_time < expire_after,
Event.label == l.label,
)
update_query.execute()
## Expire events from cameras based on the camera config
for name, camera in self.config.cameras.items():
if media == 'clips':
if media == "clips":
retain_config = camera.clips.retain
else:
retain_config = camera.snapshots.retain
# get distinct objects in database for this camera
distinct_labels = (Event.select(Event.label)
.where(Event.camera == name)
.distinct())
distinct_labels = (
Event.select(Event.label).where(Event.camera == name).distinct()
)
# loop over object types in db
for l in distinct_labels:
# get expiration time for this label
expire_days = retain_config.objects.get(l.label, retain_config.default)
expire_after = (datetime.datetime.now() - datetime.timedelta(days=expire_days)).timestamp()
expire_after = (
datetime.datetime.now() - datetime.timedelta(days=expire_days)
).timestamp()
# grab all events after specific time
expired_events = (
Event.select()
.where(Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label)
expired_events = Event.select().where(
Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label,
)
# delete the grabbed clips from disk
for event in expired_events:
media_name = f"{event.camera}-{event.id}"
media = Path(f"{os.path.join(CLIPS_DIR, media_name)}.{file_extension}")
media = Path(
f"{os.path.join(CLIPS_DIR, media_name)}.{file_extension}"
)
media.unlink(missing_ok=True)
# update the clips attribute for the db entry
update_query = (
Event.update(update_params)
.where( Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label)
update_query = Event.update(update_params).where(
Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label,
)
update_query.execute()
def purge_duplicates(self):
duplicate_query = """with grouped_events as (
select id,
label,
camera,
label,
camera,
has_snapshot,
has_clip,
row_number() over (
@@ -327,7 +382,7 @@ class EventCleanup(threading.Thread):
from event
)
select distinct id, camera, has_snapshot, has_clip from grouped_events
select distinct id, camera, has_snapshot, has_clip from grouped_events
where copy_number > 1;"""
duplicate_events = Event.raw(duplicate_query)
@@ -341,13 +396,15 @@ class EventCleanup(threading.Thread):
media = Path(f"{os.path.join(CLIPS_DIR, media_name)}.mp4")
media.unlink(missing_ok=True)
(Event.delete()
.where( Event.id << [event.id for event in duplicate_events] )
.execute())
(
Event.delete()
.where(Event.id << [event.id for event in duplicate_events])
.execute()
)
def run(self):
counter = 0
while(True):
while True:
if self.stop_event.is_set():
logger.info(f"Exiting event cleanup...")
break
@@ -359,14 +416,12 @@ class EventCleanup(threading.Thread):
continue
counter = 0
self.expire('clips')
self.expire('snapshots')
self.expire("clips")
self.expire("snapshots")
self.purge_duplicates()
# drop events from db where has_clip and has_snapshot are false
delete_query = (
Event.delete()
.where( Event.has_clip == False,
Event.has_snapshot == False)
delete_query = Event.delete().where(
Event.has_clip == False, Event.has_snapshot == False
)
delete_query.execute()