Logger for UDS Communication (#434)

UDS communication logger introduced.
Documentation and examples added.

Author: mdabrowski1990

docs/source/pages/examples.rst

@@ -8,6 +8,7 @@ and configurations.
 .. toctree::
 
   examples/client.rst
+  examples/transport.rst
   examples/CAN/can.rst
 
 .. seealso:: https://github.com/mdabrowski1990/uds/tree/main/examples

docs/source/pages/examples/transport.rst

@@ -0,0 +1,25 @@
+Transport
+=========
+Examples how to use features related to Transport layer.
+
+.. seealso:: https://github.com/mdabrowski1990/uds/tree/main/examples/transport
+
+
+.. _example-transport-logger-instance:
+
+Setup logger to Transport Interface instance
+--------------------------------------------
+Example how to configure and activate Transport Logger for Transport Interface instance.
+
+.. include:: ../../../../examples/transport/instance_logger.py
+  :code: python
+
+
+.. _example-transport-logger-class:
+
+Setup logger to Transport Interface class
+-----------------------------------------
+Example how to configure and activate Transport Logger for Transport Interface class.
+
+.. include:: ../../../../examples/transport/class_logger.py
+  :code: python

docs/source/pages/user_guide.rst

@@ -11,5 +11,6 @@ for getting started.
   user_guide/addressing.rst
   user_guide/client.rst
   user_guide/message_translation.rst
+  user_guide/logging.rst
   user_guide/can.rst
   user_guide/custom.rst

docs/source/pages/user_guide/logging.rst

@@ -0,0 +1,150 @@
+Communication Logging
+=====================
+UDS communication logging utilities are provided in :mod:`uds.transport_interface.logger` module.
+
+
+TransportLogger
+---------------
+:class:`~uds.transport_interface.logger.TransportLogger` is the recommended implementation
+of the communication logging feature.
+The logger works by wrapping transport interface methods responsible for sending and receiving
+:ref:`UDS Messages <knowledge-base-diagnostic-message>` and :ref:`Packets <knowledge-base-packet>`.
+
+This logger integrates with the built-in
+`logging <https://docs.python.org/3/library/logging.html>`_ module.
+
+Thanks to the modular package design, custom logging implementations can easily be inserted
+between existing communication layers.
+
+.. seealso:: `Python logging documentation <https://docs.python.org/3/library/logging.html>`_
+
+Attributes:
+
+- :attr:`~uds.transport_interface.logger.TransportLogger.logger`
+- :attr:`~uds.transport_interface.logger.TransportLogger.message_logging_level`
+- :attr:`~uds.transport_interface.logger.TransportLogger.packet_logging_level`
+- :attr:`~uds.transport_interface.logger.TransportLogger.log_sending`
+- :attr:`~uds.transport_interface.logger.TransportLogger.log_receiving`
+- :attr:`~uds.transport_interface.logger.TransportLogger.message_log_format`
+- :attr:`~uds.transport_interface.logger.TransportLogger.packet_log_format`
+
+
+Configuration
+`````````````
+
+Upon :class:`~uds.transport_interface.logger.TransportLogger` object creation, the user can configure how the logger
+behaves during the communication.
+
+**Example code:**
+
+.. code-block::  python
+
+  import logging
+  from uds.transport_interface import TransportLogger
+
+  # create example Transport Logger
+  transport_logger = TransportLogger(
+      logger_name="UDS",
+      message_logging_level=logging.INFO,
+      packet_logging_level=logging.DEBUG,
+      log_sending=True,
+      log_receiving=True,
+      message_log_format="{record.direction.name} {record.addressing_type.name} {record.payload}",
+      packet_log_format="{record.direction.name} {record}")
+
+  transport_logger.log_sending = False  # do not log outgoing communication
+  transport_logger.packet_logging_level = None  # do not log packets
+
+
+Activation
+``````````
+There are two ways to activate the logger:
+
+- `Decorating Transport Interface class`_
+- `Decorating Transport Interface instance`_
+
+
+Decorating Transport Interface class
+''''''''''''''''''''''''''''''''''''
+Transport Logger activation is possible by decorating Transport Interface class.
+This is the recommended (and most Pythonic) approach when implementing custom transport interfaces.
+
+**Example code:**
+
+.. code-block::  python
+
+  from uds.transport_interface import AbstractTransportInterface, TransportLogger
+
+  # let's assume that we have `transport_logger` already configured
+  transport_logger: TransportLogger
+
+  @transport_logger
+  class MyTransportInterface(AbstractTransportInterface):
+      ...  # TODO: custom implementation
+
+
+It is also possible to decorate existing Transport Interfaces.
+
+**Example code:**
+
+.. code-block::  python
+
+  from uds.transport_interface import TransportLogger
+  from uds.can import PyCanTransportInterface
+
+  # let's assume that we have `transport_logger` already configured
+  transport_logger: TransportLogger
+
+  PyCanTransportInterfaceWithLogging = transport_logger(PyCanTransportInterface)
+
+.. seealso:: :ref:`An example script <example-transport-logger-class>`
+
+
+Decorating Transport Interface instance
+'''''''''''''''''''''''''''''''''''''''
+Another option is to decorate an already existing transport interface instance.
+
+**Example code:**
+
+.. code-block::  python
+
+  from uds.transport_interface import AbstractTransportInterface, TransportLogger
+
+  # let's assume that we have `transport_interface` already configured
+  transport_interface: AbstractTransportInterface
+
+  # let's assume that we have `transport_logger` already configured
+  transport_logger: TransportLogger
+
+  # add logging to the transport_interface
+  transport_interface_with_logger = transport_logger(transport_interface)
+
+.. seealso:: :ref:`An example script <example-transport-logger-instance>`
+
+
+Customization
+`````````````
+The easiest way to create your own transport logger is to inherit after
+:class:`~uds.transport_interface.logger.TransportLogger` class and add your own features.
+This is also the easiest way to define more advanced or custom logging messages.
+
+**Example code:**
+
+.. code-block::  python
+
+  from uds.transport_interface import TransportLogger
+  from uds.addressing import TransmissionDirection
+  from uds.message import UdsMessageRecord
+  from uds.utilities import bytes_to_hex
+
+  class MyTransportLogger(TransportLogger):
+
+      def log_message(self, record: UdsMessageRecord) -> None:
+          """Log a message after receiving/transmitting UDS Message."""
+          if self.message_logging_level is not None:
+              if record.direction == TransmissionDirection.TRANSMITTED:
+                  message = f"Transmitted message with payload: {bytes_to_hex(record.payload)}"
+              else:
+                  message = f"Received message with payload: {bytes_to_hex(record.payload)}"
+              self.logger.log(level=self.message_logging_level,
+                              msg=message)

examples/debugging/Kvaser/python_can_timing_issue.py

@@ -15,19 +15,19 @@
     message = Message(data=[0x12, 0x34, 0x56, 0x78, 0x9A, 0xBC, 0xDE, 0xF0], arbitration_id=0x100)
 
     for _ in range(100):
-        timestamp_before_send = time()
+        time_before_send = time()
         Timer(interval=0.1, function=kvaser_interface_1.send, args=(message,)).start()
 
         sent_message = buffered_reader.get_message(timeout=1)
-        timestamp_after_send = time()
+        time_after_send = time()
 
         print(f"-----------------------------------------------\n"
               f"Result:\n"
-              f"Timestamp before send: {timestamp_before_send}\n"
+              f"Timestamp before send: {time_before_send}\n"
               f"Message timestamp: {sent_message.timestamp}\n"
-              f"Current timestamp: {timestamp_after_send}\n"
-              f"Timestamp before send <= Message timestamp <= Current timestamp: {timestamp_before_send <= sent_message.timestamp <= timestamp_after_send} (expected `True`)\n"
-              f"Current timestamp - Message timestamp: {timestamp_after_send - sent_message.timestamp:06f} (excepted >= 0)")
+              f"Current timestamp: {time_after_send}\n"
+              f"Timestamp before send <= Message timestamp <= Current timestamp: {time_before_send <= sent_message.timestamp <= time_after_send} (expected `True`)\n"
+              f"Current timestamp - Message timestamp: {time_after_send - sent_message.timestamp:06f} (excepted >= 0)")
 
     kvaser_interface_1.shutdown()
     kvaser_interface_2.shutdown()

