refactoring

kvmd/apps/kvmd/ugpio.py

@@ -95,7 +95,7 @@ class _GpioInput:
         }
 
     def __str__(self) -> str:
-        return f"Input({self.__channel}, driver={self.__driver.get_instance_name()}, pin={self.__pin})"
+        return f"Input({self.__channel}, driver={self.__driver}, pin={self.__pin})"
 
     __repr__ = __str__
 
@@ -136,7 +136,7 @@ class _GpioOutput:  # pylint: disable=too-many-instance-attributes
                 "max_delay": (self.__max_pulse_delay if self.__pulse_delay else 0),
             },
             "hw": {
-                "driver": self.__driver.get_instance_name(),
+                "driver": str(self.__driver),
                 "pin": self.__pin,
             },
         }
@@ -205,7 +205,7 @@ class _GpioOutput:  # pylint: disable=too-many-instance-attributes
         self.__driver.write(self.__pin, (state ^ self.__inverted))
 
     def __str__(self) -> str:
-        return f"Output({self.__channel}, driver={self.__driver.get_instance_name()}, pin={self.__pin})"
+        return f"Output({self.__channel}, driver={self.__driver}, pin={self.__pin})"
 
     __repr__ = __str__
 
@@ -218,7 +218,11 @@ class UserGpio:
         self.__state_notifier = aiotools.AioNotifier()
 
         self.__drivers = {
-            driver: get_ugpio_driver_class(drv_config.type)(**drv_config._unpack(ignore=["type"]))
+            driver: get_ugpio_driver_class(drv_config.type)(
+                instance_name=driver,
+                notifier=self.__state_notifier,
+                **drv_config._unpack(ignore=["instance_name", "notifier", "type"]),
+            )
             for (driver, drv_config) in config.drivers.items()
         }
 
@@ -261,7 +265,7 @@ class UserGpio:
     def sysprep(self) -> None:
         get_logger().info("Preparing User-GPIO drivers ...")
         for (_, driver) in sorted(self.__drivers.items(), key=operator.itemgetter(0)):
-            driver.prepare(self.__state_notifier)
+            driver.prepare()
 
     async def systask(self) -> None:
         get_logger(0).info("Running User-GPIO drivers ...")
@@ -277,7 +281,7 @@ class UserGpio:
             try:
                 driver.cleanup()
             except Exception:
-                get_logger().exception("Can't cleanup driver %r", driver.get_instance_name())
+                get_logger().exception("Can't cleanup driver %s", driver)
 
     async def switch(self, channel: str, state: bool) -> bool:
         gout = self.__outputs.get(channel)

kvmd/plugins/ugpio/__init__.py

@@ -22,6 +22,7 @@
 
 from typing import Type
 from typing import Optional
+from typing import Any
 
 from ...errors import OperationError
 
@@ -42,13 +43,20 @@ class GpioOperationError(OperationError, GpioError):
 
 class GpioDriverOfflineError(GpioOperationError):
     def __init__(self, driver: "BaseUserGpioDriver") -> None:
-        super().__init__(f"GPIO driver {driver.get_instance_name()!r} is offline")
+        super().__init__(f"GPIO driver {driver} is offline")
 
 
 # =====
 class BaseUserGpioDriver(BasePlugin):
-    def get_instance_name(self) -> str:
+    def __init__(  # pylint: disable=super-init-not-called
-        raise NotImplementedError
+        self,
+        instance_name: str,
+        notifier: aiotools.AioNotifier,
+        **_: Any,
+    ) -> None:
+
+        self._instance_name = instance_name
+        self._notifier = notifier
 
     def register_input(self, pin: int) -> None:
         raise NotImplementedError
@@ -56,7 +64,7 @@ class BaseUserGpioDriver(BasePlugin):
     def register_output(self, pin: int, initial: Optional[bool]) -> None:
         raise NotImplementedError
 
-    def prepare(self, notifier: aiotools.AioNotifier) -> None:
+    def prepare(self) -> None:
         raise NotImplementedError
 
     async def run(self) -> None:

kvmd/plugins/ugpio/gpio.py

@@ -36,7 +36,16 @@ from . import BaseUserGpioDriver
 
 # =====
 class Plugin(BaseUserGpioDriver):
-    def __init__(self, state_poll: float) -> None:  # pylint: disable=super-init-not-called
+    def __init__(
+        self,
+        instance_name: str,
+        notifier: aiotools.AioNotifier,
+
+        state_poll: float,
+    ) -> None:
+
+        super().__init__(instance_name, notifier)
+
         self.__state_poll = state_poll
 
         self.__input_pins: Set[int] = set()
@@ -50,16 +59,13 @@ class Plugin(BaseUserGpioDriver):
             "state_poll": Option(0.1, type=valid_float_f01),
         }
 
-    def get_instance_name(self) -> str:
-        return "gpio"
-
     def register_input(self, pin: int) -> None:
         self.__input_pins.add(pin)
 
     def register_output(self, pin: int, initial: Optional[bool]) -> None:
         self.__output_pins[pin] = initial
 
-    def prepare(self, notifier: aiotools.AioNotifier) -> None:
+    def prepare(self) -> None:
         assert self.__reader is None
         self.__reader = gpio.BatchReader(
             pins=set([
@@ -70,7 +76,7 @@ class Plugin(BaseUserGpioDriver):
                 ],
             ]),
             interval=self.__state_poll,
-            notifier=notifier,
+            notifier=self._notifier,
         )
 
     async def run(self) -> None:
@@ -82,3 +88,8 @@ class Plugin(BaseUserGpioDriver):
 
     def write(self, pin: int, state: bool) -> None:
         gpio.write(pin, state)
+
+    def __str__(self) -> str:
+        return f"GPIO({self._instance_name})"
+
+    __repr__ = __str__