blakeblackshear.frigate/frigate/app.py

376 lines
13 KiB
Python
Raw Normal View History

2020-11-04 13:28:07 +01:00
import json
import logging
import multiprocessing as mp
import os
from logging.handlers import QueueHandler
from typing import Dict, List
2020-12-03 15:01:03 +01:00
import sys
2020-12-05 18:14:18 +01:00
import signal
2020-11-04 13:28:07 +01:00
import yaml
2021-02-13 15:09:44 +01:00
from gevent import pywsgi
from geventwebsocket.handler import WebSocketHandler
2020-12-24 14:47:27 +01:00
from peewee_migrate import Router
2020-11-04 13:28:07 +01:00
from playhouse.sqlite_ext import SqliteExtDatabase
2021-01-24 13:53:01 +01:00
from playhouse.sqliteq import SqliteQueueDatabase
2020-11-04 13:28:07 +01:00
from frigate.config import FrigateConfig
2020-12-01 14:22:23 +01:00
from frigate.const import RECORD_DIR, CLIPS_DIR, CACHE_DIR
2020-11-04 13:28:07 +01:00
from frigate.edgetpu import EdgeTPUProcess
2020-11-24 14:27:51 +01:00
from frigate.events import EventProcessor, EventCleanup
2020-11-04 13:28:07 +01:00
from frigate.http import create_app
from frigate.log import log_process, root_configurer
from frigate.models import Event, Recordings
2020-11-04 13:28:07 +01:00
from frigate.mqtt import create_mqtt_client
from frigate.object_processing import TrackedObjectProcessor
from frigate.output import output_frames
2020-11-30 04:31:02 +01:00
from frigate.record import RecordingMaintainer
from frigate.stats import StatsEmitter, stats_init
2020-11-04 13:28:07 +01:00
from frigate.video import capture_camera, track_camera
from frigate.watchdog import FrigateWatchdog
2020-12-03 15:00:23 +01:00
from frigate.zeroconf import broadcast_zeroconf
2020-11-04 13:28:07 +01:00
logger = logging.getLogger(__name__)
2021-02-17 14:23:32 +01:00
class FrigateApp:
2020-11-04 13:28:07 +01:00
def __init__(self):
self.stop_event = mp.Event()
self.config: FrigateConfig = None
self.detection_queue = mp.Queue()
self.detectors: Dict[str, EdgeTPUProcess] = {}
self.detection_out_events: Dict[str, mp.Event] = {}
self.detection_shms: List[mp.shared_memory.SharedMemory] = []
self.log_queue = mp.Queue()
self.camera_metrics = {}
2020-12-01 14:22:23 +01:00
2021-01-16 04:33:53 +01:00
def set_environment_vars(self):
for key, value in self.config.environment_vars.items():
os.environ[key] = value
2020-12-01 14:22:23 +01:00
def ensure_dirs(self):
for d in [RECORD_DIR, CLIPS_DIR, CACHE_DIR]:
if not os.path.exists(d) and not os.path.islink(d):
2020-12-03 15:01:22 +01:00
logger.info(f"Creating directory: {d}")
2020-12-01 14:22:23 +01:00
os.makedirs(d)
2020-12-03 15:01:22 +01:00
else:
logger.debug(f"Skipping directory: {d}")
2020-12-21 14:37:42 +01:00
2020-11-04 13:28:07 +01:00
def init_logger(self):
2021-02-17 14:23:32 +01:00
self.log_process = mp.Process(
target=log_process, args=(self.log_queue,), name="log_process"
)
2020-12-05 18:14:18 +01:00
self.log_process.daemon = True
2020-11-04 13:28:07 +01:00
self.log_process.start()
root_configurer(self.log_queue)
2021-02-17 14:23:32 +01:00
2020-11-04 13:28:07 +01:00
def init_config(self):
2021-02-17 14:23:32 +01:00
config_file = os.environ.get("CONFIG_FILE", "/config/config.yml")
2020-11-04 13:28:07 +01:00
self.config = FrigateConfig(config_file=config_file)
for camera_name in self.config.cameras.keys():
# create camera_metrics
self.camera_metrics[camera_name] = {
2021-02-17 14:23:32 +01:00
"camera_fps": mp.Value("d", 0.0),
"skipped_fps": mp.Value("d", 0.0),
"process_fps": mp.Value("d", 0.0),
"detection_enabled": mp.Value(
"i", self.config.cameras[camera_name].detect.enabled
),
"detection_fps": mp.Value("d", 0.0),
"detection_frame": mp.Value("d", 0.0),
"read_start": mp.Value("d", 0.0),
"ffmpeg_pid": mp.Value("i", 0),
"frame_queue": mp.Queue(maxsize=2),
2020-11-04 13:28:07 +01:00
}
2021-02-17 14:23:32 +01:00
def check_config(self):
for name, camera in self.config.cameras.items():
2021-02-17 14:23:32 +01:00
assigned_roles = list(
set([r for i in camera.ffmpeg.inputs for r in i.roles])
)
if not camera.clips.enabled and "clips" in assigned_roles:
logger.warning(
f"Camera {name} has clips assigned to an input, but clips is not enabled."
)
elif camera.clips.enabled and not "clips" in assigned_roles:
logger.warning(
f"Camera {name} has clips enabled, but clips is not assigned to an input."
)
if not camera.record.enabled and "record" in assigned_roles:
logger.warning(
f"Camera {name} has record assigned to an input, but record is not enabled."
)
elif camera.record.enabled and not "record" in assigned_roles:
logger.warning(
f"Camera {name} has record enabled, but record is not assigned to an input."
)
if not camera.rtmp.enabled and "rtmp" in assigned_roles:
logger.warning(
f"Camera {name} has rtmp assigned to an input, but rtmp is not enabled."
)
elif camera.rtmp.enabled and not "rtmp" in assigned_roles:
logger.warning(
f"Camera {name} has rtmp enabled, but rtmp is not assigned to an input."
)
2020-12-04 13:59:03 +01:00
def set_log_levels(self):
logging.getLogger().setLevel(self.config.logger.default)
for log, level in self.config.logger.logs.items():
logging.getLogger(log).setLevel(level)
2021-02-17 14:23:32 +01:00
if not "geventwebsocket.handler" in self.config.logger.logs:
logging.getLogger("geventwebsocket.handler").setLevel("ERROR")
2020-11-04 13:28:07 +01:00
def init_queues(self):
2020-11-25 17:37:41 +01:00
# Queues for clip processing
2020-11-04 13:28:07 +01:00
self.event_queue = mp.Queue()
2020-11-25 17:37:41 +01:00
self.event_processed_queue = mp.Queue()
self.video_output_queue = mp.Queue()
2020-11-04 13:28:07 +01:00
# Queue for cameras to push tracked objects to
2021-02-17 14:23:32 +01:00
self.detected_frames_queue = mp.Queue(
maxsize=len(self.config.cameras.keys()) * 2
)
2020-11-04 13:28:07 +01:00
def init_database(self):
# Migrate DB location
old_db_path = os.path.join(CLIPS_DIR, "frigate.db")
if not os.path.isfile(self.config.database.path) and os.path.isfile(
old_db_path
):
os.rename(old_db_path, self.config.database.path)
# Migrate DB schema
2021-01-24 13:53:01 +01:00
migrate_db = SqliteExtDatabase(self.config.database.path)
2020-12-24 14:47:27 +01:00
# Run migrations
2021-02-17 14:23:32 +01:00
del logging.getLogger("peewee_migrate").handlers[:]
2021-01-24 13:53:01 +01:00
router = Router(migrate_db)
2020-12-24 14:47:27 +01:00
router.run()
2021-01-24 13:53:01 +01:00
migrate_db.close()
self.db = SqliteQueueDatabase(self.config.database.path)
models = [Event, Recordings]
2020-11-04 13:28:07 +01:00
self.db.bind(models)
def init_stats(self):
self.stats_tracking = stats_init(self.camera_metrics, self.detectors)
2020-11-04 13:28:07 +01:00
def init_web_server(self):
2021-02-17 14:23:32 +01:00
self.flask_app = create_app(
self.config,
self.db,
self.stats_tracking,
self.detected_frames_processor,
self.mqtt_client,
)
2020-11-04 13:28:07 +01:00
def init_mqtt(self):
2021-01-16 03:52:59 +01:00
self.mqtt_client = create_mqtt_client(self.config, self.camera_metrics)
2020-11-04 13:28:07 +01:00
def start_detectors(self):
model_shape = (self.config.model.height, self.config.model.width)
2020-11-04 13:28:07 +01:00
for name in self.config.cameras.keys():
self.detection_out_events[name] = mp.Event()
2021-02-17 14:23:32 +01:00
shm_in = mp.shared_memory.SharedMemory(
name=name,
create=True,
size=self.config.model.height * self.config.model.width * 3,
)
shm_out = mp.shared_memory.SharedMemory(
name=f"out-{name}", create=True, size=20 * 6 * 4
)
2020-11-04 13:28:07 +01:00
self.detection_shms.append(shm_in)
self.detection_shms.append(shm_out)
for name, detector in self.config.detectors.items():
2021-02-17 14:23:32 +01:00
if detector.type == "cpu":
self.detectors[name] = EdgeTPUProcess(
name,
self.detection_queue,
self.detection_out_events,
model_shape,
"cpu",
detector.num_threads,
)
if detector.type == "edgetpu":
self.detectors[name] = EdgeTPUProcess(
name,
self.detection_queue,
self.detection_out_events,
model_shape,
detector.device,
detector.num_threads,
)
2020-11-04 13:28:07 +01:00
def start_detected_frames_processor(self):
2021-02-17 14:23:32 +01:00
self.detected_frames_processor = TrackedObjectProcessor(
self.config,
self.mqtt_client,
self.config.mqtt.topic_prefix,
self.detected_frames_queue,
self.event_queue,
self.event_processed_queue,
self.video_output_queue,
2021-02-17 14:23:32 +01:00
self.stop_event,
)
2020-11-04 13:28:07 +01:00
self.detected_frames_processor.start()
def start_video_output_processor(self):
output_processor = mp.Process(
target=output_frames,
name=f"output_processor",
args=(
self.config,
self.video_output_queue,
),
)
output_processor.daemon = True
self.output_processor = output_processor
output_processor.start()
2021-05-30 13:45:37 +02:00
logger.info(f"Output process started: {output_processor.pid}")
2020-11-04 13:28:07 +01:00
def start_camera_processors(self):
model_shape = (self.config.model.height, self.config.model.width)
2020-11-04 13:28:07 +01:00
for name, config in self.config.cameras.items():
2021-02-17 14:23:32 +01:00
camera_process = mp.Process(
target=track_camera,
name=f"camera_processor:{name}",
args=(
name,
config,
model_shape,
self.detection_queue,
self.detection_out_events[name],
self.detected_frames_queue,
self.camera_metrics[name],
),
)
2020-11-04 13:28:07 +01:00
camera_process.daemon = True
2021-02-17 14:23:32 +01:00
self.camera_metrics[name]["process"] = camera_process
2020-11-04 13:28:07 +01:00
camera_process.start()
logger.info(f"Camera processor started for {name}: {camera_process.pid}")
def start_camera_capture_processes(self):
for name, config in self.config.cameras.items():
2021-02-17 14:23:32 +01:00
capture_process = mp.Process(
target=capture_camera,
name=f"camera_capture:{name}",
args=(name, config, self.camera_metrics[name]),
)
2020-11-04 13:28:07 +01:00
capture_process.daemon = True
2021-02-17 14:23:32 +01:00
self.camera_metrics[name]["capture_process"] = capture_process
2020-11-04 13:28:07 +01:00
capture_process.start()
logger.info(f"Capture process started for {name}: {capture_process.pid}")
2021-02-17 14:23:32 +01:00
2021-05-23 04:02:26 +02:00
def start_birdseye_outputter(self):
self.birdseye_outputter = BirdsEyeFrameOutputter(self.stop_event)
self.birdseye_outputter.start()
2020-11-04 13:28:07 +01:00
def start_event_processor(self):
2021-02-17 14:23:32 +01:00
self.event_processor = EventProcessor(
self.config,
self.camera_metrics,
self.event_queue,
self.event_processed_queue,
self.stop_event,
)
2020-11-04 13:28:07 +01:00
self.event_processor.start()
2021-02-17 14:23:32 +01:00
2020-11-24 14:27:51 +01:00
def start_event_cleanup(self):
self.event_cleanup = EventCleanup(self.config, self.stop_event)
self.event_cleanup.start()
2021-02-17 14:23:32 +01:00
2020-11-30 04:31:02 +01:00
def start_recording_maintainer(self):
self.recording_maintainer = RecordingMaintainer(self.config, self.stop_event)
self.recording_maintainer.start()
def start_stats_emitter(self):
2021-02-17 14:23:32 +01:00
self.stats_emitter = StatsEmitter(
self.config,
self.stats_tracking,
self.mqtt_client,
self.config.mqtt.topic_prefix,
self.stop_event,
)
self.stats_emitter.start()
2020-11-04 13:28:07 +01:00
def start_watchdog(self):
self.frigate_watchdog = FrigateWatchdog(self.detectors, self.stop_event)
self.frigate_watchdog.start()
def start(self):
self.init_logger()
2020-12-03 15:01:03 +01:00
try:
2020-12-05 16:47:43 +01:00
try:
self.init_config()
except Exception as e:
print(f"Error parsing config: {e}")
2020-12-05 16:47:43 +01:00
self.log_process.terminate()
sys.exit(1)
2021-01-16 04:33:53 +01:00
self.set_environment_vars()
2020-12-20 15:00:07 +01:00
self.ensure_dirs()
self.check_config()
2020-12-05 16:47:43 +01:00
self.set_log_levels()
self.init_queues()
self.init_database()
self.init_mqtt()
2020-12-03 15:01:03 +01:00
except Exception as e:
2020-12-20 15:00:07 +01:00
print(e)
2020-12-03 15:01:03 +01:00
self.log_process.terminate()
sys.exit(1)
2020-11-04 13:28:07 +01:00
self.start_detectors()
self.start_video_output_processor()
2020-11-04 13:28:07 +01:00
self.start_detected_frames_processor()
self.start_camera_processors()
self.start_camera_capture_processes()
self.init_stats()
2020-11-04 13:28:07 +01:00
self.init_web_server()
self.start_event_processor()
2020-11-24 14:27:51 +01:00
self.start_event_cleanup()
2020-11-30 04:31:02 +01:00
self.start_recording_maintainer()
self.start_stats_emitter()
2020-11-04 13:28:07 +01:00
self.start_watchdog()
2020-12-06 14:05:45 +01:00
# self.zeroconf = broadcast_zeroconf(self.config.mqtt.client_id)
2020-12-05 18:14:18 +01:00
def receiveSignal(signalNumber, frame):
self.stop()
sys.exit()
2021-02-17 14:23:32 +01:00
2020-12-05 18:14:18 +01:00
signal.signal(signal.SIGTERM, receiveSignal)
2021-02-17 14:23:32 +01:00
server = pywsgi.WSGIServer(
("127.0.0.1", 5001), self.flask_app, handler_class=WebSocketHandler
)
2021-05-18 07:52:08 +02:00
try:
server.serve_forever()
except KeyboardInterrupt:
pass
2021-02-13 15:09:44 +01:00
2020-11-04 13:28:07 +01:00
self.stop()
2021-02-17 14:23:32 +01:00
2020-11-04 13:28:07 +01:00
def stop(self):
logger.info(f"Stopping...")
self.stop_event.set()
self.detected_frames_processor.join()
self.event_processor.join()
2020-11-24 14:27:51 +01:00
self.event_cleanup.join()
2020-11-30 04:31:02 +01:00
self.recording_maintainer.join()
self.stats_emitter.join()
2020-11-04 13:28:07 +01:00
self.frigate_watchdog.join()
2021-02-07 15:38:35 +01:00
self.db.stop()
2020-11-04 13:28:07 +01:00
for detector in self.detectors.values():
detector.stop()
while len(self.detection_shms) > 0:
shm = self.detection_shms.pop()
shm.close()
shm.unlink()