blakeblackshear.frigate/frigate/events.py

304 lines
12 KiB
Python
Raw Normal View History

2020-11-04 13:31:25 +01:00
import datetime
import logging
2020-07-09 13:57:16 +02:00
import os
2020-11-04 13:31:25 +01:00
import queue
2020-07-09 13:57:16 +02:00
import threading
2020-11-04 13:31:25 +01:00
import time
2020-11-24 14:27:51 +01:00
from pathlib import Path
2020-11-04 13:31:25 +01:00
2020-11-24 15:09:16 +01:00
from peewee import fn
from frigate.config import EventsConfig, FrigateConfig, RecordConfig
from frigate.const import CLIPS_DIR
from frigate.models import Event
2020-11-04 04:26:39 +01:00
logger = logging.getLogger(__name__)
2022-04-16 15:42:44 +02:00
def should_insert_db(prev_event, current_event):
"""If current event has new clip or snapshot."""
2022-04-16 15:42:44 +02:00
return (not prev_event["has_clip"] and not prev_event["has_snapshot"]) and (
current_event["has_clip"] or current_event["has_snapshot"]
)
2021-02-17 14:23:32 +01:00
2022-04-16 15:42:44 +02:00
def should_update_db(prev_event, current_event):
"""If current_event has updated fields and (clip or snapshot)."""
2022-04-16 15:42:44 +02:00
return (current_event["has_clip"] or current_event["has_snapshot"]) and (
prev_event["top_score"] != current_event["top_score"]
or prev_event["entered_zones"] != current_event["entered_zones"]
or prev_event["thumbnail"] != current_event["thumbnail"]
or prev_event["has_clip"] != current_event["has_clip"]
or prev_event["has_snapshot"] != current_event["has_snapshot"]
)
2020-07-09 13:57:16 +02:00
class EventProcessor(threading.Thread):
2021-02-17 14:23:32 +01:00
def __init__(
self, config, camera_processes, event_queue, event_processed_queue, stop_event
):
2020-07-09 13:57:16 +02:00
threading.Thread.__init__(self)
2021-02-17 14:23:32 +01:00
self.name = "event_processor"
2020-07-09 13:57:16 +02:00
self.config = config
self.camera_processes = camera_processes
self.cached_clips = {}
self.event_queue = event_queue
2020-11-25 17:37:41 +01:00
self.event_processed_queue = event_processed_queue
2020-07-09 13:57:16 +02:00
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 run(self):
# set an end_time on events without an end_time on startup
Event.update(end_time=Event.start_time + 30).where(
Event.end_time == None
).execute()
while not self.stop_event.is_set():
2020-07-09 13:57:16 +02:00
try:
event_type, camera, event_data = self.event_queue.get(timeout=10)
except queue.Empty:
continue
logger.debug(f"Event received: {event_type} {camera} {event_data['id']}")
2020-07-09 13:57:16 +02:00
event_config: EventsConfig = self.config.cameras[camera].record.events
2021-02-17 14:23:32 +01:00
if event_type == "start":
self.events_in_process[event_data["id"]] = event_data
2020-07-09 13:57:16 +02:00
elif event_type == "update" and should_insert_db(
self.events_in_process[event_data["id"]], event_data
):
self.events_in_process[event_data["id"]] = event_data
# TODO: this will generate a lot of db activity possibly
Event.insert(
id=event_data["id"],
label=event_data["label"],
camera=camera,
start_time=event_data["start_time"] - event_config.pre_capture,
end_time=None,
top_score=event_data["top_score"],
false_positive=event_data["false_positive"],
zones=list(event_data["entered_zones"]),
thumbnail=event_data["thumbnail"],
region=event_data["region"],
box=event_data["box"],
area=event_data["area"],
has_clip=event_data["has_clip"],
has_snapshot=event_data["has_snapshot"],
).execute()
elif event_type == "update" and should_update_db(
self.events_in_process[event_data["id"]], event_data
):
self.events_in_process[event_data["id"]] = event_data
# TODO: this will generate a lot of db activity possibly
Event.update(
label=event_data["label"],
camera=camera,
start_time=event_data["start_time"] - event_config.pre_capture,
end_time=None,
top_score=event_data["top_score"],
false_positive=event_data["false_positive"],
zones=list(event_data["entered_zones"]),
thumbnail=event_data["thumbnail"],
region=event_data["region"],
box=event_data["box"],
area=event_data["area"],
ratio=event_data["ratio"],
has_clip=event_data["has_clip"],
has_snapshot=event_data["has_snapshot"],
).where(Event.id == event_data["id"]).execute()
elif event_type == "end":
if event_data["has_clip"] or event_data["has_snapshot"]:
# Full update for valid end of event
Event.update(
2021-02-17 14:23:32 +01:00
label=event_data["label"],
camera=camera,
start_time=event_data["start_time"] - event_config.pre_capture,
end_time=event_data["end_time"] + event_config.post_capture,
2021-02-17 14:23:32 +01:00
top_score=event_data["top_score"],
false_positive=event_data["false_positive"],
zones=list(event_data["entered_zones"]),
thumbnail=event_data["thumbnail"],
2021-09-23 14:31:48 +02:00
region=event_data["region"],
box=event_data["box"],
area=event_data["area"],
ratio=event_data["ratio"],
has_clip=event_data["has_clip"],
2021-02-17 14:23:32 +01:00
has_snapshot=event_data["has_snapshot"],
).where(Event.id == event_data["id"]).execute()
else:
# Event ended after clip & snapshot disabled,
# only end time should be updated.
Event.update(
end_time=event_data["end_time"] + event_config.post_capture
).where(Event.id == event_data["id"]).execute()
2021-07-09 22:14:16 +02:00
2021-02-17 14:23:32 +01:00
del self.events_in_process[event_data["id"]]
self.event_processed_queue.put((event_data["id"], camera))
2021-02-17 14:23:32 +01:00
# set an end_time on events without an end_time before exiting
Event.update(end_time=datetime.datetime.now().timestamp()).where(
Event.end_time == None
).execute()
logger.info(f"Exiting event processor...")
2020-07-09 13:57:16 +02:00
2020-11-24 14:27:51 +01:00
class EventCleanup(threading.Thread):
def __init__(self, config: FrigateConfig, stop_event):
threading.Thread.__init__(self)
2021-02-17 14:23:32 +01:00
self.name = "event_cleanup"
2020-11-24 14:27:51 +01:00
self.config = config
self.stop_event = stop_event
2021-01-13 13:49:05 +01:00
self.camera_keys = list(self.config.cameras.keys())
2020-11-24 14:27:51 +01:00
def expire(self, media_type):
2021-01-13 13:49:05 +01:00
## Expire events from unlisted cameras based on the global config
2021-06-19 15:44:25 +02:00
if media_type == "clips":
2021-08-15 16:14:13 +02:00
retain_config = self.config.record.events.retain
2021-02-17 14:23:32 +01:00
file_extension = "mp4"
update_params = {"has_clip": False}
2021-01-13 13:49:05 +01:00
else:
retain_config = self.config.snapshots.retain
2021-02-17 14:23:32 +01:00
file_extension = "jpg"
update_params = {"has_snapshot": False}
distinct_labels = (
Event.select(Event.label)
.where(Event.camera.not_in(self.camera_keys))
.distinct()
)
2021-01-13 13:49:05 +01:00
# 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)
2021-02-17 14:23:32 +01:00
expire_after = (
datetime.datetime.now() - datetime.timedelta(days=expire_days)
).timestamp()
2021-01-13 13:49:05 +01:00
# grab all events after specific time
2021-02-17 14:23:32 +01:00
expired_events = Event.select().where(
Event.camera.not_in(self.camera_keys),
Event.start_time < expire_after,
Event.label == l.label,
Event.retain_indefinitely == False,
2021-01-13 13:49:05 +01:00
)
# delete the media from disk
for event in expired_events:
media_name = f"{event.camera}-{event.id}"
2021-06-19 15:44:25 +02:00
media_path = Path(
f"{os.path.join(CLIPS_DIR, media_name)}.{file_extension}"
)
media_path.unlink(missing_ok=True)
2021-06-19 15:44:25 +02:00
if file_extension == "jpg":
media_path = Path(
f"{os.path.join(CLIPS_DIR, media_name)}-clean.png"
)
media_path.unlink(missing_ok=True)
2021-01-13 13:49:05 +01:00
# update the clips attribute for the db entry
2021-02-17 14:23:32 +01:00
update_query = Event.update(update_params).where(
Event.camera.not_in(self.camera_keys),
Event.start_time < expire_after,
Event.label == l.label,
Event.retain_indefinitely == False,
2021-01-13 13:49:05 +01:00
)
update_query.execute()
## Expire events from cameras based on the camera config
for name, camera in self.config.cameras.items():
2021-06-19 15:44:25 +02:00
if media_type == "clips":
2021-08-15 16:14:13 +02:00
retain_config = camera.record.events.retain
2021-01-13 13:49:05 +01:00
else:
retain_config = camera.snapshots.retain
# get distinct objects in database for this camera
2021-02-17 14:23:32 +01:00
distinct_labels = (
Event.select(Event.label).where(Event.camera == name).distinct()
)
2021-01-13 13:49:05 +01:00
2020-11-24 15:09:16 +01:00
# 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)
2021-02-17 14:23:32 +01:00
expire_after = (
datetime.datetime.now() - datetime.timedelta(days=expire_days)
).timestamp()
2020-11-24 15:09:16 +01:00
# grab all events after specific time
2021-02-17 14:23:32 +01:00
expired_events = Event.select().where(
Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label,
Event.retain_indefinitely == False,
2020-11-24 15:09:16 +01:00
)
# delete the grabbed clips from disk
for event in expired_events:
2021-01-13 13:49:05 +01:00
media_name = f"{event.camera}-{event.id}"
2021-06-19 15:44:25 +02:00
media_path = Path(
f"{os.path.join(CLIPS_DIR, media_name)}.{file_extension}"
)
media_path.unlink(missing_ok=True)
2021-06-19 15:44:25 +02:00
if file_extension == "jpg":
media_path = Path(
f"{os.path.join(CLIPS_DIR, media_name)}-clean.png"
)
media_path.unlink(missing_ok=True)
2021-01-13 13:49:05 +01:00
# update the clips attribute for the db entry
2021-02-17 14:23:32 +01:00
update_query = Event.update(update_params).where(
Event.camera == name,
Event.start_time < expire_after,
Event.label == l.label,
Event.retain_indefinitely == False,
2020-11-24 15:09:16 +01:00
)
2021-01-13 13:49:05 +01:00
update_query.execute()
2021-02-07 15:38:17 +01:00
def purge_duplicates(self):
duplicate_query = """with grouped_events as (
select id,
2021-02-17 14:23:32 +01:00
label,
camera,
2021-02-07 15:38:17 +01:00
has_snapshot,
has_clip,
row_number() over (
partition by label, camera, round(start_time/5,0)*5
order by end_time-start_time desc
) as copy_number
from event
)
2021-02-17 14:23:32 +01:00
select distinct id, camera, has_snapshot, has_clip from grouped_events
2021-02-07 15:38:17 +01:00
where copy_number > 1;"""
duplicate_events = Event.raw(duplicate_query)
for event in duplicate_events:
logger.debug(f"Removing duplicate: {event.id}")
media_name = f"{event.camera}-{event.id}"
2021-10-02 13:59:02 +02:00
media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}.jpg")
media_path.unlink(missing_ok=True)
media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}-clean.png")
media_path.unlink(missing_ok=True)
media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}.mp4")
media_path.unlink(missing_ok=True)
2021-02-07 15:38:17 +01:00
2021-02-17 14:23:32 +01:00
(
Event.delete()
.where(Event.id << [event.id for event in duplicate_events])
.execute()
)
2021-01-13 13:49:05 +01:00
def run(self):
# only expire events every 5 minutes
while not self.stop_event.wait(300):
2021-02-17 14:23:32 +01:00
self.expire("clips")
self.expire("snapshots")
2021-02-07 15:38:17 +01:00
self.purge_duplicates()
2021-01-13 13:49:05 +01:00
# drop events from db where has_clip and has_snapshot are false
2021-02-17 14:23:32 +01:00
delete_query = Event.delete().where(
Event.has_clip == False, Event.has_snapshot == False
2021-01-13 13:49:05 +01:00
)
delete_query.execute()
logger.info(f"Exiting event cleanup...")