blakeblackshear.frigate/frigate/events.py

284 lines
11 KiB
Python
Raw Normal View History

2020-11-04 13:31:25 +01:00
import datetime
import json
import logging
2020-07-09 13:57:16 +02:00
import os
2020-11-04 13:31:25 +01:00
import queue
import subprocess as sp
2020-07-09 13:57:16 +02:00
import threading
2020-11-04 13:31:25 +01:00
import time
2020-07-09 13:57:16 +02:00
from collections import defaultdict
2020-11-24 14:27:51 +01:00
from pathlib import Path
2020-11-04 13:31:25 +01:00
import psutil
2020-07-09 13:57:16 +02:00
2020-11-24 14:27:51 +01:00
from frigate.config import FrigateConfig
2020-11-01 21:58:51 +01:00
from frigate.models import Event
2020-11-24 15:09:16 +01:00
from peewee import fn
2020-11-04 04:26:39 +01:00
logger = logging.getLogger(__name__)
2020-07-09 13:57:16 +02:00
class EventProcessor(threading.Thread):
2020-11-01 21:58:51 +01:00
def __init__(self, config, camera_processes, event_queue, stop_event):
2020-07-09 13:57:16 +02:00
threading.Thread.__init__(self)
2020-11-04 13:28:07 +01:00
self.name = 'event_processor'
2020-07-09 13:57:16 +02:00
self.config = config
2020-11-03 15:15:58 +01:00
self.cache_dir = self.config.save_clips.cache_dir
self.clips_dir = self.config.save_clips.clips_dir
2020-07-09 13:57:16 +02:00
self.camera_processes = camera_processes
self.cached_clips = {}
self.event_queue = event_queue
self.events_in_process = {}
2020-08-02 15:46:36 +02:00
self.stop_event = stop_event
2020-07-09 13:57:16 +02:00
def refresh_cache(self):
cached_files = os.listdir(self.cache_dir)
files_in_use = []
for process_data in self.camera_processes.values():
try:
ffmpeg_process = psutil.Process(pid=process_data['ffmpeg_pid'].value)
2020-07-09 13:57:16 +02:00
flist = ffmpeg_process.open_files()
if flist:
for nt in flist:
if nt.path.startswith(self.cache_dir):
files_in_use.append(nt.path.split('/')[-1])
except:
continue
for f in cached_files:
if f in files_in_use or f in self.cached_clips:
continue
2020-07-26 14:50:42 +02:00
camera = '-'.join(f.split('-')[:-1])
start_time = datetime.datetime.strptime(f.split('-')[-1].split('.')[0], '%Y%m%d%H%M%S')
2020-07-09 13:57:16 +02:00
ffprobe_cmd = " ".join([
'ffprobe',
'-v',
'error',
'-show_entries',
'format=duration',
'-of',
'default=noprint_wrappers=1:nokey=1',
f"{os.path.join(self.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())
else:
2020-11-04 04:26:39 +01:00
logger.info(f"bad file: {f}")
2020-07-09 13:57:16 +02:00
os.remove(os.path.join(self.cache_dir,f))
continue
self.cached_clips[f] = {
'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']
else:
earliest_event = datetime.datetime.now().timestamp()
# if the earliest event exceeds the max seconds, cap it
2020-11-03 15:15:58 +01:00
max_seconds = self.config.save_clips.max_seconds
if datetime.datetime.now().timestamp()-earliest_event > max_seconds:
earliest_event = datetime.datetime.now().timestamp()-max_seconds
2020-07-09 13:57:16 +02:00
for f, data in list(self.cached_clips.items()):
if earliest_event-90 > data['start_time']+data['duration']:
del self.cached_clips[f]
os.remove(os.path.join(self.cache_dir,f))
2020-07-26 01:59:04 +02:00
def create_clip(self, camera, event_data, pre_capture):
2020-07-09 13:57:16 +02:00
# 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'])
while sorted_clips[-1]['start_time'] + sorted_clips[-1]['duration'] < event_data['end_time']:
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'])
2020-07-26 01:59:04 +02:00
playlist_start = event_data['start_time']-pre_capture
2020-07-09 13:57:16 +02:00
playlist_end = event_data['end_time']+5
playlist_lines = []
for clip in sorted_clips:
# clip ends before playlist start time, skip
if clip['start_time']+clip['duration'] < playlist_start:
continue
# clip starts after playlist ends, finish
if clip['start_time'] > playlist_end:
break
playlist_lines.append(f"file '{os.path.join(self.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 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'])}")
clip_name = f"{camera}-{event_data['id']}"
ffmpeg_cmd = [
'ffmpeg',
'-y',
'-protocol_whitelist',
'pipe,file',
'-f',
'concat',
'-safe',
'0',
'-i',
'-',
'-c',
'copy',
2020-11-01 21:58:51 +01:00
f"{os.path.join(self.clips_dir, clip_name)}.mp4"
2020-07-09 13:57:16 +02:00
]
p = sp.run(ffmpeg_cmd, input="\n".join(playlist_lines), encoding='ascii', capture_output=True)
if p.returncode != 0:
2020-11-04 04:26:39 +01:00
logger.error(p.stderr)
2020-07-09 13:57:16 +02:00
return
def run(self):
while True:
2020-08-02 15:46:36 +02:00
if self.stop_event.is_set():
2020-11-04 04:26:39 +01:00
logger.info(f"Exiting event processor...")
2020-08-02 15:46:36 +02:00
break
2020-08-08 14:39:57 +02:00
2020-07-09 13:57:16 +02:00
try:
event_type, camera, event_data = self.event_queue.get(timeout=10)
except queue.Empty:
2020-08-08 14:39:57 +02:00
if not self.stop_event.is_set():
self.refresh_cache()
2020-07-09 13:57:16 +02:00
continue
self.refresh_cache()
2020-11-03 15:15:58 +01:00
save_clips_config = self.config.cameras[camera].save_clips
# if save clips is not enabled for this camera, just continue
2020-11-03 15:15:58 +01:00
if not save_clips_config.enabled:
continue
# if specific objects are listed for this camera, only save clips for them
2020-11-03 15:15:58 +01:00
# TODO: default to all tracked objects rather than checking for None
if save_clips_config.objects:
if not event_data['label'] in save_clips_config.objects:
continue
2020-07-09 13:57:16 +02:00
if event_type == 'start':
self.events_in_process[event_data['id']] = event_data
if event_type == 'end':
if len(self.cached_clips) > 0 and not event_data['false_positive']:
2020-11-03 15:15:58 +01:00
self.create_clip(camera, event_data, save_clips_config.pre_capture)
Event.create(
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']
)
2020-07-09 13:57:16 +02:00
del self.events_in_process[event_data['id']]
2020-11-24 14:27:51 +01:00
class EventCleanup(threading.Thread):
def __init__(self, config: FrigateConfig, stop_event):
threading.Thread.__init__(self)
self.name = 'event_cleanup'
self.config = config
self.clips_dir = self.config.save_clips.clips_dir
self.stop_event = stop_event
def run(self):
counter = 0
while(True):
if self.stop_event.is_set():
logger.info(f"Exiting event cleanup...")
break
2020-11-24 15:09:16 +01:00
# only expire events every 10 minutes, but check for stop events every 10 seconds
2020-11-24 16:16:54 +01:00
time.sleep(10)
2020-11-24 15:09:16 +01:00
counter = counter + 1
if counter < 60:
continue
counter = 0
camera_keys = list(self.config.cameras.keys())
# Expire events from unlisted cameras based on the global config
retain_config = self.config.save_clips.retain
distinct_labels = (Event.select(Event.label)
.where(Event.camera.not_in(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()
# grab all events after specific time
expired_events = (
Event.select()
.where(Event.camera.not_in(camera_keys),
Event.start_time < expire_after,
Event.label == l.label)
)
# delete the grabbed clips from disk
for event in expired_events:
clip_name = f"{event.camera}-{event.id}"
clip = Path(f"{os.path.join(self.clips_dir, clip_name)}.mp4")
clip.unlink(missing_ok=True)
# delete the event for this type from the db
delete_query = (
Event.delete()
.where(Event.camera.not_in(camera_keys),
Event.start_time < expire_after,
Event.label == l.label)
)
delete_query.execute()
# Expire events from cameras based on the camera config
2020-11-24 14:27:51 +01:00
for name, camera in self.config.cameras.items():
retain_config = camera.save_clips.retain
# get distinct objects in database for this camera
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)
2020-11-24 15:09:16 +01:00
expire_after = (datetime.datetime.now() - datetime.timedelta(days=expire_days)).timestamp()
2020-11-24 14:27:51 +01:00
# grab all events after specific time
expired_events = (
Event.select()
2020-11-24 15:09:16 +01:00
.where(Event.camera == name,
2020-11-24 14:27:51 +01:00
Event.start_time < expire_after,
Event.label == l.label)
)
# delete the grabbed clips from disk
for event in expired_events:
clip_name = f"{event.camera}-{event.id}"
clip = Path(f"{os.path.join(self.clips_dir, clip_name)}.mp4")
clip.unlink(missing_ok=True)
# delete the event for this type from the db
delete_query = (
Event.delete()
.where( Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label)
)
delete_query.execute()