diff --git a/frigate/__main__.py b/frigate/__main__.py index 7d24f07cd..af4ec11ac 100644 --- a/frigate/__main__.py +++ b/frigate/__main__.py @@ -1,161 +1,13 @@ import faulthandler; faulthandler.enable() -import os -import json -import logging -import yaml -import multiprocessing as mp import sys +import threading +threading.current_thread().name = "frigate" -from logging.handlers import QueueHandler -from playhouse.sqlite_ext import SqliteExtDatabase -from typing import Dict, List - -from frigate.config import FrigateConfig -from frigate.edgetpu import EdgeTPUProcess -from frigate.events import EventProcessor -from frigate.http import create_app -from frigate.log import root_configurer, log_process -from frigate.models import Event -from frigate.mqtt import create_mqtt_client -from frigate.object_processing import TrackedObjectProcessor -from frigate.video import track_camera, capture_camera -from frigate.watchdog import FrigateWatchdog - -logger = logging.getLogger(__name__) +from frigate.app import FrigateApp cli = sys.modules['flask.cli'] cli.show_server_banner = lambda *x: None -class FrigateApp(): - 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 = {} - - def init_logger(self): - self.log_process = mp.Process(target=log_process, args=(self.log_queue,)) - self.log_process.start() - root_configurer(self.log_queue) - - def init_config(self): - config_file = os.environ.get('CONFIG_FILE', '/config/config.yml') - self.config = FrigateConfig(config_file=config_file) - - for camera_name in self.config.cameras.keys(): - # create camera_metrics - self.camera_metrics[camera_name] = { - 'camera_fps': mp.Value('d', 0.0), - 'skipped_fps': mp.Value('d', 0.0), - 'process_fps': mp.Value('d', 0.0), - '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) - } - - def init_queues(self): - # Queue for clip processing - self.event_queue = mp.Queue() - - # Queue for cameras to push tracked objects to - self.detected_frames_queue = mp.Queue(maxsize=len(self.config.cameras.keys())*2) - - def init_database(self): - self.db = SqliteExtDatabase(f"/{os.path.join(self.config.save_clips.clips_dir, 'frigate.db')}") - models = [Event] - self.db.bind(models) - self.db.create_tables(models, safe=True) - - def init_web_server(self): - self.flask_app = create_app(self.config, self.db, self.camera_metrics, self.detectors, self.detected_frames_processor) - - def init_mqtt(self): - self.mqtt_client = create_mqtt_client(self.config.mqtt) - - def start_detectors(self): - for name in self.config.cameras.keys(): - self.detection_out_events[name] = mp.Event() - shm_in = mp.shared_memory.SharedMemory(name=name, create=True, size=300*300*3) - shm_out = mp.shared_memory.SharedMemory(name=f"out-{name}", create=True, size=20*6*4) - self.detection_shms.append(shm_in) - self.detection_shms.append(shm_out) - - for name, detector in self.config.detectors.items(): - if detector.type == 'cpu': - self.detectors[name] = EdgeTPUProcess(self.detection_queue, out_events=self.detection_out_events, tf_device='cpu') - if detector.type == 'edgetpu': - self.detectors[name] = EdgeTPUProcess(self.detection_queue, out_events=self.detection_out_events, tf_device=detector.device) - - def start_detected_frames_processor(self): - self.detected_frames_processor = TrackedObjectProcessor(self.config.cameras, self.mqtt_client, self.config.mqtt.topic_prefix, - self.detected_frames_queue, self.event_queue, self.stop_event) - self.detected_frames_processor.start() - - def start_camera_processors(self): - for name, config in self.config.cameras.items(): - camera_process = mp.Process(target=track_camera, args=(name, config, - self.detection_queue, self.detection_out_events[name], self.detected_frames_queue, - self.camera_metrics[name])) - camera_process.daemon = True - self.camera_metrics[name]['process'] = camera_process - 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(): - capture_process = mp.Process(target=capture_camera, args=(name, config, - self.camera_metrics[name])) - capture_process.daemon = True - self.camera_metrics[name]['capture_process'] = capture_process - capture_process.start() - logger.info(f"Capture process started for {name}: {capture_process.pid}") - - def start_event_processor(self): - self.event_processor = EventProcessor(self.config, self.camera_metrics, self.event_queue, self.stop_event) - self.event_processor.start() - - def start_watchdog(self): - self.frigate_watchdog = FrigateWatchdog(self.detectors, self.stop_event) - self.frigate_watchdog.start() - - def start(self): - self.init_logger() - self.init_config() - self.init_queues() - self.init_database() - self.init_mqtt() - self.start_detectors() - self.start_detected_frames_processor() - self.start_camera_processors() - self.start_camera_capture_processes() - self.init_web_server() - self.start_event_processor() - self.start_watchdog() - self.flask_app.run(host='0.0.0.0', port=self.config.web_port, debug=False) - self.stop() - - def stop(self): - logger.info(f"Stopping...") - self.stop_event.set() - - self.detected_frames_processor.join() - self.event_processor.join() - self.frigate_watchdog.join() - - for detector in self.detectors.values(): - detector.stop() - - while len(self.detection_shms) > 0: - shm = self.detection_shms.pop() - shm.close() - shm.unlink() - if __name__ == '__main__': frigate_app = FrigateApp() diff --git a/frigate/app.py b/frigate/app.py new file mode 100644 index 000000000..d8ef000a3 --- /dev/null +++ b/frigate/app.py @@ -0,0 +1,152 @@ +import json +import logging +import multiprocessing as mp +import os +from logging.handlers import QueueHandler +from typing import Dict, List + +import yaml +from playhouse.sqlite_ext import SqliteExtDatabase + +from frigate.config import FrigateConfig +from frigate.edgetpu import EdgeTPUProcess +from frigate.events import EventProcessor +from frigate.http import create_app +from frigate.log import log_process, root_configurer +from frigate.models import Event +from frigate.mqtt import create_mqtt_client +from frigate.object_processing import TrackedObjectProcessor +from frigate.video import capture_camera, track_camera +from frigate.watchdog import FrigateWatchdog + +logger = logging.getLogger(__name__) + +class FrigateApp(): + 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 = {} + + def init_logger(self): + self.log_process = mp.Process(target=log_process, args=(self.log_queue,), name='log_process') + self.log_process.start() + root_configurer(self.log_queue) + + def init_config(self): + config_file = os.environ.get('CONFIG_FILE', '/config/config.yml') + self.config = FrigateConfig(config_file=config_file) + + for camera_name in self.config.cameras.keys(): + # create camera_metrics + self.camera_metrics[camera_name] = { + 'camera_fps': mp.Value('d', 0.0), + 'skipped_fps': mp.Value('d', 0.0), + 'process_fps': mp.Value('d', 0.0), + '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) + } + + def init_queues(self): + # Queue for clip processing + self.event_queue = mp.Queue() + + # Queue for cameras to push tracked objects to + self.detected_frames_queue = mp.Queue(maxsize=len(self.config.cameras.keys())*2) + + def init_database(self): + self.db = SqliteExtDatabase(f"/{os.path.join(self.config.save_clips.clips_dir, 'frigate.db')}") + models = [Event] + self.db.bind(models) + self.db.create_tables(models, safe=True) + + def init_web_server(self): + self.flask_app = create_app(self.config, self.db, self.camera_metrics, self.detectors, self.detected_frames_processor) + + def init_mqtt(self): + self.mqtt_client = create_mqtt_client(self.config.mqtt) + + def start_detectors(self): + for name in self.config.cameras.keys(): + self.detection_out_events[name] = mp.Event() + shm_in = mp.shared_memory.SharedMemory(name=name, create=True, size=300*300*3) + shm_out = mp.shared_memory.SharedMemory(name=f"out-{name}", create=True, size=20*6*4) + self.detection_shms.append(shm_in) + self.detection_shms.append(shm_out) + + for name, detector in self.config.detectors.items(): + if detector.type == 'cpu': + self.detectors[name] = EdgeTPUProcess(name, self.detection_queue, out_events=self.detection_out_events, tf_device='cpu') + if detector.type == 'edgetpu': + self.detectors[name] = EdgeTPUProcess(name, self.detection_queue, out_events=self.detection_out_events, tf_device=detector.device) + + def start_detected_frames_processor(self): + self.detected_frames_processor = TrackedObjectProcessor(self.config.cameras, self.mqtt_client, self.config.mqtt.topic_prefix, + self.detected_frames_queue, self.event_queue, self.stop_event) + self.detected_frames_processor.start() + + def start_camera_processors(self): + for name, config in self.config.cameras.items(): + camera_process = mp.Process(target=track_camera, name=f"camera_processor:{name}", args=(name, config, + self.detection_queue, self.detection_out_events[name], self.detected_frames_queue, + self.camera_metrics[name])) + camera_process.daemon = True + self.camera_metrics[name]['process'] = camera_process + 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(): + capture_process = mp.Process(target=capture_camera, name=f"camera_capture:{name}", args=(name, config, + self.camera_metrics[name])) + capture_process.daemon = True + self.camera_metrics[name]['capture_process'] = capture_process + capture_process.start() + logger.info(f"Capture process started for {name}: {capture_process.pid}") + + def start_event_processor(self): + self.event_processor = EventProcessor(self.config, self.camera_metrics, self.event_queue, self.stop_event) + self.event_processor.start() + + def start_watchdog(self): + self.frigate_watchdog = FrigateWatchdog(self.detectors, self.stop_event) + self.frigate_watchdog.start() + + def start(self): + self.init_logger() + self.init_config() + self.init_queues() + self.init_database() + self.init_mqtt() + self.start_detectors() + self.start_detected_frames_processor() + self.start_camera_processors() + self.start_camera_capture_processes() + self.init_web_server() + self.start_event_processor() + self.start_watchdog() + self.flask_app.run(host='0.0.0.0', port=self.config.web_port, debug=False) + self.stop() + + def stop(self): + logger.info(f"Stopping...") + self.stop_event.set() + + self.detected_frames_processor.join() + self.event_processor.join() + self.frigate_watchdog.join() + + for detector in self.detectors.values(): + detector.stop() + + while len(self.detection_shms) > 0: + shm = self.detection_shms.pop() + shm.close() + shm.unlink() diff --git a/frigate/edgetpu.py b/frigate/edgetpu.py index 103fba817..d810e87c4 100644 --- a/frigate/edgetpu.py +++ b/frigate/edgetpu.py @@ -1,16 +1,19 @@ -import os import datetime import hashlib -import multiprocessing as mp -import queue import logging -from multiprocessing.connection import Connection +import multiprocessing as mp +import os +import queue +import threading from abc import ABC, abstractmethod +from multiprocessing.connection import Connection from typing import Dict + import numpy as np import tflite_runtime.interpreter as tflite from tflite_runtime.interpreter import load_delegate -from frigate.util import EventsPerSecond, listen, SharedMemoryFrameManager + +from frigate.util import EventsPerSecond, SharedMemoryFrameManager, listen logger = logging.getLogger(__name__) @@ -102,7 +105,8 @@ class LocalObjectDetector(ObjectDetector): return detections -def run_detector(detection_queue: mp.Queue, out_events: Dict[str, mp.Event], avg_speed, start, tf_device): +def run_detector(name: str, detection_queue: mp.Queue, out_events: Dict[str, mp.Event], avg_speed, start, tf_device): + threading.current_thread().name = f"detector:{name}" logging.info(f"Starting detection process: {os.getpid()}") listen() frame_manager = SharedMemoryFrameManager() @@ -135,7 +139,8 @@ def run_detector(detection_queue: mp.Queue, out_events: Dict[str, mp.Event], avg avg_speed.value = (avg_speed.value*9 + duration)/10 class EdgeTPUProcess(): - def __init__(self, detection_queue, out_events, tf_device=None): + def __init__(self, name, detection_queue, out_events, tf_device=None): + self.name = name self.out_events = out_events self.detection_queue = detection_queue self.avg_inference_speed = mp.Value('d', 0.01) @@ -157,7 +162,7 @@ class EdgeTPUProcess(): self.detection_start.value = 0.0 if (not self.detect_process is None) and self.detect_process.is_alive(): self.stop() - self.detect_process = mp.Process(target=run_detector, args=(self.detection_queue, self.out_events, self.avg_inference_speed, self.detection_start, self.tf_device)) + self.detect_process = mp.Process(target=run_detector, name=f"detector:{self.name}", args=(self.name, self.detection_queue, self.out_events, self.avg_inference_speed, self.detection_start, self.tf_device)) self.detect_process.daemon = True self.detect_process.start() @@ -199,4 +204,4 @@ class RemoteObjectDetector(): def cleanup(self): self.shm.unlink() - self.out_shm.unlink() \ No newline at end of file + self.out_shm.unlink() diff --git a/frigate/events.py b/frigate/events.py index 98e331bc4..c0a1d7446 100644 --- a/frigate/events.py +++ b/frigate/events.py @@ -16,6 +16,7 @@ logger = logging.getLogger(__name__) class EventProcessor(threading.Thread): def __init__(self, config, camera_processes, event_queue, stop_event): threading.Thread.__init__(self) + self.name = 'event_processor' self.config = config self.cache_dir = self.config.save_clips.cache_dir self.clips_dir = self.config.save_clips.clips_dir diff --git a/frigate/log.py b/frigate/log.py index f562c5018..290da1072 100644 --- a/frigate/log.py +++ b/frigate/log.py @@ -6,7 +6,7 @@ from logging import handlers def listener_configurer(): root = logging.getLogger() console_handler = logging.StreamHandler() - formatter = logging.Formatter('%(processName)-12s %(threadName)-12s %(name)-16s %(levelname)-8s: %(message)s') + formatter = logging.Formatter('%(threadName)-25s %(name)-16s %(levelname)-8s: %(message)s') console_handler.setFormatter(formatter) root.addHandler(console_handler) root.setLevel(logging.INFO) @@ -18,6 +18,7 @@ def root_configurer(queue): root.setLevel(logging.INFO) def log_process(queue): + threading.current_thread().name = f"logger" listener_configurer() while True: record = queue.get() diff --git a/frigate/mqtt.py b/frigate/mqtt.py index 858093f41..bb5f15b7a 100644 --- a/frigate/mqtt.py +++ b/frigate/mqtt.py @@ -1,5 +1,6 @@ import logging import paho.mqtt.client as mqtt +import threading from frigate.config import MqttConfig @@ -8,8 +9,7 @@ logger = logging.getLogger(__name__) def create_mqtt_client(config: MqttConfig): client = mqtt.Client(client_id=config.client_id) def on_connect(client, userdata, flags, rc): - # TODO: use logging library - logger.info("On connect called") + threading.current_thread().name = "mqtt" if rc != 0: if rc == 3: logger.error("MQTT Server unavailable") @@ -19,6 +19,8 @@ def create_mqtt_client(config: MqttConfig): logger.error("MQTT Not authorized") else: logger.error("Unable to connect to MQTT: Connection refused. Error code: " + str(rc)) + + logger.info("MQTT connected") client.publish(config.topic_prefix+'/available', 'online', retain=True) client.on_connect = on_connect client.will_set(config.topic_prefix+'/available', payload='offline', qos=1, retain=True) diff --git a/frigate/object_processing.py b/frigate/object_processing.py index d05a4de29..3cf793ed6 100644 --- a/frigate/object_processing.py +++ b/frigate/object_processing.py @@ -255,6 +255,7 @@ class CameraState(): class TrackedObjectProcessor(threading.Thread): def __init__(self, camera_config: Dict[str, CameraConfig], client, topic_prefix, tracked_objects_queue, event_queue, stop_event): threading.Thread.__init__(self) + self.name = "detected_frames_processor" self.camera_config = camera_config self.client = client self.topic_prefix = topic_prefix diff --git a/frigate/video.py b/frigate/video.py index f850b25fd..d9be271f3 100755 --- a/frigate/video.py +++ b/frigate/video.py @@ -127,9 +127,10 @@ def capture_frames(ffmpeg_process, camera_name, frame_shape, frame_manager: Fram frame_queue.put(current_frame.value) class CameraWatchdog(threading.Thread): - def __init__(self, name, config, frame_queue, camera_fps, ffmpeg_pid): + def __init__(self, camera_name, config, frame_queue, camera_fps, ffmpeg_pid): threading.Thread.__init__(self) - self.name = name + self.name = f"watchdog:{camera_name}" + self.camera_name = camera_name self.config = config self.capture_thread = None self.ffmpeg_process = None @@ -148,7 +149,7 @@ class CameraWatchdog(threading.Thread): if not self.capture_thread.is_alive(): self.start_ffmpeg() elif now - self.capture_thread.current_frame.value > 5: - logger.info(f"No frames received from {self.name} in 5 seconds. Exiting ffmpeg...") + logger.info(f"No frames received from {self.camera_name} in 5 seconds. Exiting ffmpeg...") self.ffmpeg_process.terminate() try: logger.info("Waiting for ffmpeg to exit gracefully...") @@ -164,14 +165,15 @@ class CameraWatchdog(threading.Thread): def start_ffmpeg(self): self.ffmpeg_process = start_or_restart_ffmpeg(self.config.ffmpeg_cmd, self.frame_size) self.ffmpeg_pid.value = self.ffmpeg_process.pid - self.capture_thread = CameraCapture(self.name, self.ffmpeg_process, self.frame_shape, self.frame_queue, + self.capture_thread = CameraCapture(self.camera_name, self.ffmpeg_process, self.frame_shape, self.frame_queue, self.camera_fps) self.capture_thread.start() class CameraCapture(threading.Thread): - def __init__(self, name, ffmpeg_process, frame_shape, frame_queue, fps): + def __init__(self, camera_name, ffmpeg_process, frame_shape, frame_queue, fps): threading.Thread.__init__(self) - self.name = name + self.name = f"capture:{camera_name}" + self.camera_name = camera_name self.frame_shape = frame_shape self.frame_queue = frame_queue self.fps = fps @@ -183,7 +185,7 @@ class CameraCapture(threading.Thread): def run(self): self.skipped_fps.start() - capture_frames(self.ffmpeg_process, self.name, self.frame_shape, self.frame_manager, self.frame_queue, + capture_frames(self.ffmpeg_process, self.camera_name, self.frame_shape, self.frame_manager, self.frame_queue, self.fps, self.skipped_fps, self.current_frame) def capture_camera(name, config: CameraConfig, process_info): @@ -193,6 +195,7 @@ def capture_camera(name, config: CameraConfig, process_info): camera_watchdog.join() def track_camera(name, config: CameraConfig, detection_queue, result_connection, detected_objects_queue, process_info): + threading.current_thread().name = f"process:{name}" listen() frame_queue = process_info['frame_queue'] diff --git a/frigate/watchdog.py b/frigate/watchdog.py index f12268dd4..c209dd2c1 100644 --- a/frigate/watchdog.py +++ b/frigate/watchdog.py @@ -8,6 +8,7 @@ logger = logging.getLogger(__name__) class FrigateWatchdog(threading.Thread): def __init__(self, detectors, stop_event): threading.Thread.__init__(self) + self.name = 'frigate_watchdog' self.detectors = detectors self.stop_event = stop_event