DriverParamsMixin

qtoggleserver/core/ports.py

@@ -840,7 +840,7 @@ async def to_json(self) -> GenericJSONDict:
         attrs: GenericJSONDict = await self.get_attrs()
 
         if self._enabled:
-            attrs["value"] = self._last_read_value[0] if self._last_read_value else None
+            attrs["value"] = self.get_last_read_value()
             attrs["pending_value"] = self.get_pending_value()
         else:
             attrs["value"] = None

qtoggleserver/peripherals/__init__.py

@@ -31,29 +31,19 @@ def get(peripheral_id: str) -> Peripheral | None:
 
 async def add(peripheral_args: dict[str, Any], static: bool = False) -> Peripheral:
     peripheral_args = peripheral_args.copy()
-    class_path = peripheral_args.pop("driver")
-    peripheral_args.pop("static", None)
-    peripheral_args.pop("enabled", None)  # computed at runtime
-    peripheral_args.pop("online", None)  # computed at runtime
-
-    params = peripheral_args.pop("params", None)
-    if params is None:
-        # Backward compatibility with older persisted payloads where params were flattened.
-        params = {k: v for k, v in peripheral_args.items() if k not in {"name", "id", "display_name", "force_enabled"}}
-    elif not isinstance(params, dict):
-        raise TypeError("params must be a dictionary")
+    class_path = peripheral_args["driver"]
+
+    # Merge params into peripheral args
+    params = peripheral_args.pop("params", {})
+    peripheral_args.update(params)
 
     logger.debug('creating peripheral with driver "%s"', class_path)
     try:
         peripheral_class = dynload_utils.load_attr(class_path)
     except Exception:
         raise NoSuchDriver(class_path)
 
-    # Supply actual peripheral params both via constructor kwargs (to pass to concrete class) and as dedicated `params`
-    # arg (to be able to retrieve params using `get_params()`).
-    peripheral_args.update(params)
-
-    p: Peripheral = peripheral_class(params=params, driver=class_path, static=static, **peripheral_args)
+    p: Peripheral = peripheral_class(static=static, **peripheral_args)
     if p.get_id() in _registered_peripherals:
         raise DuplicatePeripheral(f"Peripheral {p.get_id()} already exists")
 

qtoggleserver/peripherals/peripheral.py

@@ -13,14 +13,15 @@
 from qtoggleserver.utils import asyncio as asyncio_utils
 from qtoggleserver.utils import logging as logging_utils
 from qtoggleserver.utils import runner as runner_utils
+from qtoggleserver.utils.driver_params import DriverParamsMixin
 
 from .exceptions import NotOurPort
 
 
 logger = logging.getLogger(__package__)
 
 
-class Peripheral(logging_utils.LoggableMixin, metaclass=abc.ABCMeta):
+class Peripheral(DriverParamsMixin, logging_utils.LoggableMixin, metaclass=abc.ABCMeta):
     RUNNER_CLASS = runner_utils.ThreadedRunner
     RUNNER_QUEUE_SIZE = 64
 
@@ -29,20 +30,16 @@ class Peripheral(logging_utils.LoggableMixin, metaclass=abc.ABCMeta):
     def __init__(
         self,
         *,
-        params: dict[str, Any],
-        driver: str | None = None,
         name: str | None = None,
         display_name: str = "",
         force_enabled: bool | None = None,
         static: bool = False,
         **kwargs,
     ) -> None:
-        self._params: dict[str, Any] = params
-        self._driver: str = driver or f"{self.__class__.__module__}.{self.__class__.__name__}"
         self._name: str | None = name
         self._id: str = name or ""  # name will always be used as id, if supplied
         if not self._id:
-            sorted_params = self._sorted_tuples_dict(params)
+            sorted_params = self._sorted_tuples_dict(self.get_params())
             auto_id_to_hash = f"{self.__class__.__module__}.{self.__class__.__name__}:{name}:{sorted_params}"
             self._id = f"peripheral_{hashlib.sha256(auto_id_to_hash.encode()).hexdigest()[:8]}"
         self._display_name: str = display_name or ""
@@ -73,18 +70,12 @@ def get_id(self) -> str:
     def get_name(self) -> str | None:
         return self._name
 
