diff options
| -rw-r--r-- | kvmd/apps/kvmd/ugpio.py | 16 | ||||
| -rw-r--r-- | kvmd/plugins/ugpio/__init__.py | 16 | ||||
| -rw-r--r-- | kvmd/plugins/ugpio/gpio.py | 23 |
3 files changed, 39 insertions, 16 deletions
diff --git a/kvmd/apps/kvmd/ugpio.py b/kvmd/apps/kvmd/ugpio.py index f25c511c..8e7ae90a 100644 --- a/kvmd/apps/kvmd/ugpio.py +++ b/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) diff --git a/kvmd/plugins/ugpio/__init__.py b/kvmd/plugins/ugpio/__init__.py index 9dbc0166..0cc8662a 100644 --- a/kvmd/plugins/ugpio/__init__.py +++ b/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: - raise NotImplementedError + def __init__( # pylint: disable=super-init-not-called + 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: diff --git a/kvmd/plugins/ugpio/gpio.py b/kvmd/plugins/ugpio/gpio.py index 526397e1..96c3eee7 100644 --- a/kvmd/plugins/ugpio/gpio.py +++ b/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__ |