Ruff/pyright adapters logging devices.init

Author: Jack Doughty

lewis/core/adapters.py

@@ -24,6 +24,7 @@
 """
 
 import inspect
+import logging
 import threading
 from collections import namedtuple
 from types import TracebackType
@@ -237,6 +238,7 @@ def __init__(self, *args: Adapter) -> None:
         self._threads = {}
         self._running = {}
         self._lock = threading.Lock()
+        self.log: logging.Logger
 
         for adapter in args:
             self.add_adapter(adapter)
@@ -301,7 +303,7 @@ def connect(self, *args: str) -> None:
 
     def _start_server(self, adapter: Adapter) -> None:
         if adapter.protocol not in self._threads:
-            self.log.info("Connecting device interface for protocol '%s'", adapter.protocol)  # type: ignore not sure?
+            self.log.info("Connecting device interface for protocol '%s'", adapter.protocol)
 
             adapter_thread = threading.Thread(target=self._adapter_loop, args=(adapter, 0.01))
             adapter_thread.daemon = True
@@ -322,7 +324,7 @@ def _adapter_loop(self, adapter: Adapter, dt: float) -> None:
 
         self._running[adapter.protocol].set()
 
-        self.log.debug("Starting adapter loop for protocol %s.", adapter.protocol)  # type: ignore not sure?
+        self.log.debug("Starting adapter loop for protocol %s.", adapter.protocol)
         while self._running[adapter.protocol].is_set():
             adapter.handle(dt)
 
@@ -340,7 +342,7 @@ def disconnect(self, *args: str) -> None:
 
     def _stop_server(self, adapter: Adapter) -> None:
         if adapter.protocol in self._threads:
-            self.log.info("Disconnecting device interface for protocol '%s'", adapter.protocol)  # type: ignore not sure?
+            self.log.info("Disconnecting device interface for protocol '%s'", adapter.protocol)
 
             self._running[adapter.protocol].clear()
             self._threads[adapter.protocol].join()

lewis/core/logging.py

@@ -33,11 +33,26 @@
 """
 
 import logging
+from typing import Callable, overload, ParamSpec, TypeVar, Type, Protocol
+
+
+class HasLog(Protocol):
+    log: logging.Logger
+
+
+P = ParamSpec("P")
+T = TypeVar("T")
 
 root_logger_name = "lewis"
 default_log_format = "%(asctime)s %(levelname)s %(name)s: %(message)s"
 
 
+@overload
+def has_log(target: Type[T]) -> Type[T]: ...
+@overload
+def has_log(target: Callable[P, T]) -> Callable[P, T]: ...
+
+
 def has_log(target):
     """
     This is a decorator to add logging functionality to a class or function.
@@ -95,15 +110,15 @@ def foo(bar):
     """
     logger_name = target.__name__
 
-    def get_logger_name(context=None):
+    def get_logger_name(context: object = None) -> str:
         log_names = [root_logger_name, logger_name]
 
         if context is not None:
             log_names.insert(1, context if isinstance(context, str) else context.__class__.__name__)
 
         return ".".join(log_names)
 
-    def _set_logging_context(obj, context):
+    def _set_logging_context(obj: HasLog, context: object) -> None:
         """
         Changes the logger name of this class using the supplied context
         according to the rules described in the documentation of :func:`has_log`. To

lewis/devices/__init__.py

@@ -23,6 +23,7 @@
 using a state machine.
 """
 
+import logging
 from typing import Callable
 
 from lewis.core.devices import DeviceBase
@@ -103,7 +104,8 @@ def __init__(
     ) -> None:
         super(StateMachineDevice, self).__init__()
 
-        self.log.info("Creating device, setting up state machine")  # type: ignore not sure?
+        self.log: logging.Logger
+        self.log.info("Creating device, setting up state machine")
 
         self._initialize_data()
         self._override_data(override_initial_data)
@@ -193,7 +195,7 @@ def _override_data(self, overrides: dict[str, float] | None) -> None:
         """
         if overrides is not None:
             for name, val in overrides.items():
-                self.log.debug("Trying to override initial data (%s=%s)", name, val)  # type: ignore not sure?
+                self.log.debug("Trying to override initial data (%s=%s)", name, val)
                 if name not in dir(self):
                     raise AttributeError(
                         "Can not override non-existing attribute" "'{}' of class '{}'.".format(