-    def get_driver(self) -> str:
-        return self._driver
-
     def get_display_name(self) -> str:
         return self._display_name
 
     def set_display_name(self, display_name: str) -> None:
         self._display_name = display_name
 
-    def get_params(self) -> dict[str, Any]:
-        return self._params
-
     def is_static(self) -> bool:
         return self._static
 

qtoggleserver/utils/driver_params.py

@@ -0,0 +1,91 @@
+import inspect
+
+from typing import Any
+
+
+class DriverParamsMixin:
+    """Mixin that captures __init__ parameters in _params for introspection.
+
+    When a class hierarchy inherits from this mixin, all keyword arguments that
+    match parameter names in the class hierarchy's __init__ signatures are
+    captured in the _params dict. This enables introspection of how driver
+    instances were configured.
+
+    The _driver attribute is automatically set to either:
+      - The value of the 'driver' kwarg if provided, OR
+      - The fully-qualified class name (module.ClassName) as default
+
+    Behavior:
+      - Walks the MRO to collect all __init__ parameter names from classes
+        appearing before DriverParamsMixin in the hierarchy
+      - Captures kwargs matching those parameter names in _params dict
+      - Only keyword arguments are captured (positional args are not tracked)
+      - Kwargs not matching any parameter name in the hierarchy are silently dropped
+
+    Usage:
+        class Base(DriverParamsMixin):
+            def __init__(self, b1: str, b2: int, **kwargs: Any) -> None:
+                super().__init__(**kwargs)
+                self.b1 = b1
+                self.b2 = b2
+
+        class Child(Base):
+            def __init__(self, b1: str, b2: int, c1: str, c2: float, **kwargs: Any) -> None:
+                super().__init__(b1=b1, b2=b2, **kwargs)
+                self.c1 = c1
+                self.c2 = c2
+
+        # All kwargs matching hierarchy parameter names are captured
+        child = Child(b1="v1", b2=42, c1="cv1", c2=3.14)
+        assert child.get_params() == {"b1": "v1", "b2": 42, "c1": "cv1", "c2": 3.14}
+
+        # Extra kwargs not in any signature are dropped
+        child = Child(b1="v1", b2=42, c1="cv1", c2=3.14, unknown="dropped")
+        assert child.get_params() == {"b1": "v1", "b2": 42, "c1": "cv1", "c2": 3.14}
+
+        # Custom driver name
+        child = Child(b1="v1", b2=42, c1="cv1", c2=3.14, driver="custom.Driver")
+        assert child.get_driver() == "custom.Driver"
+    """
+
+    _params: dict[str, Any]
+    _driver: str
+
+    def __new__(cls: type[DriverParamsMixin], *args: Any, **kwargs: Any) -> DriverParamsMixin:
+        instance = super().__new__(cls)
+
+        # Find the direct parent class that is not DriverParamsMixin
+        parents = []
+        for i, base in enumerate(cls.__mro__):
+            if i < len(cls.__mro__) - 1 and cls.__mro__[i + 1] is DriverParamsMixin:
+                break
+            parents.append(cls.__mro__[i])
+
+        # Only capture driver params if we have a parent (i.e., cls is not the base)
+        # Get parameters defined by the *direct parent* class
+        driver_param_names: set[str] = set()
+        for parent in parents:
+            try:
+                sig = inspect.signature(parent.__init__)
+                driver_param_names.update({p for p in sig.parameters.keys() if p != "self"})
+            except ValueError, TypeError:
+                pass
+
+        # Driver params = any kwarg NOT in direct parent
+        driver_params = {k: v for k, v in kwargs.items() if k in driver_param_names}
+
+        # Store driver params on the instance before __init__ runs
+        instance._params = driver_params
+        instance._driver = kwargs.get("driver") or f"{cls.__module__}.{cls.__name__}"
+
+        return instance
+
+    def __init__(self, **kwargs: Any) -> None:
+        """Consume kwargs to prevent them from reaching object.__init__()."""
+        pass
+
+    def get_driver(self) -> str:
+        return self._driver
+
+    def get_params(self) -> dict[str, Any]:
+        return self._params