configurable gpio devices

This commit is contained in:
Devaev Maxim
2020-12-25 11:08:56 +03:00
parent 4447e49abb
commit 0adfe17f70
11 changed files with 53 additions and 21 deletions

View File

@@ -28,7 +28,6 @@ import gpiod
from ...logging import get_logger
from ... import env
from ... import aiotools
from ... import aiogp
@@ -37,6 +36,7 @@ from ...yamlconf import Option
from ...validators.basic import valid_bool
from ...validators.basic import valid_float_f0
from ...validators.basic import valid_float_f01
from ...validators.os import valid_abs_path
from ...validators.hw import valid_gpio_pin
from . import AtxIsBusyError
@@ -47,6 +47,8 @@ from . import BaseAtx
class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
def __init__( # pylint: disable=too-many-arguments,super-init-not-called
self,
device_path: str,
power_led_pin: int,
power_led_inverted: bool,
power_led_debounce: float,
@@ -61,6 +63,8 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
long_click_delay: float,
) -> None:
self.__device_path = device_path
self.__power_led_pin = power_led_pin
self.__hdd_led_pin = hdd_led_pin
self.__power_switch_pin = power_switch_pin
@@ -77,7 +81,7 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
self.__reset_switch_line: Optional[gpiod.Line] = None
self.__reader = aiogp.AioReader(
path=env.GPIO_DEVICE_PATH,
path=self.__device_path,
consumer="kvmd::atx::leds",
pins={
power_led_pin: aiogp.AioReaderPinParams(power_led_inverted, power_led_debounce),
@@ -89,6 +93,8 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
@classmethod
def get_plugin_options(cls) -> Dict:
return {
"device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="device_path"),
"power_led_pin": Option(-1, type=valid_gpio_pin),
"power_led_inverted": Option(False, type=valid_bool),
"power_led_debounce": Option(0.1, type=valid_float_f0),
@@ -108,7 +114,7 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
assert self.__power_switch_line is None
assert self.__reset_switch_line is None
self.__chip = gpiod.Chip(env.GPIO_DEVICE_PATH)
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])

View File

@@ -46,6 +46,7 @@ from ....validators.basic import valid_bool
from ....validators.basic import valid_int_f0
from ....validators.basic import valid_int_f1
from ....validators.basic import valid_float_f01
from ....validators.os import valid_abs_path
from ....validators.hw import valid_gpio_pin_optional
from .. import BaseHid
@@ -113,6 +114,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
self,
phy: BasePhy,
gpio_device_path: str,
reset_pin: int,
reset_inverted: bool,
reset_delay: float,
@@ -133,7 +135,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
self.__noop = noop
self.__phy = phy
self.__gpio = Gpio(reset_pin, reset_inverted, reset_delay)
self.__gpio = Gpio(gpio_device_path, reset_pin, reset_inverted, reset_delay)
self.__events_queue: "multiprocessing.Queue[BaseEvent]" = multiprocessing.Queue()
@@ -149,6 +151,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
@classmethod
def get_plugin_options(cls) -> Dict:
return {
"gpio_device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="gpio_device_path"),
"reset_pin": Option(-1, type=valid_gpio_pin_optional),
"reset_inverted": Option(False, type=valid_bool),
"reset_delay": Option(0.1, type=valid_float_f01),

View File

@@ -30,18 +30,18 @@ import gpiod
from ....logging import get_logger
from .... import env
# =====
class Gpio:
def __init__(
self,
device_path: str,
reset_pin: int,
reset_inverted: bool,
reset_delay: float,
) -> None:
self.__device_path = device_path
self.__reset_pin = reset_pin
self.__reset_inverted = reset_inverted
self.__reset_delay = reset_delay
@@ -53,7 +53,7 @@ class Gpio:
if self.__reset_pin >= 0:
assert self.__chip is None
assert self.__reset_line is None
self.__chip = gpiod.Chip(env.GPIO_DEVICE_PATH)
self.__chip = gpiod.Chip(self.__device_path)
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)])

View File

