Redesigned the logging to be more suitable for a library rather than a top level application.

Author: CreatingNull

CHANGES.rst

@@ -5,6 +5,11 @@ Version 0.5.0
 
 * BREAKING! Remaps address for redesign of UOS protocol.
 * PEP 484 implicit optional typing fixes.
+* Redesigned the logging to be more suitable for a library.
+  A global logger is configured at the top level and used throughout
+  the project.
+  The logger has a NullHandler so is suppressed unless the client enables.
+  Removed configure_logs function as this was no longer functional.
 
 Version 0.4.0
 -------------

uoshardware/__init__.py

@@ -1,7 +1,6 @@
 """The high level interface for communicating with UOS devices."""
+import logging
 from enum import Enum
-from logging import FileHandler, Formatter, getLogger
-from pathlib import Path
 
 __author__ = "Steve Richardson (Creating Null)"
 __copyright__ = f"2023, {__author__}"
@@ -27,23 +26,6 @@ class Loading(Enum):
     EAGER = 1
 
 
-# Dead code false positive as interface intended to be used by client.
-def configure_logs(name: str, level: int, base_path: Path):  # dead: disable
-    """Per-package logs must be manually configured to prefix correctly."""
-    logger = getLogger(name)
-    logger.setLevel(level)
-    # Don't capture to console as custom messages only, root logger captures stderr
-    logger.propagate = False
-    log_dir = Path(base_path.joinpath(Path("logs/")))
-    if not log_dir.exists():
-        log_dir.mkdir()
-    file_handler = FileHandler(log_dir.joinpath(Path(name + ".log")))
-    file_handler.setFormatter(
-        Formatter("%(asctime)s : %(levelname)s : %(name)s : %(message)s")
-    )
-    logger.addHandler(file_handler)
-
-
 class UOSError(Exception):
     """Base class exception for all UOS Interface Errors."""
 
@@ -54,3 +36,9 @@ class UOSUnsupportedError(UOSError):
 
 class UOSCommunicationError(UOSError):
     """Exception while communicating with a UOS Device."""
+
+
+# Configures the global logger for the library
+# Note: Clients need to initialize logging otherwise no output will be visible.
+logger = logging.getLogger(__name__)
+logger.addHandler(logging.NullHandler())

uoshardware/api.py

@@ -1,7 +1,5 @@
 """Provides the HAL layer for communicating with the hardware."""
