GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2025-12-12 09:10:30 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

ugpio plugins

Browse Source
This commit is contained in:
Devaev Maxim 2020-09-06 08:47:43 +03:00
parent e8bd1e2648
commit a6dac4bd84
18 changed files with 321 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
kvmd/apps/__init__.py
View File

@ -40,6 +40,7 @@ from ..plugins.auth import get_auth_service_class
from ..plugins.hid import get_hid_class
from ..plugins.atx import get_atx_class
from ..plugins.msd import get_msd_class
from ..plugins.ugpio import get_ugpio_driver_class
from ..yamlconf import ConfigError
from ..yamlconf import make_config
@ -174,6 +175,16 @@ def _patch_dynamic( # pylint: disable=too-many-locals
rebuild = True
if load_gpio:
for (driver, params) in { # type: ignore
"gpio": {},
**(raw_config.get("kvmd", {}).get("gpio", {}).get("drivers", {})),
}.items():
driver_type = valid_stripped_string_not_empty(params.get("type", "gpio"))
scheme["kvmd"]["gpio"]["drivers"][driver] = {
"type": Option(driver_type, type=valid_stripped_string_not_empty),
**get_ugpio_driver_class(driver_type).get_plugin_options()
}
for (channel, params) in raw_config.get("kvmd", {}).get("gpio", {}).get("scheme", {}).items():
try:
mode = valid_ugpio_mode(params.get("mode", ""))
@ -181,6 +192,7 @@ def _patch_dynamic( # pylint: disable=too-many-locals
pass
finally:
ch_scheme: Dict = {
"driver": Option("gpio"),
"pin": Option(-1, type=valid_gpio_pin),
"mode": Option("", type=valid_ugpio_mode),
"inverted": Option(False, type=valid_bool),
@ -196,7 +208,8 @@ def _patch_dynamic( # pylint: disable=too-many-locals
"max_delay": Option(0.1, type=valid_float_f01),
},
})
scheme["kvmd"]["gpio"]["scheme"][channel] = ch_scheme
scheme["kvmd"]["gpio"]["scheme"][channel] = ch_scheme
rebuild = True
return rebuild
@ -326,6 +339,7 @@ def _get_config_scheme() -> Dict:
"gpio": {
"state_poll": Option(0.1, type=valid_float_f01),
"drivers": {}, # Dynamic content
"scheme": {}, # Dymanic content
"view": {
"header": {

12
kvmd/apps/cleanup/__init__.py
View File

@ -61,16 +61,16 @@ def _clear_gpio(config: Section) -> None:
("streamer/cap", config.streamer.cap_pin),
("streamer/conv", config.streamer.conv_pin),
*([
(f"gpio/{channel}", params.pin)
for (channel, params) in config.gpio.scheme.items()
if params.mode == "output"
]),
# *([
# (f"gpio/{channel}", params.pin)
# for (channel, params) in config.gpio.scheme.items()
# if params.mode == "output"
# ]),
]:
if pin >= 0:
logger.info("Writing 0 to GPIO pin=%d (%s)", pin, name)
try:
gpio.set_output(pin, initial=False)
gpio.set_output(pin, False)
except Exception:
logger.exception("Can't clear GPIO pin=%d (%s)", pin, name)

9
kvmd/apps/kvmd/api/export.py
View File

@ -54,13 +54,18 @@ class ExportApi:
self.__user_gpio.get_state(),
])
rows: List[str] = []
self.__append_prometheus_rows(rows, atx_state["enabled"], "pikvm_atx_enabled")
self.__append_prometheus_rows(rows, atx_state["leds"]["power"], "pikvm_atx_power")
for mode in ["input", "output"]:
for (channel, gch) in gpio_state[f"{mode}s"].items():
self.__append_prometheus_rows(rows, gch["state"], f"pikvm_gpio_input_{channel}")
for (channel, ch_state) in gpio_state[f"{mode}s"].items():
for key in ["online", "state"]:
self.__append_prometheus_rows(rows, ch_state["state"], f"pikvm_gpio_{mode}_{key}_{channel}")
if hw_state is not None:
self.__append_prometheus_rows(rows, hw_state["health"], "pikvm_hw")
return Response(text="\n".join(rows))
def __append_prometheus_rows(self, rows: List[str], value: Any, path: str) -> None:

3
kvmd/apps/kvmd/api/ugpio.py
View File

@ -59,5 +59,6 @@ class UserGpioApi:
async def __pulse_handler(self, request: Request) -> Response:
channel = valid_ugpio_channel(request.query.get("channel"))
delay = valid_float_f0(request.query.get("delay", "0"))
await self.__user_gpio.pulse(channel, delay)
wait = valid_bool(request.query.get("wait", "0"))
await self.__user_gpio.pulse(channel, delay, wait)
return make_json_response()

13
kvmd/apps/kvmd/server.py
View File

@ -106,20 +106,21 @@ class StreamerResolutionNotSupported(OperationError):
# =====
@dataclasses.dataclass(frozen=True)
class _Component:
class _Component: # pylint: disable=too-many-instance-attributes
name: str
event_type: str
obj: object
sysprep: Optional[Callable[[], None]] = None
systask: Optional[Callable[[], Coroutine[Any, Any, None]]] = None
get_state: Optional[Callable[[], Coroutine[Any, Any, Dict]]] = None
poll_state: Optional[Callable[[], AsyncGenerator[Dict, None]]] = None
systask: Optional[Callable[[], Coroutine[Any, Any, None]]] = None
cleanup: Optional[Callable[[], Coroutine[Any, Any, Dict]]] = None
def __post_init__(self) -> None:
if isinstance(self.obj, BasePlugin):
object.__setattr__(self, "name", f"{self.name} ({self.obj.get_plugin_name()})")
for field in ["get_state", "poll_state", "systask", "cleanup"]:
for field in ["sysprep", "systask", "get_state", "poll_state", "cleanup"]:
object.__setattr__(self, field, getattr(self.obj, field, None))
if self.get_state or self.poll_state:
assert self.event_type, self
@ -278,7 +279,9 @@ class KvmdServer(HttpServer): # pylint: disable=too-many-arguments,too-many-ins
# ===== SYSTEM STUFF
def run(self, **kwargs: Any) -> None: # type: ignore # pylint: disable=arguments-differ
self.__hid.start()
for component in self.__components:
if component.sysprep:
component.sysprep()
aioproc.rename_process("main")
super().run(**kwargs)
@ -307,7 +310,7 @@ class KvmdServer(HttpServer): # pylint: disable=too-many-arguments,too-many-ins
async def wrapper() -> None:
try:
await method(*args)
raise RuntimeError(f"Dead system task: {method.__name__}"
raise RuntimeError(f"Dead system task: {method}"
f"({', '.join(getattr(arg, '__name__', str(arg)) for arg in args)})")
except asyncio.CancelledError:
pass

4
kvmd/apps/kvmd/streamer.py
View File

@ -141,8 +141,8 @@ class Streamer: # pylint: disable=too-many-instance-attributes
**params_kwargs: Any,
) -> None:
self.__cap_pin = (gpio.set_output(cap_pin) if cap_pin >= 0 else -1)
self.__conv_pin = (gpio.set_output(conv_pin) if conv_pin >= 0 else -1)
self.__cap_pin = (gpio.set_output(cap_pin, False) if cap_pin >= 0 else -1)
self.__conv_pin = (gpio.set_output(conv_pin, False) if conv_pin >= 0 else -1)
self.__sync_delay = sync_delay
self.__init_delay = init_delay

132
kvmd/apps/kvmd/ugpio.py
View File

@ -30,62 +30,89 @@ from typing import Optional
from ...logging import get_logger
from ...plugins.ugpio import GpioError
from ...plugins.ugpio import GpioOperationError
from ...plugins.ugpio import GpioDriverOfflineError
from ...plugins.ugpio import BaseUserGpioDriver
from ...plugins.ugpio import get_ugpio_driver_class
from ... import aiotools
from ... import gpio
from ...yamlconf import Section
from ...errors import OperationError
from ...errors import IsBusyError
# =====
class GpioChannelNotFoundError(OperationError):
class GpioChannelNotFoundError(GpioOperationError):
def __init__(self) -> None:
super().__init__("GPIO channel is not found")
class GpioSwitchNotSupported(OperationError):
class GpioSwitchNotSupported(GpioOperationError):
def __init__(self) -> None:
super().__init__("This GPIO channel does not support switching")
class GpioPulseNotSupported(OperationError):
class GpioPulseNotSupported(GpioOperationError):
def __init__(self) -> None:
super().__init__("This GPIO channel does not support pulsing")
class GpioChannelIsBusyError(IsBusyError):
class GpioChannelIsBusyError(IsBusyError, GpioError):
def __init__(self) -> None:
super().__init__("Performing another GPIO operation on this channel, please try again later")
# =====
class _GpioInput:
def __init__(self, channel: str, config: Section, reader: gpio.BatchReader) -> None:
def __init__(
self,
channel: str,
config: Section,
driver: BaseUserGpioDriver,
) -> None:
self.__channel = channel
self.__pin: int = config.pin
self.__inverted: bool = config.inverted
self.__reader = reader
self.__driver = driver
self.__driver.register_input(self.__pin)
def get_scheme(self) -> Dict:
return {}
def get_state(self) -> Dict:
return {"state": (self.__reader.get(self.__pin) ^ self.__inverted)}
(online, state) = (True, False)
try:
state = (self.__driver.read(self.__pin) ^ self.__inverted)
except GpioDriverOfflineError:
online = False
return {
"online": online,
"state": state,
}
def __str__(self) -> str:
return f"Input({self.__channel}, pin={self.__pin})"
return f"Input({self.__channel}, driver={self.__driver.get_instance_name()}, pin={self.__pin})"
__repr__ = __str__
class _GpioOutput: # pylint: disable=too-many-instance-attributes
def __init__(self, channel: str, config: Section, notifier: aiotools.AioNotifier) -> None:
def __init__(
self,
channel: str,
config: Section,
driver: BaseUserGpioDriver,
notifier: aiotools.AioNotifier,
) -> None:
self.__channel = channel
self.__pin: int = config.pin
self.__inverted: bool = config.inverted
self.__initial: bool = config.initial
self.__switch: bool = config.switch
@ -95,6 +122,9 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
self.__busy_delay: float = config.busy_delay
self.__driver = driver
self.__driver.register_output(self.__pin, (config.initial ^ config.inverted))
self.__region = aiotools.AioExclusiveRegion(GpioChannelIsBusyError, notifier)
def get_scheme(self) -> Dict:
@ -105,20 +135,31 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
"min_delay": (self.__min_pulse_delay if self.__pulse_delay else 0),
"max_delay": (self.__max_pulse_delay if self.__pulse_delay else 0),
},
"hw": {
"driver": self.__driver.get_instance_name(),
"pin": self.__pin,
},
}
def get_state(self) -> Dict:
busy = self.__region.is_busy()
(online, state) = (True, False)
if not busy:
try:
state = self.__read()
except GpioDriverOfflineError:
online = False
return {
"state": (self.__read() if not busy else False),
"online": online,
"state": state,
"busy": busy,
}
def cleanup(self) -> None:
try:
gpio.write(self.__pin, False)
self.__driver.write(self.__pin, (self.__initial ^ self.__inverted))
except Exception:
get_logger().exception("Can't cleanup GPIO %s", self)
get_logger().exception("Can't cleanup %s", self)
async def switch(self, state: bool) -> bool:
if not self.__switch:
@ -126,21 +167,25 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
async with self.__region:
if state != self.__read():
self.__write(state)
get_logger(0).info("Switched %s to %d", self, state)
get_logger(0).info("Switched %s to state=%d", self, state)
await asyncio.sleep(self.__busy_delay)
return True
await asyncio.sleep(self.__busy_delay)
return False
@aiotools.atomic
async def pulse(self, delay: float) -> None:
async def pulse(self, delay: float, wait: bool) -> None:
if not self.__pulse_delay:
raise GpioPulseNotSupported()
delay = min(max((delay or self.__pulse_delay), self.__min_pulse_delay), self.__max_pulse_delay)
await aiotools.run_region_task(
f"Can't perform pulse of {self} or operation was not completed",
self.__region, self.__inner_pulse, delay,
)
if wait:
async with self.__region:
await self.__inner_pulse(delay)
else:
await aiotools.run_region_task(
f"Can't perform pulse of {self} or operation was not completed",
self.__region, self.__inner_pulse, delay,
)
@aiotools.atomic
async def __inner_pulse(self, delay: float) -> None:
@ -153,13 +198,13 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
get_logger(0).info("Pulsed %s with delay=%.2f", self, delay)
def __read(self) -> bool:
return (gpio.read(self.__pin) ^ self.__inverted)
return (self.__driver.read(self.__pin) ^ self.__inverted)
def __write(self, state: bool) -> None:
gpio.write(self.__pin, (state ^ self.__inverted))
self.__driver.write(self.__pin, (state ^ self.__inverted))
def __str__(self) -> str:
return f"Output({self.__channel}, pin={self.__pin})"
return f"Output({self.__channel}, driver={self.__driver.get_instance_name()}, pin={self.__pin})"
__repr__ = __str__
@ -170,27 +215,23 @@ class UserGpio:
self.__view = config.view
self.__state_notifier = aiotools.AioNotifier()
self.__reader = gpio.BatchReader(
pins=[
(
gpio.set_input(ch_config.pin)
if ch_config.mode == "input" else
gpio.set_output(ch_config.pin, (ch_config.initial ^ ch_config.inverted))
)
for ch_config in config.scheme.values()
],
interval=config.state_poll,
notifier=self.__state_notifier,
)
self.__drivers = {
driver: get_ugpio_driver_class(drv_config.type)(**drv_config._unpack(ignore=["type"]))
for (driver, drv_config) in config.drivers.items()
}
self.__inputs: Dict[str, _GpioInput] = {}
self.__outputs: Dict[str, _GpioOutput] = {}
for (channel, ch_config) in sorted(config.scheme.items(), key=operator.itemgetter(0)):
driver = self.__drivers.get(ch_config.driver)
if driver is None:
raise RuntimeError(f"Missing User-GPIO driver configuration: {ch_config.driver}")
if ch_config.mode == "input":
self.__inputs[channel] = _GpioInput(channel, ch_config, self.__reader)
self.__inputs[channel] = _GpioInput(channel, ch_config, driver)
else: # output:
self.__outputs[channel] = _GpioOutput(channel, ch_config, self.__state_notifier)
self.__outputs[channel] = _GpioOutput(channel, ch_config, driver, self.__state_notifier)
async def get_model(self) -> Dict:
return {
@ -216,13 +257,26 @@ class UserGpio:
prev_state = state
await self.__state_notifier.wait()
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)
async def systask(self) -> None:
get_logger(0).info("Polling User-GPIO inputs ...")
await self.__reader.poll()
get_logger(0).info("Running User-GPIO drivers ...")
await asyncio.gather(*[
driver.run()
for (_, driver) in sorted(self.__drivers.items(), key=operator.itemgetter(0))
])
async def cleanup(self) -> None:
for gout in self.__outputs.values():
gout.cleanup()
for driver in self.__drivers.values():
try:
driver.cleanup()
except Exception:
get_logger().exception("Can't cleanup driver %r", driver.get_instance_name())
async def switch(self, channel: str, state: bool) -> bool:
gout = self.__outputs.get(channel)