@@ -44,8 +44,6 @@ from ...validators.basic import valid_int_f1
from ...validators.basic import valid_float_f01
from ...validators.hw import valid_gpio_pin_optional
from ... import env
from ._mcu import BasePhyConnection
from ._mcu import BasePhy
from ._mcu import BaseMcuHid
@@ -100,6 +98,7 @@ class _SpiPhyConnection(BasePhyConnection):
class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes
def __init__(
self,
gpio_device_path: str,
bus: int,
chip: int,
hw_cs: bool,
@@ -109,6 +108,7 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes
read_timeout: float,
) -> None:
self.__gpio_device_path = gpio_device_path
self.__bus = bus
self.__chip = chip
self.__hw_cs = hw_cs
@@ -145,7 +145,7 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes
@contextlib.contextmanager
def __sw_cs_connected(self) -> Generator[Optional[gpiod.Line], None, None]:
if self.__sw_cs_pin > 0:
with contextlib.closing(gpiod.Chip(env.GPIO_DEVICE_PATH)) as chip:
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
@@ -157,6 +157,7 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes
class Plugin(BaseMcuHid):
def __init__(self, **kwargs: Any) -> None:
phy_kwargs: Dict = {key: kwargs.pop(key) for key in self.__get_phy_options()}
phy_kwargs["gpio_device_path"] = kwargs["gpio_device_path"]
super().__init__(phy=_SpiPhy(**phy_kwargs), **kwargs)
@classmethod

View File

@@ -62,6 +62,7 @@ from .drive import DeviceInfo
class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
def __init__( # pylint: disable=super-init-not-called
self,
gpio_device_path: str,
target_pin: int,
reset_pin: int,
@@ -75,7 +76,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
self.__init_delay = init_delay
self.__init_retries = init_retries
self.__gpio = Gpio(target_pin, reset_pin, reset_delay)
self.__gpio = Gpio(gpio_device_path, target_pin, reset_pin, reset_delay)
self.__device_info: Optional[DeviceInfo] = None
self.__connected = False
@@ -97,8 +98,9 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
@classmethod
def get_plugin_options(cls) -> Dict:
return {
"target_pin": Option(-1, type=valid_gpio_pin),
"reset_pin": Option(-1, type=valid_gpio_pin),
"gpio_device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="gpio_device_path"),
"target_pin": Option(-1, type=valid_gpio_pin),
"reset_pin": Option(-1, type=valid_gpio_pin),
"device": Option("", type=valid_abs_path, unpack_as="device_path"),
"init_delay": Option(1.0, type=valid_float_f01),

View File

@@ -24,7 +24,6 @@ from typing import Optional
import gpiod
from .... import env
from .... import aiogp
@@ -32,11 +31,13 @@ from .... import aiogp
class Gpio:
def __init__(
self,
device_path: str,
target_pin: int,
reset_pin: int,
reset_delay: float,
) -> None:
self.__device_path = device_path
self.__target_pin = target_pin
self.__reset_pin = reset_pin
self.__reset_delay = reset_delay
@@ -50,7 +51,7 @@ class Gpio:
assert self.__target_line is None
assert self.__reset_line is None
self.__chip = gpiod.Chip(env.GPIO_DEVICE_PATH)
self.__chip = gpiod.Chip(self.__device_path)
self.__target_line = self.__chip.get_line(self.__target_pin)
self.__target_line.request("kvmd::msd::target", gpiod.LINE_REQ_DIR_OUT, default_vals=[0])

View File

@@ -25,10 +25,13 @@ from typing import Optional
import gpiod
from ... import env
from ... import aiotools
from ... import aiogp
from ...yamlconf import Option
from ...validators.os import valid_abs_path
from . import BaseUserGpioDriver
@@ -38,10 +41,14 @@ class Plugin(BaseUserGpioDriver):
self,
instance_name: str,
notifier: aiotools.AioNotifier,
device_path: str,
) -> None:
super().__init__(instance_name, notifier)
self.__device_path = device_path
self.__input_pins: Dict[int, aiogp.AioReaderPinParams] = {}
self.__output_pins: Dict[int, Optional[bool]] = {}
@@ -50,6 +57,12 @@ class Plugin(BaseUserGpioDriver):
self.__chip: Optional[gpiod.Chip] = None
self.__output_lines: Dict[int, gpiod.Line] = {}
@classmethod
def get_plugin_options(cls) -> Dict:
return {
"device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="device_path"),
}
def register_input(self, pin: int, debounce: float) -> None:
self.__input_pins[pin] = aiogp.AioReaderPinParams(False, debounce)
@@ -59,13 +72,13 @@ class Plugin(BaseUserGpioDriver):
def prepare(self) -> None:
assert self.__reader is None
self.__reader = aiogp.AioReader(
path=env.GPIO_DEVICE_PATH,
path=self.__device_path,
consumer="kvmd::gpio::inputs",
pins=self.__input_pins,
notifier=self._notifier,
)
self.__chip = gpiod.Chip(env.GPIO_DEVICE_PATH)
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)])