feat: Added individual VISA command logging for each VISA device.

CHANGELOG.md

@@ -26,6 +26,12 @@ Things to be included in the next release go here.
 - Added the `get_errors()` method to the `Device` class to enable easy access to the current error code and messages on any device.
 - Added more details to the Architectural Overview page of the documentation as well as highlighting to the device driver diagram on the page.
 - Added regex matching to the `verify_values()` helper function to allow for more flexible value verification.
+- A main logfile is now created by default (can be disabled if desired) that contains all the logging output of the entire tm_devices package during execution.
+    - Use the `configure_logging()` function to set the logging levels for stdout and file logging.
+    - The default settings will log all messages to the log file and maintain the current printout functionality on stdout.
+- A logfile is now created that contains each command sent to a VISA device.
+    - This file is located next to the main log file and will start with the same name, but have the unique address of the device appended.
+    - This file will only be created if file logging is enabled for the package (which is the default behavior).
 
 ### Changed
 

src/tm_devices/driver_mixins/device_control/pi_control.py

@@ -1,13 +1,16 @@
 """Base Programmable Interface (PI) control class module."""
 
+import contextlib
 import logging
+import logging.handlers
 import os
 import socket
 import time
 import warnings
 
 from abc import ABC
 from contextlib import contextmanager
+from pathlib import Path
 from typing import final, Generator, List, Optional, Sequence, Tuple, Union
 
 import pyvisa as visa
@@ -35,6 +38,7 @@
     verify_values,
 )
 from tm_devices.helpers import ReadOnlyCachedProperty as cached_property  # noqa: N813
+from tm_devices.helpers.constants_and_dataclasses import PACKAGE_NAME
 
 _logger: logging.Logger = logging.getLogger(__name__)
 
