Logging image efficiency and test (#68)

* Logging image efficiency and test

* User settable max attempts

* PR comment

* Don't use 0 as default

* Better conditional organization

* Type annotation

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)

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)

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)