12
kvmd/gpio.py
View File

@ -24,7 +24,7 @@ import asyncio
import contextlib
from typing import Tuple
from typing import List
from typing import Set
from typing import Generator
from typing import Optional
@ -48,7 +48,7 @@ def bcm() -> Generator[None, None, None]:
logger.info("GPIO cleaned")
def set_output(pin: int, initial: bool=False) -> int:
def set_output(pin: int, initial: Optional[bool]) -> int:
assert pin >= 0, pin
GPIO.setup(pin, GPIO.OUT, initial=initial)
return pin
@ -71,10 +71,10 @@ def write(pin: int, state: bool) -> None:
class BatchReader:
def __init__(self, pins: List[int], interval: float, notifier: aiotools.AioNotifier) -> None:
self.__pins = pins
self.__flags: Tuple[Optional[bool], ...] = (None,) * len(pins)
self.__state = {pin: read(pin) for pin in pins}
def __init__(self, pins: Set[int], interval: float, notifier: aiotools.AioNotifier) -> None:
self.__pins = sorted(pins)
self.__flags: Tuple[Optional[bool], ...] = (None,) * len(self.__pins)
self.__state = {pin: read(pin) for pin in self.__pins}
self.__interval = interval
self.__notifier = notifier

6
kvmd/plugins/atx/gpio.py
View File