examples/debugging/Kvaser/python_can_timing_issue_2.py

@@ -14,19 +14,19 @@
     message = Message(data=[0x12, 0x34, 0x56, 0x78, 0x9A, 0xBC, 0xDE, 0xF0], arbitration_id=0x100)
 
     for _ in range(100):
-        timestamp_before_send = time()
+        time_before_send = time()
         kvaser_interface_1.send(message)
         sent_message = buffered_reader.get_message(timeout=1)
-        timestamp_after_send = time()
+        time_after_send = time()
 
         print(f"-----------------------------------------------\n"
               f"Result:\n"
-              f"Timestamp before send: {timestamp_before_send}\n"
+              f"Timestamp before send: {time_before_send}\n"
               f"Message timestamp: {sent_message.timestamp}\n"
-              f"Current timestamp: {timestamp_after_send}\n"
-              f"Message timestamp - Timestamp before send: {sent_message.timestamp - timestamp_before_send:06f} (expected > 0)\n"
-              f"Current timestamp - Message timestamp: {timestamp_after_send - sent_message.timestamp:06f} (expected > 0)\n"
-              f"Timestamp before send <= Message timestamp <= Current timestamp: {timestamp_before_send <= sent_message.timestamp <= timestamp_after_send} (expected `True`)")
+              f"Current timestamp: {time_after_send}\n"
+              f"Message timestamp - Timestamp before send: {sent_message.timestamp - time_before_send:06f} (expected > 0)\n"
+              f"Current timestamp - Message timestamp: {time_after_send - sent_message.timestamp:06f} (expected > 0)\n"
+              f"Timestamp before send <= Message timestamp <= Current timestamp: {time_before_send <= sent_message.timestamp <= time_after_send} (expected `True`)")
 
     kvaser_interface_1.shutdown()
     kvaser_interface_2.shutdown()

