libgpiod 2.x api

8 files changed, 142 insertions, 137 deletions

diff --git a/PKGBUILD b/PKGBUILD
index 10ad5bc0..048925c0 100644
--- a/PKGBUILD
+++ b/PKGBUILD
@@ -76,7 +76,7 @@ depends=(
 	python-pyrad
 	python-ldap
 	python-zstandard
-	libgpiod1
+	"libgpiod1>=2.0"
 	freetype2
 	"v4l-utils>=1.22.1-1"
 	"nginx-mainline>=1.25.1"
diff --git a/kvmd/aiogp.py b/kvmd/aiogp.py
index 7965666b..7559a9d9 100644
--- a/kvmd/aiogp.py
+++ b/kvmd/aiogp.py
@@ -30,16 +30,6 @@ from . import aiotools
 
 
 # =====
-async def pulse(line: gpiod.Line, delay: float, final: float, inverted: bool=False) -> None:
-    try:
-        line.set_value(int(not inverted))
-        await asyncio.sleep(delay)
-    finally:
-        line.set_value(int(inverted))
-        await asyncio.sleep(final)
-
-
-# =====
 @dataclasses.dataclass(frozen=True)
 class AioReaderPinParams:
     inverted: bool
@@ -57,7 +47,7 @@ class AioReader:  # pylint: disable=too-many-instance-attributes
 
         self.__path = path
         self.__consumer = consumer
-        self.__pins = pins
+        self.__pins = dict(pins)
         self.__notifier = notifier
 
         self.__values: (dict[int, _DebouncedValue] | None) = None
@@ -87,44 +77,47 @@ class AioReader:  # pylint: disable=too-many-instance-attributes
     def __run(self) -> None:
         assert self.__values is None
         assert self.__loop
-        with gpiod.Chip(self.__path) as chip:
-            pins = sorted(self.__pins)
-            lines = chip.get_lines(pins)
-            lines.request(self.__consumer, gpiod.LINE_REQ_EV_BOTH_EDGES)
 
-            lines.event_wait(nsec=1)
+        pins = sorted(self.__pins)
+        with gpiod.request_lines(
+            self.__path,
+            consumer=self.__consumer,
+            config={tuple(pins): gpiod.LineSettings(edge_detection=gpiod.line.Edge.BOTH)},
+        ) as line_request:
+
+            line_request.wait_edge_events(0.1)
             self.__values = {
                 pin: _DebouncedValue(
-                    initial=bool(value),
+                    initial=bool(value.value),
                     debounce=self.__pins[pin].debounce,
                     notifier=self.__notifier,
                     loop=self.__loop,
                 )
-                for (pin, value) in zip(pins, lines.get_values())
+                for (pin, value) in zip(pins, line_request.get_values(pins))
             }
             self.__loop.call_soon_threadsafe(self.__notifier.notify)
 
             while not self.__stop_event.is_set():
-                ev_lines = lines.event_wait(1)
-                if ev_lines:
-                    for ev_line in ev_lines:
-                        events = ev_line.event_read_multiple()
-                        if events:
-                            (pin, value) = self.__parse_event(events[-1])
-                            self.__values[pin].set(bool(value))
+                if line_request.wait_edge_events(1):
+                    new: dict[int, bool] = {}
+                    for event in line_request.read_edge_events():
+                        (pin, value) = self.__parse_event(event)
+                        new[pin] = value
+                    for (pin, value) in new.items():
+                        self.__values[pin].set(value)
                 else:  # Timeout
+                    # XXX: Лимит был актуален для 1.6. Надо проверить, поменялось ли это в 2.x.
                     # Размер буфера ядра - 16 эвентов на линии. При превышении этого числа,
                     # новые эвенты потеряются. Это не баг, это фича, как мне объяснили в LKML.
                     # Штош. Будем с этим жить и синхронизировать состояния при таймауте.
-                    for (pin, value) in zip(pins, lines.get_values()):
-                        self.__values[pin].set(bool(value))
-
-    def __parse_event(self, event: gpiod.LineEvent) -> tuple[int, int]:
-        pin = event.source.offset()
-        if event.type == gpiod.LineEvent.RISING_EDGE:
-            return (pin, 1)
-        elif event.type == gpiod.LineEvent.FALLING_EDGE:
-            return (pin, 0)
+                    for (pin, value) in zip(pins, line_request.get_values(pins)):
+                        self.__values[pin].set(bool(value.value))  # type: ignore
+
+    def __parse_event(self, event: gpiod.EdgeEvent) -> tuple[int, bool]:
+        if event.event_type == event.Type.RISING_EDGE:
+            return (event.line_offset, True)
+        elif event.event_type == event.Type.FALLING_EDGE:
+            return (event.line_offset, False)
         raise RuntimeError(f"Invalid event {event} type: {event.type}")
 
 
diff --git a/kvmd/plugins/atx/gpio.py b/kvmd/plugins/atx/gpio.py
index 58ed8a81..476dacb0 100644
--- a/kvmd/plugins/atx/gpio.py
+++ b/kvmd/plugins/atx/gpio.py
@@ -20,6 +20,8 @@
 # ========================================================================== #
 
 
+import asyncio
+
 from typing import AsyncGenerator
 
 import gpiod
@@ -74,13 +76,11 @@ class Plugin(BaseAtx):  # pylint: disable=too-many-instance-attributes
         self.__notifier = aiotools.AioNotifier()
         self.__region = aiotools.AioExclusiveRegion(AtxIsBusyError, self.__notifier)
 
-        self.__chip: (gpiod.Chip | None) = None
-        self.__power_switch_line: (gpiod.Line | None) = None
-        self.__reset_switch_line: (gpiod.Line | None) = None
+        self.__line_request: (gpiod.LineRequest | None) = None
 
         self.__reader = aiogp.AioReader(
             path=self.__device_path,
-            consumer="kvmd::atx::leds",
+            consumer="kvmd::atx",
             pins={
                 power_led_pin: aiogp.AioReaderPinParams(power_led_inverted, power_led_debounce),
                 hdd_led_pin: aiogp.AioReaderPinParams(hdd_led_inverted, hdd_led_debounce),
@@ -108,17 +108,17 @@ class Plugin(BaseAtx):  # pylint: disable=too-many-instance-attributes
         }
 
     def sysprep(self) -> None:
-        assert self.__chip is None
-        assert self.__power_switch_line is None
-        assert self.__reset_switch_line is None
-
-        self.__chip = gpiod.Chip(self.__device_path)
-
-        self.__power_switch_line = self.__chip.get_line(self.__power_switch_pin)
-        self.__power_switch_line.request("kvmd::atx::power_switch", gpiod.LINE_REQ_DIR_OUT, default_vals=[0])
-
-        self.__reset_switch_line = self.__chip.get_line(self.__reset_switch_pin)
-        self.__reset_switch_line.request("kvmd::atx::reset_switch", gpiod.LINE_REQ_DIR_OUT, default_vals=[0])
+        assert self.__line_request is None
+        self.__line_request = gpiod.request_lines(
+            self.__device_path,
+            consumer="kvmd::atx",
+            config={
+                (self.__power_switch_pin, self.__reset_switch_pin): gpiod.LineSettings(
+                    direction=gpiod.line.Direction.OUTPUT,
+                    output_value=gpiod.line.Value(False),
+                ),
+            },
+        )
 
     async def get_state(self) -> dict:
         return {
@@ -143,9 +143,9 @@ class Plugin(BaseAtx):  # pylint: disable=too-many-instance-attributes
         await self.__reader.poll()
 
     async def cleanup(self) -> None:
-        if self.__chip:
+        if self.__line_request:
             try:
-                self.__chip.close()
+                self.__line_request.release()
             except Exception:
                 pass
 
@@ -170,13 +170,13 @@ class Plugin(BaseAtx):  # pylint: disable=too-many-instance-attributes
     # =====
 
     async def click_power(self, wait: bool) -> None:
-        await self.__click("power", self.__power_switch_line, self.__click_delay, wait)
+        await self.__click("power", self.__power_switch_pin, self.__click_delay, wait)
 
     async def click_power_long(self, wait: bool) -> None:
-        await self.__click("power_long", self.__power_switch_line, self.__long_click_delay, wait)
+        await self.__click("power_long", self.__power_switch_pin, self.__long_click_delay, wait)
 
     async def click_reset(self, wait: bool) -> None:
-        await self.__click("reset", self.__reset_switch_line, self.__click_delay, wait)
+        await self.__click("reset", self.__reset_switch_pin, self.__click_delay, wait)
 
     # =====
 
@@ -184,17 +184,23 @@ class Plugin(BaseAtx):  # pylint: disable=too-many-instance-attributes
         return (await self.get_state())["leds"]["power"]
 
     @aiotools.atomic_fg
-    async def __click(self, name: str, line: gpiod.Line, delay: float, wait: bool) -> None:
+    async def __click(self, name: str, pin: int, delay: float, wait: bool) -> None:
         if wait:
             async with self.__region:
-                await self.__inner_click(name, line, delay)
+                await self.__inner_click(name, pin, delay)
         else:
             await aiotools.run_region_task(
                 f"Can't perform ATX {name} click or operation was not completed",
-                self.__region, self.__inner_click, name, line, delay,
+                self.__region, self.__inner_click, name, pin, delay,
             )
 
     @aiotools.atomic_fg
-    async def __inner_click(self, name: str, line: gpiod.Line, delay: float) -> None:
-        await aiogp.pulse(line, delay, 1)
+    async def __inner_click(self, name: str, pin: int, delay: float) -> None:
+        assert self.__line_request
+        try:
+            self.__line_request.set_value(pin, gpiod.line.Value(True))
+            await asyncio.sleep(delay)
+        finally:
+            self.__line_request.set_value(pin, gpiod.line.Value(False))
+            await asyncio.sleep(1)
         get_logger(0).info("Clicked ATX button %r", name)
diff --git a/kvmd/plugins/hid/_mcu/gpio.py b/kvmd/plugins/hid/_mcu/gpio.py
index 07b0b598..648ab447 100644
--- a/kvmd/plugins/hid/_mcu/gpio.py
+++ b/kvmd/plugins/hid/_mcu/gpio.py
@@ -47,30 +47,29 @@ class Gpio:  # pylint: disable=too-many-instance-attributes
         self.__reset_inverted = reset_inverted
         self.__reset_delay = reset_delay
 
-        self.__chip: (gpiod.Chip | None) = None
-        self.__power_detect_line: (gpiod.Line | None) = None
-        self.__reset_line: (gpiod.Line | None) = None
-
+        self.__line_request: (gpiod.LineRequest | None) = None
         self.__last_power: (bool | None) = None
 
     def __enter__(self) -> None:
         if self.__power_detect_pin >= 0 or self.__reset_pin >= 0:
-            assert self.__chip is None
-            self.__chip = gpiod.Chip(self.__device_path)
+            assert self.__line_request is None
+            config: dict[int, gpiod.LineSettings] = {}
             if self.__power_detect_pin >= 0:
-                assert self.__power_detect_line is None
-                self.__power_detect_line = self.__chip.get_line(self.__power_detect_pin)
-                self.__power_detect_line.request(
-                    "kvmd::hid::power_detect", gpiod.LINE_REQ_DIR_IN,
-                    flags=(gpiod.LINE_REQ_FLAG_BIAS_PULL_DOWN if self.__power_detect_pull_down else 0),
+                config[self.__power_detect_pin] = gpiod.LineSettings(
+                    direction=gpiod.line.Direction.INPUT,
+                    bias=(gpiod.line.Bias.PULL_DOWN if self.__power_detect_pull_down else gpiod.line.Bias.AS_IS),
                 )
             if self.__reset_pin >= 0:
-                assert self.__reset_line is None
-                self.__reset_line = self.__chip.get_line(self.__reset_pin)
-                self.__reset_line.request(
-                    "kvmd::hid::reset", gpiod.LINE_REQ_DIR_OUT,
-                    default_vals=[int(self.__reset_inverted)],
+                config[self.__reset_pin] = gpiod.LineSettings(
+                    direction=gpiod.line.Direction.OUTPUT,
+                    output_value=gpiod.line.Value(self.__reset_inverted),
                 )
+            assert len(config) > 0
+            self.__line_request = gpiod.request_lines(
+                self.__device_path,
+                consumer="kvmd::hid",
+                config=config,
+            )
 
     def __exit__(
         self,
@@ -79,19 +78,18 @@ class Gpio:  # pylint: disable=too-many-instance-attributes
         _tb: types.TracebackType,
     ) -> None:
 
-        if self.__chip:
+        if self.__line_request:
             try:
-                self.__chip.close()
+                self.__line_request.release()
             except Exception:
                 pass
             self.__last_power = None
-            self.__power_detect_line = None
-            self.__reset_line = None
-            self.__chip = None
+            self.__line_request = None
 
     def is_powered(self) -> bool:
-        if self.__power_detect_line is not None:
-            power = bool(self.__power_detect_line.get_value())
+        if self.__power_detect_pin >= 0:
+            assert self.__line_request
+            power = bool(self.__line_request.get_value(self.__power_detect_pin).value)
             if power != self.__last_power:
                 get_logger(0).info("HID power state changed: %s -> %s", self.__last_power, power)
                 self.__last_power = power
@@ -100,11 +98,11 @@ class Gpio:  # pylint: disable=too-many-instance-attributes
 
     def reset(self) -> None:
         if self.__reset_pin >= 0:
-            assert self.__reset_line
+            assert self.__line_request
             try:
-                self.__reset_line.set_value(int(not self.__reset_inverted))
+                self.__line_request.set_value(self.__reset_pin, gpiod.line.Value(not self.__reset_inverted))
                 time.sleep(self.__reset_delay)
             finally:
-                self.__reset_line.set_value(int(self.__reset_inverted))
+                self.__line_request.set_value(self.__reset_pin, gpiod.line.Value(self.__reset_inverted))
                 time.sleep(1)
             get_logger(0).info("Reset HID performed")
diff --git a/kvmd/plugins/hid/spi.py b/kvmd/plugins/hid/spi.py
index 6b093f38..74c3d3b8 100644
--- a/kvmd/plugins/hid/spi.py
+++ b/kvmd/plugins/hid/spi.py
@@ -121,7 +121,7 @@ class _SpiPhy(BasePhy):  # pylint: disable=too-many-instance-attributes
 
     @contextlib.contextmanager
     def connected(self) -> Generator[_SpiPhyConnection, None, None]:  # type: ignore
-        with self.__sw_cs_connected() as sw_cs_line:
+        with self.__sw_cs_connected() as switch_cs:
             with contextlib.closing(spidev.SpiDev(self.__bus, self.__chip)) as spi:
                 spi.mode = 0
                 spi.no_cs = (not self.__hw_cs)
@@ -129,12 +129,12 @@ class _SpiPhy(BasePhy):  # pylint: disable=too-many-instance-attributes
 
                 def inner_xfer(data: bytes) -> bytes:
                     try:
-                        if sw_cs_line is not None:
-                            sw_cs_line.set_value(0)
+                        if switch_cs is not None:
+                            switch_cs(False)
                         return spi.xfer(data, self.__max_freq, self.__block_usec)
                     finally:
-                        if sw_cs_line is not None:
-                            sw_cs_line.set_value(1)
+                        if switch_cs is not None:
+                            switch_cs(True)
 
                 if self.__sw_cs_per_byte:
                     # Режим для Pico, когда CS должен взводиться для отдельных байтов
@@ -153,12 +153,19 @@ class _SpiPhy(BasePhy):  # pylint: disable=too-many-instance-attributes
                 )
 
     @contextlib.contextmanager
-    def __sw_cs_connected(self) -> Generator[(gpiod.Line | None), None, None]:
+    def __sw_cs_connected(self) -> Generator[(Callable[[bool], bool] | None), None, None]:
         if self.__sw_cs_pin > 0:
-            with contextlib.closing(gpiod.Chip(self.__gpio_device_path)) as chip:
-                line = chip.get_line(self.__sw_cs_pin)
-                line.request("kvmd::hid::sw_cs", gpiod.LINE_REQ_DIR_OUT, default_vals=[1])
-                yield line
+            with gpiod.request_lines(
+                self.__gpio_device_path,
+                consumer="kvmd::hid",
+                config={
+                    self.__sw_cs_pin: gpiod.LineSettings(
+                        direction=gpiod.line.Direction.OUTPUT,
+                        output_value=gpiod.line.Value(True),
+                    ),
+                },
+            ) as line_request:
+                yield (lambda state: line_request.set_value(self.__sw_cs_pin, gpiod.line.Value(state)))
         else:
             yield None
 
diff --git a/kvmd/plugins/ugpio/gpio.py b/kvmd/plugins/ugpio/gpio.py
index 39c379f3..b86799a1 100644
--- a/kvmd/plugins/ugpio/gpio.py
+++ b/kvmd/plugins/ugpio/gpio.py
@@ -54,9 +54,7 @@ class Plugin(BaseUserGpioDriver):
         self.__output_pins: dict[int, (bool | None)] = {}
 
         self.__reader: (aiogp.AioReader | None) = None
-
-        self.__chip: (gpiod.Chip | None) = None
-        self.__output_lines: dict[int, gpiod.Line] = {}
+        self.__outputs_request: (gpiod.LineRequest | None) = None
 
     @classmethod
     def get_plugin_options(cls) -> dict:
@@ -76,27 +74,34 @@ class Plugin(BaseUserGpioDriver):
 
     def prepare(self) -> None:
         assert self.__reader is None
+        assert self.__outputs_request is None
         self.__reader = aiogp.AioReader(
             path=self.__device_path,
             consumer="kvmd::gpio::inputs",
             pins=self.__input_pins,
             notifier=self._notifier,
         )
-
-        self.__chip = gpiod.Chip(self.__device_path)
-        for (pin, initial) in self.__output_pins.items():
-            line = self.__chip.get_line(pin)
-            line.request("kvmd::gpio::outputs", gpiod.LINE_REQ_DIR_OUT, default_vals=[int(initial or False)])
-            self.__output_lines[pin] = line
+        if self.__output_pins:
+            self.__outputs_request = gpiod.request_lines(
+                self.__device_path,
+                consumer="kvmd::gpiod::outputs",
+                config={
+                    pin: gpiod.LineSettings(
+                        direction=gpiod.line.Direction.OUTPUT,
+                        output_value=gpiod.line.Value(initial or False),
+                    )
+                    for (pin, initial) in self.__output_pins.items()
+                },
+            )
 
     async def run(self) -> None:
         assert self.__reader
         await self.__reader.poll()
 
     async def cleanup(self) -> None:
-        if self.__chip:
+        if self.__outputs_request:
             try:
-                self.__chip.close()
+                self.__outputs_request.release()
             except Exception:
                 pass
 
@@ -105,10 +110,15 @@ class Plugin(BaseUserGpioDriver):
         pin_int = int(pin)
         if pin_int in self.__input_pins:
             return self.__reader.get(pin_int)
-        return bool(self.__output_lines[pin_int].get_value())
+        assert self.__outputs_request
+        assert pin_int in self.__output_pins
+        return bool(self.__outputs_request.get_value(pin_int).value)
 
     async def write(self, pin: str, state: bool) -> None:
-        self.__output_lines[int(pin)].set_value(int(state))
+        assert self.__outputs_request
+        pin_int = int(pin)
+        assert pin_int in self.__output_pins
+        self.__outputs_request.set_value(pin_int, gpiod.line.Value(state))
 
     def __str__(self) -> str:
         return f"GPIO({self._instance_name})"
diff --git a/kvmd/plugins/ugpio/locator.py b/kvmd/plugins/ugpio/locator.py
index b30c0530..79c10df7 100644
--- a/kvmd/plugins/ugpio/locator.py
+++ b/kvmd/plugins/ugpio/locator.py
@@ -53,9 +53,7 @@ class Plugin(BaseUserGpioDriver):
         self.__device_path = device_path
 
         self.__tasks: dict[int, (asyncio.Task | None)] = {}
-
-        self.__chip: (gpiod.Chip | None) = None
-        self.__lines: dict[int, gpiod.Line] = {}
+        self.__line_request: (gpiod.LineRequest | None) = None
 
     @classmethod
     def get_plugin_options(cls) -> dict:
@@ -76,11 +74,16 @@ class Plugin(BaseUserGpioDriver):
         self.__tasks[int(pin)] = None
 
     def prepare(self) -> None:
-        self.__chip = gpiod.Chip(self.__device_path)
-        for pin in self.__tasks:
-            line = self.__chip.get_line(pin)
-            line.request("kvmd::locator::outputs", gpiod.LINE_REQ_DIR_OUT, default_vals=[0])
-            self.__lines[pin] = line
+        self.__line_request = gpiod.request_lines(
+            self.__device_path,
+            consumer="kvmd::locator",
+            config={
+                tuple(self.__tasks): gpiod.LineSettings(
+                    direction=gpiod.line.Direction.OUTPUT,
+                    output_value=gpiod.line.Value(False),
+                ),
+            },
+        )
 
     async def cleanup(self) -> None:
         tasks = [
@@ -91,9 +94,9 @@ class Plugin(BaseUserGpioDriver):
         for task in tasks:
             task.cancel()
         await asyncio.gather(*tasks, return_exceptions=True)
-        if self.__chip:
+        if self.__line_request:
             try:
-                self.__chip.close()
+                self.__line_request.release()
             except Exception:
                 pass
 
@@ -111,17 +114,18 @@ class Plugin(BaseUserGpioDriver):
             self.__tasks[pin_int] = None
 
     async def __blink(self, pin: int) -> None:
-        line = self.__lines[pin]
+        assert pin in self.__tasks
+        assert self.__line_request
         try:
-            state = 1
+            state = True
             while True:
-                line.set_value(state)
-                state = int(not state)
+                self.__line_request.set_value(pin, gpiod.line.Value(state))
+                state = (not state)
                 await asyncio.sleep(0.1)
         except asyncio.CancelledError:
             pass
         finally:
-            line.set_value(0)
+            self.__line_request.set_value(pin, gpiod.line.Value(False))
 
     def __str__(self) -> str:
         return f"Locator({self._instance_name})"
diff --git a/testenv/Dockerfile b/testenv/Dockerfile
index 2874fed6..036a8380 100644
--- a/testenv/Dockerfile
+++ b/testenv/Dockerfile
@@ -95,19 +95,6 @@ RUN git clone https://github.com/pikvm/ustreamer \
 	&& cd - \
 	&& rm -rf ustreamer
 
-# FIXME: workaroung for legacy 1.x libgpido
-RUN pacman --noconfirm --ask=4 -R libgpiod
-ENV LIBGPIOD_PKG libgpiod-1.6.4
-RUN curl \
-		-o $LIBGPIOD_PKG.tar.gz \
-		https://git.kernel.org/pub/scm/libs/libgpiod/libgpiod.git/snapshot/$LIBGPIOD_PKG.tar.gz \
-	&& tar -xzvf $LIBGPIOD_PKG.tar.gz \
-	&& cd $LIBGPIOD_PKG \
-	&& ./autogen.sh --prefix=/usr --enable-tools=yes --enable-bindings-python \
-	&& make PREFIX=/usr install \
-	&& cd - \
-	&& rm -rf $LIBGPIOD_PKG{,.tar.gz}
-
 RUN mkdir -p \
 		/etc/kvmd/{nginx,vnc} \
 		/var/lib/kvmd/msd \