GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-01-29 00:51:53 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

another try

Browse Source
This commit is contained in:
Devaev Maxim
2020-11-10 13:38:55 +03:00
parent e07cdd60f3
commit a8a075c203
4 changed files with 93 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
kvmd/plugins/hid/spi.py
View File

@@ -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 {
"bus": Option(0, type=valid_int_f0),
"chip": Option(0, type=valid_int_f0),
"max_freq": Option(400000, 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),
**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),
"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(0.5, type=valid_float_f01),
}