Logging module

config.yaml

@@ -2,7 +2,11 @@
 
 queue_max_size: 10
 
-log_directory_path: "logs"
+logger:
+    directory_path: "logs"
+    file_datetime_format: "%Y-%m-%d_%H-%M-%S"
+    format: "%(asctime)s: [%(levelname)s] %(message)s"
+    datetime_format: "%I:%M:%S"
 
 video_input:
     camera_name: 0

main_2024.py

@@ -3,6 +3,8 @@
 """
 
 import argparse
+import datetime
+import inspect
 import multiprocessing as mp
 import pathlib
 import queue
@@ -19,6 +21,7 @@
 from modules.data_merge import data_merge_worker
 from modules.geolocation import geolocation_worker
 from modules.geolocation import camera_properties
+from modules.logger import logger
 from utilities.workers import queue_proxy_wrapper
 from utilities.workers import worker_controller
 from utilities.workers import worker_manager
@@ -64,18 +67,23 @@ def main() -> int:
         QUEUE_MAX_SIZE = config["queue_max_size"]
 
         LOG_DIRECTORY_PATH = config["log_directory_path"]
+        start_time = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
 
         VIDEO_INPUT_CAMERA_NAME = config["video_input"]["camera_name"]
         VIDEO_INPUT_WORKER_PERIOD = config["video_input"]["worker_period"]
         VIDEO_INPUT_SAVE_NAME_PREFIX = config["video_input"]["save_prefix"]
-        VIDEO_INPUT_SAVE_PREFIX = f"{LOG_DIRECTORY_PATH}/{VIDEO_INPUT_SAVE_NAME_PREFIX}"
+        VIDEO_INPUT_SAVE_PREFIX = (
+            f"{LOG_DIRECTORY_PATH}/{start_time}/{VIDEO_INPUT_SAVE_NAME_PREFIX}"
+        )
 
         DETECT_TARGET_WORKER_COUNT = config["detect_target"]["worker_count"]
         DETECT_TARGET_DEVICE = "cpu" if args.cpu else config["detect_target"]["device"]
         DETECT_TARGET_MODEL_PATH = config["detect_target"]["model_path"]
         DETECT_TARGET_OVERRIDE_FULL_PRECISION = args.full
         DETECT_TARGET_SAVE_NAME_PREFIX = config["detect_target"]["save_prefix"]
-        DETECT_TARGET_SAVE_PREFIX = f"{LOG_DIRECTORY_PATH}/{DETECT_TARGET_SAVE_NAME_PREFIX}"
+        DETECT_TARGET_SAVE_PREFIX = (
+            f"{LOG_DIRECTORY_PATH}/{start_time}/{DETECT_TARGET_SAVE_NAME_PREFIX}"
+        )
         DETECT_TARGET_SHOW_ANNOTATED = args.show_annotated
 
         FLIGHT_INTERFACE_ADDRESS = config["flight_interface"]["address"]
@@ -100,6 +108,12 @@ def main() -> int:
         return -1
 
     pathlib.Path(LOG_DIRECTORY_PATH).mkdir(exist_ok=True)
+    pathlib.Path(f"{LOG_DIRECTORY_PATH}/{start_time}").mkdir()
+
+    result, main_logger = logger.Logger.create("main")
+    if result:
+        frame = inspect.currentframe()
+        main_logger.info("main logger initialized", frame)
 
     # Setup
     controller = worker_controller.WorkerController()

modules/logger/logger.py

@@ -0,0 +1,142 @@
+"""
+Logs debug messages.
+"""
+
+import datetime
+import inspect
+import logging
+import pathlib
+import os
+import types
+import typing
+import yaml
+
+CONFIG_FILE_PATH = pathlib.Path("config.yaml")
+
+
+class Logger:
+    """
+    Instantiates Logger objects.
+    """
+
+    __create_key = object()
+
+    @classmethod
+    def create(cls, name: str) -> "tuple[bool, Logger | None]":
+        """
+        Create and configure a logger.
+        """
+
+        # Open config file.
+        try:
+            with CONFIG_FILE_PATH.open("r", encoding="utf8") as file:
+                try:
+                    config = yaml.safe_load(file)
+                except yaml.YAMLError as exc:
+                    print(f"Error parsing YAML file: {exc}")
+                    return -1
+        except FileNotFoundError:
+            print(f"File not found: {CONFIG_FILE_PATH}")
+            return False, None
+        except IOError as exc:
+            print(f"Error when opening file: {exc}")
+            return False, None
+
+        try:
+            log_directory_path = config["logger"]["directory_path"]
+            file_datetime_format = config["logger"]["file_datetime_format"]
+            logger_format = config["logger"]["format"]
+            logger_datetime_format = config["logger"]["datetime_format"]
+        except KeyError:
+            print("Config key(s) not found")
+            return False, None
+
+        # Get the path to the logs directory.
+        entries = os.listdir(log_directory_path)
+        log_names = [
+            entry for entry in entries if os.path.isdir(os.path.join(log_directory_path, entry))
+        ]
+
+        # Find the log directory for the current run, which is the most recent timestamp.
+        log_path = max(
+            log_names,
+            key=lambda datetime_string: datetime.datetime.strptime(
+                datetime_string, file_datetime_format
+            ),
+        )
+
+        filename = f"{log_directory_path}/{log_path}/{name}.log"
+
+        # Formatting configurations for the logger.
+        file_handler = logging.FileHandler(filename=filename, mode="w")  # Handles logging to file.
+        stream_handler = logging.StreamHandler()  # Handles logging to terminal.
+
+        formatter = logging.Formatter(
+            fmt=logger_format,
+            datefmt=logger_datetime_format,
+        )
+
+        file_handler.setFormatter(formatter)
+        stream_handler.setFormatter(formatter)
+
+        # Create a unique logger instance and configure it.
+        logger = logging.getLogger(name)
+        logger.setLevel(logging.DEBUG)
+        logger.addHandler(file_handler)
+        logger.addHandler(stream_handler)
+
+        return True, Logger(cls.__create_key, logger)
+
+    def __init__(self, class_create_private_key: object, logger: logging.Logger) -> None:
+        """
+        Private constructor, use create() method.
+        """
+        assert class_create_private_key is Logger.__create_key, "Use create() method."
+
+        self.logger = logger
+
+    @staticmethod
+    def message_and_metadata(message: str, frame: typing.Optional[types.FrameType]) -> str:
+        """
+        Extracts metadata from frame and appends it to the message.
+        """
+        function_name = frame.f_code.co_name
+        filename = frame.f_code.co_filename
+        line_number = inspect.getframeinfo(frame).lineno
+
+        return f"[{filename} | {function_name} | {line_number}] {message}"
+
+    def debug(self, message: str, frame: typing.Optional[types.FrameType]) -> None:
+        """
+        Logs a debug level message.
+        """
+        message = self.message_and_metadata(message, frame)
+        self.logger.debug(message)
+
+    def info(self, message: str, frame: typing.Optional[types.FrameType]) -> None:
+        """
+        Logs an info level message.
+        """
+        message = self.message_and_metadata(message, frame)
+        self.logger.info(message)
+
+    def warning(self, message: str, frame: typing.Optional[types.FrameType]) -> None:
+        """
+        Logs a warning level message.
+        """
+        message = self.message_and_metadata(message, frame)
+        self.logger.warning(message)
+
+    def error(self, message: str, frame: typing.Optional[types.FrameType]) -> None:
+        """
+        Logs an error level message.
+        """
+        message = self.message_and_metadata(message, frame)
+        self.logger.error(message)
+
+    def critical(self, message: str, frame: typing.Optional[types.FrameType]) -> None:
+        """
+        Logs a critical level message.
+        """
+        message = self.message_and_metadata(message, frame)
+        self.logger.critical(message)