examples/transport/class_logger.py

@@ -0,0 +1,32 @@
+"""Configure logger for class of Transport Interface."""
+
+import logging
+
+from uds.transport_interface import AbstractTransportInterface, TransportLogger
+
+# configure your logging
+# https://docs.python.org/3/library/logging.html
+logger = logging.getLogger("UDS")  # example logger name
+logger.setLevel(logging.DEBUG)  # example logging level
+# optionally configure logging to file
+file_handler = logging.FileHandler(filename="uds.log", encoding="utf-8")
+file_handler.setLevel(logging.DEBUG)
+logger.addHandler(file_handler)
+# optionally configure logging to python console
+stream_handler = logging.StreamHandler()
+stream_handler.setLevel(logging.INFO)
+logger.addHandler(stream_handler)
+
+# configure your logger
+# https://uds.readthedocs.io/en/stable/pages/user_guide/logging.html#configuration
+transport_logger = TransportLogger(logger_name="UDS",  # the same name as previously configured logger
+                                   message_logging_level=logging.INFO,
+                                   packet_logging_level=logging.DEBUG,
+                                   log_sending=True,
+                                   log_receiving=True)
+
+# activate your logger
+# https://uds.readthedocs.io/en/stable/pages/user_guide/logging.html#decorating-transport-interface-class
+@transport_logger
+class CustomTransportInterface(AbstractTransportInterface):
+    ...  # TODO: implement your interface class

examples/transport/instance_logger.py