@ -62,8 +62,8 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
self.__power_led_pin = gpio.set_input(power_led_pin)
self.__hdd_led_pin = gpio.set_input(hdd_led_pin)
self.__power_switch_pin = gpio.set_output(power_switch_pin)
self.__reset_switch_pin = gpio.set_output(reset_switch_pin)
self.__power_switch_pin = gpio.set_output(power_switch_pin, False)
self.__reset_switch_pin = gpio.set_output(reset_switch_pin, False)
self.__power_led_inverted = power_led_inverted
self.__hdd_led_inverted = hdd_led_inverted
@ -75,7 +75,7 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
self.__region = aiotools.AioExclusiveRegion(AtxIsBusyError, self.__state_notifier)
self.__reader = gpio.BatchReader(
pins=[self.__power_led_pin, self.__hdd_led_pin],
pins=set([self.__power_led_pin, self.__hdd_led_pin]),
interval=state_poll,
notifier=self.__state_notifier,
)

4
kvmd/plugins/hid/__init__.py
View File

@ -32,8 +32,8 @@ from .. import get_plugin_class
# =====
class BaseHid(BasePlugin):
def start(self) -> None:
pass
def sysprep(self) -> None:
raise NotImplementedError
async def get_state(self) -> Dict:
raise NotImplementedError

2
kvmd/plugins/hid/otg/__init__.py
View File

@ -74,7 +74,7 @@ class Plugin(BaseHid):
"noop": Option(False, type=valid_bool),
}
def start(self) -> None:
def sysprep(self) -> None:
self.__keyboard_proc.start()
self.__mouse_proc.start()

6
kvmd/plugins/hid/serial.py
View File

@ -175,7 +175,7 @@ class Plugin(BaseHid, multiprocessing.Process): # pylint: disable=too-many-inst
multiprocessing.Process.__init__(self, daemon=True)
self.__reset_pin = gpio.set_output(reset_pin)
self.__reset_pin = gpio.set_output(reset_pin, False)
self.__reset_delay = reset_delay
self.__device_path = device_path
@ -217,9 +217,9 @@ class Plugin(BaseHid, multiprocessing.Process): # pylint: disable=too-many-inst
"noop": Option(False, type=valid_bool),
}
def start(self) -> None:
def sysprep(self) -> None:
get_logger(0).info("Starting HID daemon ...")
multiprocessing.Process.start(self)
self.start()
async def get_state(self) -> Dict:
state = await self.__state_flags.get()

4
kvmd/plugins/msd/relay.py
View File

@ -165,8 +165,8 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
reset_delay: float,
) -> None:
self.__target_pin = gpio.set_output(target_pin)
self.__reset_pin = gpio.set_output(reset_pin)
self.__target_pin = gpio.set_output(target_pin, False)
self.__reset_pin = gpio.set_output(reset_pin, False)
self.__device_path = device_path
self.__init_delay = init_delay

77
kvmd/plugins/ugpio/__init__.py Normal file
View File

@ -0,0 +1,77 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# 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 <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
from typing import Type
from typing import Optional
from ...errors import OperationError
from ... import aiotools
from .. import BasePlugin
from .. import get_plugin_class
# =====
class GpioError(Exception):
pass
class GpioOperationError(OperationError, GpioError):
pass
class GpioDriverOfflineError(GpioOperationError):
def __init__(self, driver: "BaseUserGpioDriver") -> None:
super().__init__(f"GPIO driver {driver.get_instance_name()!r} is offline")
# =====
class BaseUserGpioDriver(BasePlugin):
def get_instance_name(self) -> str:
raise NotImplementedError
def register_input(self, pin: int) -> None:
raise NotImplementedError
def register_output(self, pin: int, initial: Optional[bool]) -> None:
raise NotImplementedError
def prepare(self, notifier: aiotools.AioNotifier) -> None:
raise NotImplementedError
async def run(self) -> None:
raise NotImplementedError
def cleanup(self) -> None:
pass
def read(self, pin: int) -> bool:
raise NotImplementedError
def write(self, pin: int, state: bool) -> None:
raise NotImplementedError
# =====
def get_ugpio_driver_class(name: str) -> Type[BaseUserGpioDriver]:
return get_plugin_class("ugpio", name) # type: ignore

86
kvmd/plugins/ugpio/gpio.py Normal file
View File

@ -0,0 +1,86 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# 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 <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
from typing import Dict
from typing import Set
from typing import Optional
from ... import aiotools
from ... import gpio
from ...yamlconf import Option
from ...validators.basic import valid_float_f01
from . import BaseUserGpioDriver
# =====
class Plugin(BaseUserGpioDriver):
def __init__(self, state_poll: float) -> None: # pylint: disable=super-init-not-called
self.__state_poll = state_poll
self.__input_pins: Set[int] = set()
self.__output_pins: Dict[int, Optional[bool]] = {}
self.__reader: Optional[gpio.BatchReader] = None
@classmethod
def get_plugin_options(cls) -> Dict:
return {
"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:
assert self.__reader is None
self.__reader = gpio.BatchReader(
pins=set([
*map(gpio.set_input, self.__input_pins),
*[
gpio.set_output(pin, initial)
for (pin, initial) in self.__output_pins.items()
],
]),
interval=self.__state_poll,
notifier=notifier,
)
async def run(self) -> None:
assert self.__reader
await self.__reader.poll()
def read(self, pin: int) -> bool:
assert self.__reader
return self.__reader.get(pin)
def write(self, pin: int, state: bool) -> None:
assert self.__reader
gpio.write(pin, state)

1
setup.py
View File

@ -87,6 +87,7 @@ def main() -> None:
"kvmd.plugins.atx",
"kvmd.plugins.msd",
"kvmd.plugins.msd.otg",
"kvmd.plugins.ugpio",
"kvmd.clients",
"kvmd.apps",
"kvmd.apps.kvmd",

4
testenv/tests/test_gpio.py
View File

@ -29,7 +29,7 @@ from kvmd import gpio
@pytest.mark.parametrize("pin", [0, 1, 13])
def test_ok__loopback_initial_false(pin: int) -> None:
with gpio.bcm():
assert gpio.set_output(pin) == pin
assert gpio.set_output(pin, False) == pin
assert gpio.read(pin) is False
gpio.write(pin, True)
assert gpio.read(pin) is True
@ -53,6 +53,6 @@ def test_ok__input(pin: int) -> None:
def test_fail__invalid_pin() -> None:
with pytest.raises(AssertionError):
gpio.set_output(-1)
gpio.set_output(-1, False)
with pytest.raises(AssertionError):
gpio.set_input(-1)

5
testenv/v2-hdmi-rpi4.override.yaml
View File

@ -38,6 +38,11 @@ kvmd:
- "--no-log-colors"
gpio:
drivers:
gpio2:
type: gpio
state_poll: 0.3
scheme:
host1: # any name like foo_bar_baz
pin: 1