# ========================================================================== # # # # KVMD - The main PiKVM daemon. # # # # Copyright (C) 2018-2022 Maxim Devaev # # # # This program is free software: you can redistribute it and/or modify # # it under the terms of the GNU General Public License as published by # # the Free Software Foundation, either version 3 of the License, or # # (at your option) any later version. # # # # This program is distributed in the hope that it will be useful, # # but WITHOUT ANY WARRANTY; without even the implied warranty of # # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # # GNU General Public License for more details. # # # # You should have received a copy of the GNU General Public License # # along with this program. If not, see . # # # # ========================================================================== # import asyncio import threading import dataclasses import gpiod 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 debounce: float class AioReader: # pylint: disable=too-many-instance-attributes def __init__( self, path: str, consumer: str, pins: dict[int, AioReaderPinParams], notifier: aiotools.AioNotifier, ) -> None: self.__path = path self.__consumer = consumer self.__pins = pins self.__notifier = notifier self.__values: (dict[int, _DebouncedValue] | None) = None self.__thread = threading.Thread(target=self.__run, daemon=True) self.__stop_event = threading.Event() self.__loop: (asyncio.AbstractEventLoop | None) = None def get(self, pin: int) -> bool: value = (self.__values[pin].get() if self.__values is not None else False) return (value ^ self.__pins[pin].inverted) async def poll(self) -> None: if not self.__pins: await aiotools.wait_infinite() else: assert self.__loop is None self.__loop = asyncio.get_running_loop() self.__thread.start() try: await aiotools.run_async(self.__thread.join) finally: self.__stop_event.set() await aiotools.run_async(self.__thread.join) 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) self.__values = { pin: _DebouncedValue( initial=bool(value), debounce=self.__pins[pin].debounce, notifier=self.__notifier, loop=self.__loop, ) for (pin, value) in zip(pins, lines.get_values()) } 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)) else: # Timeout # Размер буфера ядра - 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) raise RuntimeError(f"Invalid event {event} type: {event.type}") class _DebouncedValue: def __init__( self, initial: bool, debounce: float, notifier: aiotools.AioNotifier, loop: asyncio.AbstractEventLoop, ) -> None: self.__value = initial self.__debounce = debounce self.__notifier = notifier self.__loop = loop self.__queue: "asyncio.Queue[bool]" = asyncio.Queue() # type: ignore self.__task = loop.create_task(self.__consumer_task_loop()) def set(self, value: bool) -> None: if self.__loop.is_running(): self.__check_alive() self.__loop.call_soon_threadsafe(self.__queue.put_nowait, value) def get(self) -> bool: return self.__value def __check_alive(self) -> None: if self.__task.done() and not self.__task.cancelled(): raise RuntimeError("Dead debounce consumer") async def __consumer_task_loop(self) -> None: while True: value = await self.__queue.get() while not self.__queue.empty(): value = await self.__queue.get() if self.__value != value: self.__value = value self.__notifier.notify() await asyncio.sleep(self.__debounce)