diff options
Diffstat (limited to 'kvmd')
| -rw-r--r-- | kvmd/aiogp.py | 61 | ||||
| -rw-r--r-- | kvmd/plugins/atx/gpio.py | 56 | ||||
| -rw-r--r-- | kvmd/plugins/hid/_mcu/gpio.py | 50 | ||||
| -rw-r--r-- | kvmd/plugins/hid/spi.py | 27 | ||||
| -rw-r--r-- | kvmd/plugins/ugpio/gpio.py | 36 | ||||
| -rw-r--r-- | kvmd/plugins/ugpio/locator.py | 34 |
6 files changed, 141 insertions, 123 deletions
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})" |