refactoring and reuse gpio pulse code

3 files changed, 17 insertions, 33 deletions

diff --git a/kvmd/plugins/atx/gpio.py b/kvmd/plugins/atx/gpio.py
index 70354da4..5ca611ef 100644
--- a/kvmd/plugins/atx/gpio.py
+++ b/kvmd/plugins/atx/gpio.py
@@ -20,8 +20,6 @@
 # ========================================================================== #
 
 
-import asyncio
-
 from typing import Dict
 from typing import AsyncGenerator
 from typing import Optional
@@ -77,8 +75,10 @@ class Plugin(BaseAtx):  # pylint: disable=too-many-instance-attributes
         self.__reader = aiogp.AioPinsReader(
             path="/dev/gpiochip0",
             consumer="kvmd/atx-gpio/leds",
-            pins=[power_led_pin, hdd_led_pin],
-            inverted=[power_led_inverted, hdd_led_inverted],
+            pins={
+                power_led_pin: power_led_inverted,
+                hdd_led_pin: hdd_led_inverted,
+            },
             notifier=self.__notifier,
         )
 
@@ -185,10 +185,5 @@ class Plugin(BaseAtx):  # pylint: disable=too-many-instance-attributes
 
     @aiotools.atomic
     async def __inner_click(self, name: str, line: gpiod.Line, delay: float) -> None:
-        try:
-            line.set_value(1)
-            await asyncio.sleep(delay)
-        finally:
-            line.set_value(0)
-            await asyncio.sleep(1)
+        await aiogp.pulse(line, delay, 1)
         get_logger(0).info("Clicked ATX button %r", name)
diff --git a/kvmd/plugins/hid/serial.py b/kvmd/plugins/hid/serial.py
index ff018826..27218323 100644
--- a/kvmd/plugins/hid/serial.py
+++ b/kvmd/plugins/hid/serial.py
@@ -21,7 +21,6 @@
 
 
 import os
-import asyncio
 import multiprocessing
 import multiprocessing.queues
 import dataclasses
@@ -47,6 +46,7 @@ from ...keyboard.mappings import KEYMAP
 from ... import aiotools
 from ... import aiomulti
 from ... import aioproc
+from ... import aiogp
 
 from ...yamlconf import Option
 
@@ -186,16 +186,11 @@ class _Gpio:
         if self.__reset_pin >= 0:
             assert self.__reset_line
             if not self.__reset_wip:
+                self.__reset_wip = True
                 try:
-                    self.__reset_wip = True
-                    self.__reset_line.set_value(1)
-                    await asyncio.sleep(self.__reset_delay)
+                    await aiogp.pulse(self.__reset_line, self.__reset_delay, 1)
                 finally:
-                    try:
-                        self.__reset_line.set_value(0)
-                        await asyncio.sleep(1)
-                    finally:
-                        self.__reset_wip = False
+                    self.__reset_wip = False
                 get_logger(0).info("Reset HID performed")
             else:
                 get_logger(0).info("Another reset HID in progress")
diff --git a/kvmd/plugins/msd/relay.py b/kvmd/plugins/msd/relay.py
index 64eab56e..227dff46 100644
--- a/kvmd/plugins/msd/relay.py
+++ b/kvmd/plugins/msd/relay.py
@@ -41,6 +41,7 @@ from ...logging import get_logger
 
 from ... import aiotools
 from ... import aiofs
+from ... import aiogp
 
 from ...yamlconf import Option
 
@@ -196,16 +197,9 @@ class _Gpio:
         assert self.__target_line
         self.__target_line.set_value(1)
 
-    @contextlib.asynccontextmanager
-    async def reset(self) -> AsyncGenerator[None, None]:
+    async def reset(self) -> None:
         assert self.__reset_line
-        try:
-            self.__reset_line.set_value(1)
-            await asyncio.sleep(self.__reset_delay)
-            self.__reset_line.set_value(0)
-            yield
-        finally:
-            self.__reset_line.set_value(0)
+        await aiogp.pulse(self.__reset_line, self.__reset_delay, 0)
 
 
 # =====
@@ -302,11 +296,11 @@ class Plugin(BaseMsd):  # pylint: disable=too-many-instance-attributes
 
     @aiotools.atomic
     async def __inner_reset(self) -> None:
-        async with self.__gpio.reset():
-            self.__gpio.switch_to_local()
-            self.__connected = False
-            await self.__load_device_info()
-            get_logger(0).info("MSD reset has been successful")
+        await self.__gpio.reset()
+        self.__gpio.switch_to_local()
+        self.__connected = False
+        await self.__load_device_info()
+        get_logger(0).info("MSD reset has been successful")
 
     @aiotools.atomic
     async def cleanup(self) -> None: