diff --git a/modules/logger/logger.py b/modules/logger/logger.py index 8454adf..bcbd9c5 100644 --- a/modules/logger/logger.py +++ b/modules/logger/logger.py @@ -8,13 +8,14 @@ import os import pathlib import sys +import time # Used in type annotation of logger parameters # pylint: disable-next=unused-import import types +import cv2 import numpy as np -from PIL import Image from ..read_yaml import read_yaml @@ -66,48 +67,57 @@ def create(cls, name: str, enable_log_to_file: bool) -> "tuple[bool, Logger | No stream_handler.setFormatter(formatter) logger.addHandler(stream_handler) + if not enable_log_to_file: + return True, Logger(cls.__create_key, logger, None) + # Handles logging to file - if enable_log_to_file: - # Get the path to the logs directory. - entries = os.listdir(log_directory_path) - - if len(entries) == 0: - print("ERROR: The directory for this log session was not found.") - return False, None - - 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 - ), - ) - - filepath = pathlib.Path(log_directory_path, log_path, f"{name}.log") - try: - file = os.open(filepath, os.O_RDWR | os.O_EXCL | os.O_CREAT) - os.close(file) - except OSError: - print("ERROR: Log file already exists.") - return False, None - - file_handler = logging.FileHandler(filename=filepath, mode="w") - file_handler.setFormatter(formatter) - logger.addHandler(file_handler) - - return True, Logger(cls.__create_key, logger) - - def __init__(self, class_create_private_key: object, logger: logging.Logger) -> None: + + # Get the path to the logs directory. + entries = os.listdir(log_directory_path) + + if len(entries) == 0: + print("ERROR: The directory for this log session was not found.") + return False, None + + 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 + ), + ) + + filepath = pathlib.Path(log_directory_path, log_path, f"{name}.log") + try: + file = os.open(filepath, os.O_RDWR | os.O_EXCL | os.O_CREAT) + os.close(file) + except OSError: + print("ERROR: Log file already exists.") + return False, None + + file_handler = logging.FileHandler(filename=filepath, mode="w") + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + return True, Logger(cls.__create_key, logger, pathlib.Path(log_directory_path, log_path)) + + def __init__( + self, + class_create_private_key: object, + logger: logging.Logger, + maybe_log_directory: pathlib.Path | None, + ) -> None: """ Private constructor, use create() method. """ assert class_create_private_key is Logger.__create_key, "Use create() method." self.logger = logger + self.__maybe_log_directory = maybe_log_directory @staticmethod def message_and_metadata(message: str, frame: "types.FrameType | None") -> str: @@ -179,8 +189,7 @@ def critical(self, message: str, log_with_frame_info: bool = True) -> None: def save_image( self, image: np.ndarray, - filename: str, - log_info_message: bool = False, + filename: str = "", log_with_frame_info: bool = True, ) -> None: """ @@ -191,14 +200,20 @@ def save_image( filename: The filename to save the image as. log_with_frame_info: Whether to log the frame info. """ - img = Image.fromarray(image) + if self.__maybe_log_directory is None: + self.logger.warning("Image not saved: Logger not set up with file logging") + return + + # Get Pylance to stop complaining + assert self.__maybe_log_directory is not None + + full_file_name = ( + f"{self.logger.name}_{int(time.time())}_{filename}.png" + if filename != "" + else f"{self.logger.name}_{int(time.time())}.png" + ) + filepath = pathlib.Path(self.__maybe_log_directory, full_file_name) - img.save(filename) + cv2.imwrite(str(filepath), image) - if log_info_message: - message = f"{filename} saved" - if log_with_frame_info: - logger_frame = inspect.currentframe() - caller_frame = logger_frame.f_back - filename = self.message_and_metadata(message, caller_frame) - self.logger.info(message) + self.info(f"Image saved as: {full_file_name}", log_with_frame_info) diff --git a/modules/logger/logger_main_setup.py b/modules/logger/logger_main_setup.py index 3a08932..ca56717 100644 --- a/modules/logger/logger_main_setup.py +++ b/modules/logger/logger_main_setup.py @@ -9,11 +9,15 @@ MAIN_LOGGER_NAME = "main" +MAX_ATTEMPTS = 3 def setup_main_logger( - config: "dict", main_logger_name: str = MAIN_LOGGER_NAME, enable_log_to_file: bool = True -) -> "tuple[bool, logger.Logger | None, pathlib.Path | None]": + config: "dict", + main_logger_name: str = MAIN_LOGGER_NAME, + enable_log_to_file: bool = True, + max_attempts: int = MAX_ATTEMPTS, +) -> tuple[True, logger.Logger, pathlib.Path] | tuple[False, None, None]: """ Setup prerequisites for logging in `main()` . @@ -25,15 +29,26 @@ def setup_main_logger( try: log_directory_path = config["logger"]["directory_path"] log_path_format = config["logger"]["file_datetime_format"] - start_time = datetime.datetime.now().strftime(log_path_format) except KeyError as exception: print(f"ERROR: Config key(s) not found: {exception}") return False, None, None - logging_path = pathlib.Path(log_directory_path, start_time) - # Create logging directory - logging_path.mkdir(exist_ok=True, parents=True) + start_time = datetime.datetime.now() + success = False + for i in range(0, max_attempts): + offset = datetime.timedelta(seconds=i) + logging_path = pathlib.Path( + log_directory_path, (start_time + offset).strftime(log_path_format) + ) + if not logging_path.exists(): + success = True + break + + if not success: + print("ERROR: Could not create new log directory") + + logging_path.mkdir(exist_ok=False, parents=True) # Setup logger result, main_logger = logger.Logger.create(main_logger_name, enable_log_to_file) diff --git a/tests/unit/test_logger.py b/tests/unit/test_logger.py index 43d2fc5..780f001 100644 --- a/tests/unit/test_logger.py +++ b/tests/unit/test_logger.py @@ -6,6 +6,8 @@ import pathlib import re +import cv2 +import numpy as np import pytest from modules.logger import logger @@ -14,16 +16,21 @@ @pytest.fixture -def main_logger_instance_and_log_file_path() -> logger.Logger: # type: ignore +def main_logger_instance_and_logging_path() -> tuple[logger.Logger, pathlib.Path]: # type: ignore """ Returns the main logger with logging to file enabled and sets up logging directory. """ result, config = read_yaml.open_config(logger.CONFIG_FILE_PATH) assert result + assert config is not None - result, instance, log_file_path = logger_main_setup.setup_main_logger(config=config) + # Increase max attempts for every use of this fixture + result, instance, logging_path = logger_main_setup.setup_main_logger(config, max_attempts=2) assert result - yield instance, log_file_path + assert instance is not None + assert logging_path is not None + + yield instance, logging_path @pytest.fixture @@ -33,6 +40,8 @@ def logger_instance_to_file_enabled() -> logger.Logger: # type: ignore """ result, instance = logger.Logger.create("test_logger_to_file_enabled", True) assert result + assert instance is not None + yield instance @@ -58,7 +67,7 @@ def test_message_and_metadata_with_frame(self) -> None: frame = inspect.currentframe() message = "Test message" expected = ( - f"[{__file__} | {self.test_message_and_metadata_with_frame.__name__} | 65] Test message" + f"[{__file__} | {self.test_message_and_metadata_with_frame.__name__} | 74] Test message" ) # Get line number of this function call @@ -108,14 +117,14 @@ def test_log_with_frame_info( def test_log_to_file( self, - main_logger_instance_and_log_file_path: "tuple[logger.Logger | None, pathlib.Path | None]", + main_logger_instance_and_logging_path: tuple[logger.Logger, pathlib.Path], logger_instance_to_file_enabled: logger.Logger, ) -> None: """ Test if messages are logged to file All levels are done in one test since they will all be logged to the same file """ - main_logger_instance, log_file_path = main_logger_instance_and_log_file_path + main_logger_instance, logging_path = main_logger_instance_and_logging_path main_message = "main message" main_logger_instance.debug(main_message, False) @@ -131,13 +140,13 @@ def test_log_to_file( logger_instance_to_file_enabled.error(test_message, False) logger_instance_to_file_enabled.critical(test_message, False) - main_log_file_path = pathlib.Path(log_file_path, "main.log") - test_log_file_path = pathlib.Path(log_file_path, "test_logger_to_file_enabled.log") + main_logging_path = pathlib.Path(logging_path, "main.log") + test_logging_path = pathlib.Path(logging_path, "test_logger_to_file_enabled.log") - with open(main_log_file_path, "r", encoding="utf8") as log_file: + with open(main_logging_path, "r", encoding="utf8") as log_file: actual_main = log_file.read() - with open(test_log_file_path, "r", encoding="utf8") as log_file: + with open(test_logging_path, "r", encoding="utf8") as log_file: actual_test = log_file.read() for level in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]: @@ -222,3 +231,30 @@ def test_debug_log_critical_to_stdout( expected_pattern = re.compile(r"CRITICAL.*" + re.escape(test_message)) assert re.search(expected_pattern, actual) is not None + + def test_log_images( + self, + main_logger_instance_and_logging_path: tuple[logger.Logger, pathlib.Path], + logger_instance_to_file_enabled: logger.Logger, + ) -> None: + """ + Test logging images. + """ + # Setup + expected_image = np.empty((480, 640, 3), dtype=np.uint8) + expected_image.fill(255) + + main_logger_instance, logging_path = main_logger_instance_and_logging_path + + # Run + main_logger_instance.save_image(expected_image) + logger_instance_to_file_enabled.save_image(expected_image) + + # Check + image_paths = list(logging_path.glob("*.png")) + assert len(image_paths) == 2 + + for image_path in image_paths: + actual_image = cv2.imread(image_path) + + assert np.array_equal(actual_image, expected_image)