@@ -339,14 +343,13 @@ def query(  # pylint: disable=arguments-differ
             Error: An error occurred while sending the command.
             SystemError: An empty string was returned from the device.
         """
-        # TODO: logging: Log just the query to a device specific file
         _logger.log(
             logging.INFO if self._verbose and verbose else logging.DEBUG,
             "(%s) Query >>  %r",
             self._name_and_alias,
             query,
         )
-
+        self._command_logger.debug(query)
         try:
             response = self._visa_resource.query(query).strip()
             if remove_quotes:
@@ -386,14 +389,13 @@ def query_binary(self, query: str, verbose: bool = True) -> Sequence[float]:
             Error: An error occurred while sending the command.
             SystemError: An empty string was returned from the device.
         """
-        # TODO: logging: Log just the query to a device specific file
         _logger.log(
             logging.INFO if self._verbose and verbose else logging.DEBUG,
             "(%s) Query Binary Values >>  %r",
             self._name_and_alias,
             query,
         )
-
+        self._command_logger.debug(query)
         try:
             response = self._visa_resource.query_binary_values(query)  # pyright: ignore[reportUnknownMemberType]
         except (visa.VisaIOError, socket.error) as error:
@@ -516,14 +518,13 @@ def query_raw_binary(self, query: str, verbose: bool = True) -> bytes:
             Error: An error occurred while sending the command.
             SystemError: An empty string was returned from the device.
         """
-        # TODO: logging: Log just the query to a device specific file
         _logger.log(
             logging.INFO if self._verbose and verbose else logging.DEBUG,
             "(%s) Query Raw Binary >>  %r",
             self._name_and_alias,
             query,
         )
-
+        self._command_logger.debug(query)
         try:
             self._visa_resource.write(query)
             response = self.read_raw()
@@ -841,7 +842,6 @@ def write(self, command: str, opc: bool = False, verbose: bool = True) -> None:
             Error: An error occurred while sending the command.
             SystemError: ``*OPC?`` did not return "1" after sending the command.
         """
-        # TODO: logging: Log just the write to a device specific file
         if "\n" in command:
             _logger.log(
                 logging.INFO if self._verbose and verbose else logging.DEBUG,
@@ -856,7 +856,7 @@ def write(self, command: str, opc: bool = False, verbose: bool = True) -> None:
                 self._name_and_alias,
                 command,
             )
-
+        self._command_logger.debug(command)
         try:
             self._visa_resource.write(command)
         except (visa.VisaIOError, socket.error) as error:
@@ -879,14 +879,13 @@ def write_raw(self, command: bytes, verbose: bool = True) -> None:
         Raises:
             Error: An error occurred while sending the command.
         """
-        # TODO: logging: Log just the write to a device specific file
         _logger.log(
             logging.INFO if self._verbose and verbose else logging.DEBUG,
             "(%s) Write Raw >>  %r",
             self._name_and_alias,
             command,
         )
-
+        self._command_logger.debug(command)
         try:
             self._visa_resource.write_raw(command)
         except (visa.VisaIOError, socket.error) as error:
@@ -956,6 +955,38 @@ def _close(self) -> None:
         self._visa_resource = None  # pyright: ignore[reportAttributeAccessIssue]
         self._is_open = False
 
+    @cached_property
+    def _command_logger(self) -> logging.Logger:
+        """Create a logger that will be used to log commands sent via write/query-like methods."""
+        # Create the logger
+        command_logger = logging.getLogger(f"{self._config_entry.address}-visa_logger")
+        command_logger.setLevel(logging.DEBUG)
+        command_logger.propagate = False
+        command_logger.addHandler(logging.NullHandler())
+        with contextlib.suppress(IndexError, StopIteration):
+            # Get the top-level log filepath
+            main_log_file = Path(
+                next(
+                    x
+                    for x in logging.getLogger(PACKAGE_NAME).handlers
+                    if isinstance(x, logging.FileHandler)
+                ).baseFilename
+            )
+            # Create the handler with the filename based on the main log file
+            command_log_handler = logging.FileHandler(
+                main_log_file.as_posix().replace(
+                    main_log_file.suffix, f"_visa_commands_{self._config_entry.address}.log"
+                ),
+                mode="w",
+                encoding="utf-8",
+            )
+            # Create the formatter
+            command_log_formatter = logging.Formatter("%(message)s")
+            command_log_handler.setFormatter(command_log_formatter)
+            command_log_handler.setLevel(logging.DEBUG)
+            command_logger.addHandler(command_log_handler)
+        return command_logger
+
     def _open(self) -> bool:
         """Open necessary resources and components and return a boolean indicating success."""
         opened = True

src/tm_devices/helpers/logging.py

@@ -78,7 +78,6 @@ def configure_logging(
         log_file_name = f"{PACKAGE_NAME}_{time.strftime('%m-%d-%Y_%H-%M-%S', time.localtime())}.log"
     log_filepath = log_file_directory / log_file_name
 
-    # TODO: logging: look into rotating log files
     if file_logging_level != LoggingLevels.NONE:
         # Set up logger for tm_devices
         log_filepath.parent.mkdir(parents=True, exist_ok=True)

tests/conftest.py

@@ -36,7 +36,9 @@ def emit(self, record: logging.LogRecord) -> None:
 
 _logger = configure_logging(
     console_logging_level=LoggingLevels.NONE,
-    file_logging_level=LoggingLevels.NONE,
+    file_logging_level=LoggingLevels.DEBUG,
+    log_file_name="unit_test.log",
+    log_file_directory=Path(__file__).parent / "logs",
 )
 _unit_test_console_handler = _DynamicStreamHandler(stream=sys.stdout)
 _unit_test_console_handler.setLevel(logging.DEBUG)

tests/test_all_device_drivers.py

@@ -4,6 +4,7 @@
 import contextlib
 
 from collections import Counter
+from pathlib import Path
 from typing import Generator, List, Optional
 
 import pytest
@@ -209,3 +210,22 @@ def test_all_device_drivers() -> None:
     print(  # noqa: T201
         f"{len(drivers_with_auto_generated_commands)} device drivers have auto-generated commands"
     )
+
+
+@pytest.mark.order(3)
+@pytest.mark.depends(on=["test_device_driver"])
+def test_visa_device_command_logging() -> None:
+    """Test the VISA command log file contents."""
+    generated_file = Path(__file__).parent / "logs/unit_test_visa_commands_SMU2410-HOSTNAME.log"
+    assert generated_file.exists(), f"File not found: {generated_file}"
+    assert (
+        generated_file.read_text()
+        == """*CLS
+*RST
+*OPC?
+*ESR?
+SYSTEM:ERROR?
+*ESR?
+SYSTEM:ERROR?
+"""
+    )

tests/test_logging.py

@@ -0,0 +1,39 @@
+"""Tests for the logging functionality."""
+
+import contextlib
+import logging
+
+from typing import Generator, TYPE_CHECKING
+
+import pytest
+
+from tm_devices import DeviceManager, PACKAGE_NAME
+
+if TYPE_CHECKING:
+    from tm_devices.drivers import MSO2
+
+
+@pytest.fixture(name="remove_log_file_handler")
+def _remove_log_file_handler(device_manager: DeviceManager) -> Generator[None, None, None]:  # pyright: ignore[reportUnusedFunction]
+    """Remove the file handler from the logger."""
+    device_manager.remove_all_devices()
+    logger = logging.getLogger(PACKAGE_NAME)
+    file_handler = None
+    with contextlib.suppress(StopIteration):
+        file_handler = next(
+            handler for handler in logger.handlers if isinstance(handler, logging.FileHandler)
+        )
+        logger.removeHandler(file_handler)
+    yield
+    if file_handler is not None:
+        logger.addHandler(file_handler)
+    device_manager.remove_all_devices()
+
+
+def test_visa_command_logging_edge_cases(
+    device_manager: DeviceManager,
+    remove_log_file_handler: None,  # noqa: ARG001
+) -> None:
+    """Test VISA command logging edge cases."""
+    scope: MSO2 = device_manager.add_scope("MSO22-HOSTNAME")
+    assert scope.model == "MSO22"