diff options
Diffstat (limited to 'kvmd/plugins')
| -rw-r--r-- | kvmd/plugins/hid/spi.py | 57 |
1 files changed, 29 insertions, 28 deletions
diff --git a/kvmd/plugins/hid/spi.py b/kvmd/plugins/hid/spi.py index d71a4094..3dcd0dab 100644 --- a/kvmd/plugins/hid/spi.py +++ b/kvmd/plugins/hid/spi.py @@ -36,9 +36,9 @@ from ...logging import get_logger from ...yamlconf import Option +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_f0 from ...validators.basic import valid_float_f01 from ._mcu import BasePhyConnection @@ -52,26 +52,33 @@ class _SpiPhyConnection(BasePhyConnection): self, xfer: Callable[[bytes], bytes], read_timeout: float, - read_delay: float, ) -> None: self.__xfer = xfer self.__read_timeout = read_timeout - self.__read_delay = read_delay def send(self, request: bytes) -> bytes: assert len(request) == 8 + assert request[0] == 0x33 + + deadline_ts = time.time() + self.__read_timeout + dummy = b"\x00" * 8 + while time.time() < deadline_ts: + if bytes(self.__xfer(dummy)) == dummy: + break + else: + get_logger(0).error("SPI timeout reached while garbage reading") + return b"" + self.__xfer(request) response: List[int] = [] deadline_ts = time.time() + self.__read_timeout found = False while time.time() < deadline_ts: - if not found: - time.sleep(self.__read_delay) for byte in self.__xfer(b"\x00" * (4 - len(response))): if not found: - if byte == 0: + if byte != 0x33: continue found = True response.append(byte) @@ -90,18 +97,18 @@ class _SpiPhy(BasePhy): self, bus: int, chip: int, + cs: bool, max_freq: int, block_usec: int, read_timeout: float, - read_delay: float, ) -> None: self.__bus = bus self.__chip = chip + self.__cs = cs self.__max_freq = max_freq self.__block_usec = block_usec self.__read_timeout = read_timeout - self.__read_delay = read_delay def has_device(self) -> bool: return os.path.exists(f"/dev/spidev{self.__bus}.{self.__chip}") @@ -110,6 +117,7 @@ class _SpiPhy(BasePhy): def connected(self) -> Generator[_SpiPhyConnection, None, None]: # type: ignore with contextlib.closing(spidev.SpiDev(self.__bus, self.__chip)) as spi: spi.mode = 0 + spi.no_cs = (not self.__cs) spi.max_speed_hz = self.__max_freq def xfer(data: bytes) -> bytes: @@ -118,36 +126,29 @@ class _SpiPhy(BasePhy): yield _SpiPhyConnection( xfer=xfer, read_timeout=self.__read_timeout, - read_delay=self.__read_delay, ) # ===== class Plugin(BaseMcuHid): - def __init__( - self, - bus: int, - chip: int, - max_freq: int, - block_usec: int, - read_timeout: float, - read_delay: float, - **kwargs: Any, - ) -> None: - - super().__init__( - phy=_SpiPhy(bus, chip, max_freq, block_usec, read_timeout, read_delay), - **kwargs, - ) + def __init__(self, **kwargs: Any) -> None: + phy_kwargs: Dict = {key: kwargs.pop(key) for key in self.__get_phy_options()} + super().__init__(phy=_SpiPhy(**phy_kwargs), **kwargs) @classmethod def get_plugin_options(cls) -> Dict: return { + **cls.__get_phy_options(), + **BaseMcuHid.get_plugin_options(), + } + + @classmethod + def __get_phy_options(cls) -> Dict: + return { "bus": Option(0, type=valid_int_f0), "chip": Option(0, type=valid_int_f0), - "max_freq": Option(400000, type=valid_int_f1), + "cs": Option(False, type=valid_bool), + "max_freq": Option(200000, type=valid_int_f1), "block_usec": Option(1, type=valid_int_f0), - "read_timeout": Option(2.0, type=valid_float_f01), - "read_delay": Option(0.001, type=valid_float_f0), - **BaseMcuHid.get_plugin_options(), + "read_timeout": Option(0.5, type=valid_float_f01), } |