naming threads and processes for logs

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 frigate.app import FrigateApp
-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__)
 
 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()
 

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()

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()
+        self.out_shm.unlink()

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

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()

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
+        threading.current_thread().name = "mqtt"
-        logger.info("On connect called")
         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)

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

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']

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