-from logging import getLogger as Log
-
-from uoshardware import Loading, Persistence, UOSUnsupportedError
+from uoshardware import Loading, Persistence, UOSUnsupportedError, logger
 from uoshardware.abstractions import (
     ComResult,
     Device,
@@ -114,7 +112,7 @@ def __init__(
             self.loading == Loading.EAGER
         ):  # eager connections open when they are created
             self.open()
-        Log(__name__).debug("Created device %s", self.__device_interface.__repr__())
+        logger.debug("Created device %s", self.__device_interface.__repr__())
 
     def __enter__(self):
         """Dunder function for opening the interface as a context manager."""
@@ -264,7 +262,7 @@ def __execute_instruction(
             or instruction_data.volatility
             not in self.device.functions_enabled[function.name]
         ):
-            Log(__name__).debug(
+            logger.debug(
                 "Known functions %s", str(self.device.functions_enabled.keys())
             )
             raise UOSUnsupportedError(
@@ -299,7 +297,7 @@ def __execute_instruction(
                                     current_packet[1:-2]
                                 )
                             )
-                            Log(__name__).debug(
+                            logger.debug(
                                 "Calculated checksum %s must match rx %s",
                                 computed_checksum,
                                 current_packet[-2],

uoshardware/interface/serial.py

@@ -1,13 +1,12 @@
 """Module defining the low level UOSImplementation for serial port devices."""
 import platform
-from logging import getLogger as Log
 from time import sleep, time_ns
 
 import serial
 from serial.serialutil import SerialException
 from serial.tools import list_ports
 
-from uoshardware import UOSCommunicationError
+from uoshardware import UOSCommunicationError, logger
 from uoshardware.abstractions import ComResult, UOSInterface
 
 if platform.system() == "Linux":
@@ -38,44 +37,42 @@ def __init__(self, connection: str, **kwargs):
         self._port = self.check_port_exists(connection)
         self._kwargs = kwargs
         if self._port is None:
-            Log(__name__).error("%s port does not exist", connection)
+            logger.error("%s port does not exist", connection)
         else:
-            Log(__name__).debug("%s located", self._port)
+            logger.debug("%s located", self._port)
 
     def open(self):
         """Open a connection to the port and creates the device object."""
         try:
             self._port = self.check_port_exists(self._connection)
             if self._port is None:
-                Log(__name__).error(
-                    "%s device was not present to open", self._connection
-                )
+                logger.error("%s device was not present to open", self._connection)
                 raise UOSCommunicationError("Device could not be found on system.")
             self._device = serial.Serial()
             self._device.port = self._connection
             if "baudrate" in self._kwargs:
                 self._device.baudrate = self._kwargs["baudrate"]
             if platform.system() == "Linux":  # DTR transient workaround for Unix
-                Log(__name__).debug("Linux platform found so using DTR workaround")
+                logger.debug("Linux platform found so using DTR workaround")
                 with open(self._connection, mode="rb") as port:
                     attrs = termios.tcgetattr(port)
                     attrs[2] = attrs[2] & ~termios.HUPCL
                     termios.tcsetattr(port, termios.TCSAFLUSH, attrs)
             else:  # DTR transient workaround for Windows
                 self._device.dtr = False
             self._device.open()
-            Log(__name__).debug("%s opened successfully", self._port.device)
+            logger.debug("%s opened successfully", self._port.device)
             return
         except (SerialException, FileNotFoundError) as exception:
-            Log(__name__).error(
+            logger.error(
                 "Opening %s threw error %s",
                 self._port.device if self._port is not None else "None",
                 str(exception),
             )
             if (
                 exception.errno == 13
             ):  # permission denied another connection open to this device.
-                Log(__name__).error(
+                logger.error(
                     "Cannot open connection, account has insufficient permissions."
                 )
                 raise UOSCommunicationError(
@@ -92,11 +89,11 @@ def close(self):
         try:
             self._device.close()
         except SerialException as exception:
-            Log(__name__).debug("Closing the connection threw error %s", str(exception))
+            logger.debug("Closing the connection threw error %s", str(exception))
             raise UOSCommunicationError(
                 f"Closing connection threw error '{exception}'."
             ) from exception
-        Log(__name__).debug("Connection closed successfully")
+        logger.debug("Connection closed successfully")
         self._device = None
 
     # Kwargs is defined in the abstractmethod definition, false positive.
@@ -112,11 +109,11 @@ def execute_instruction(self, address, payload, **kwargs):  # dead: disable
                 "Connection must be open to execute instructions."
             )
         packet = self.get_npc_packet(to_addr=address, from_addr=0, payload=payload)
-        Log(__name__).debug("packet formed %s", packet)
+        logger.debug("packet formed %s", packet)
         try:  # Send the packet.
             num_bytes = self._device.write(packet)
             self._device.flush()
-            Log(__name__).debug("Sent %s bytes of data", num_bytes)
+            logger.debug("Sent %s bytes of data", num_bytes)
         except serial.SerialException as exception:
             raise UOSCommunicationError(
                 f"Executing instruction threw error '{exception}'"
@@ -163,7 +160,7 @@ def read_response(self, expect_packets: int, timeout_s: float):
                         packet = []
                     byte_index += 1
                 sleep(0.05)  # Don't churn CPU cycles waiting for data
-            Log(__name__).debug("Packet received %s", packet)
+            logger.debug("Packet received %s", packet)
             if expect_packets != packet_index or len(packet) < 6 or byte_index != -2:
                 response_object.rx_packets.append(packet)
                 response_object.exception = "did not receive all the expected data"
@@ -182,7 +179,7 @@ def hard_reset(self):
         """
         if self._device is None:
             raise UOSCommunicationError("Connection must be open to hard reset device.")
-        Log(__name__).debug("Resetting the device using the DTR line")
+        logger.debug("Resetting the device using the DTR line")
         self._device.dtr = not self._device.dtr
         sleep(0.2)
         self._device.dtr = not self._device.dtr
@@ -217,7 +214,7 @@ def decode_and_capture(
             if byte_in == b">":
                 byte_index += 1
         if byte_index >= 0:
-            Log(__name__).debug(
+            logger.debug(
                 "read %s byte index = %s",
                 byte_in,
                 byte_index,
@@ -226,7 +223,7 @@ def decode_and_capture(
             if byte_index == 3 + 2 + payload_len:  # End packet symbol
                 if byte_in == b"<":
                     byte_index = -2  # packet complete
-                    Log(__name__).debug("Found end packet symbol")
+                    logger.debug("Found end packet symbol")
                 else:  # Errored data
                     byte_index = -1
                     packet = []