@@ -0,0 +1,36 @@
+"""Configure logger for instance of Transport Interface."""
+
+import logging
+
+from uds.transport_interface import AbstractTransportInterface, TransportLogger
+
+# configure your own Transport Interface
+# https://uds.readthedocs.io/en/stable/pages/user_guide/quickstart.html#create-transport-interface
+transport_interface: AbstractTransportInterface = ...  # TODO: provide your implementation here
+
+# configure your logging
+# https://docs.python.org/3/library/logging.html
+logger = logging.getLogger("UDS")  # example logger name
+logger.setLevel(logging.DEBUG)  # example logging level
+# optionally configure logging to file
+file_handler = logging.FileHandler(filename="uds.log", encoding="utf-8")
+file_handler.setLevel(logging.DEBUG)
+logger.addHandler(file_handler)
+# optionally configure logging to python console
+stream_handler = logging.StreamHandler()
+stream_handler.setLevel(logging.INFO)
+logger.addHandler(stream_handler)
+
+# configure your logger
+# https://uds.readthedocs.io/en/stable/pages/user_guide/logging.html#configuration
+transport_logger = TransportLogger(logger_name="UDS",  # the same name as previously configured logger
+                                   message_logging_level=logging.INFO,
+                                   packet_logging_level=logging.DEBUG,
+                                   log_sending=True,
+                                   log_receiving=True)
+
+# activate your logger
+# https://uds.readthedocs.io/en/stable/pages/user_guide/logging.html#decorating-transport-interface-instance
+transport_interface_with_logger = transport_logger(transport_interface)
+
+# TODO: use `transport_interface_with_logger` the same way as `transport_interface`

tests/software_tests/can/transport_interface/test_python_can.py

@@ -545,7 +545,7 @@ def test_teardown_sync_listening__notifier(self):
         assert PyCanTransportInterface._PyCanTransportInterface__teardown_sync_listening(
             self.mock_can_transport_interface) is None
         assert self.mock_can_transport_interface.notifier is None
-        mock_notifier.stop.assert_called_once_with(self.mock_can_transport_interface._MIN_NOTIFIER_TIMEOUT)
+        mock_notifier.stop.assert_called_once_with()
         self.mock_warn.assert_called_once()
 
     def test_teardown_sync_listening__notifier_with_suppressed_warning(self):
@@ -554,7 +554,7 @@ def test_teardown_sync_listening__notifier_with_suppressed_warning(self):
         assert PyCanTransportInterface._PyCanTransportInterface__teardown_sync_listening(
             self.mock_can_transport_interface, suppress_warning=True) is None
         assert self.mock_can_transport_interface.notifier is None
-        mock_notifier.stop.assert_called_once_with(self.mock_can_transport_interface._MIN_NOTIFIER_TIMEOUT)
+        mock_notifier.stop.assert_called_once_with()
         self.mock_warn.assert_not_called()
 
     # __teardown_async_listening
@@ -572,7 +572,7 @@ def test_teardown_async_listening__notifier(self):
         assert PyCanTransportInterface._PyCanTransportInterface__teardown_async_listening(
             self.mock_can_transport_interface) is None
         assert self.mock_can_transport_interface.async_notifier is None
-        mock_notifier.stop.assert_called_once_with(self.mock_can_transport_interface._MIN_NOTIFIER_TIMEOUT)
+        mock_notifier.stop.assert_called_once_with()
         self.mock_warn.assert_called_once()
 
     def test_teardown_async_listening__notifier_with_suppressed_warning(self):
@@ -581,7 +581,7 @@ def test_teardown_async_listening__notifier_with_suppressed_warning(self):
         assert PyCanTransportInterface._PyCanTransportInterface__teardown_async_listening(
             self.mock_can_transport_interface, suppress_warning=True) is None
         assert self.mock_can_transport_interface.async_notifier is None
-        mock_notifier.stop.assert_called_once_with(self.mock_can_transport_interface._MIN_NOTIFIER_TIMEOUT)
+        mock_notifier.stop.assert_called_once_with()
         self.mock_warn.assert_not_called()
 
     # __validate_timeout