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

rp2040 hid

Browse Source
This commit is contained in:
Maxim Devaev
2023-08-03 05:47:27 +03:00
parent 1a8f98a64f
commit c96057772c
29 changed files with 1933 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
kvmd/plugins/hid/_mcu/__init__.py
View File

@@ -108,6 +108,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
reset_pin: int,
reset_inverted: bool,
reset_delay: float,
reset_self: bool,
read_retries: int,
common_retries: int,
@@ -126,6 +127,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
self.__phy = phy
self.__gpio = Gpio(gpio_device_path, reset_pin, reset_inverted, reset_delay)
self.__reset_self = reset_self
self.__reset_required_event = multiprocessing.Event()
self.__events_queue: "multiprocessing.Queue[BaseEvent]" = multiprocessing.Queue()
@@ -146,6 +148,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
"reset_pin": Option(4, type=valid_gpio_pin_optional),
"reset_inverted": Option(False, type=valid_bool),
"reset_delay": Option(0.1, type=valid_float_f01),
"reset_self": Option(False, type=valid_bool),
"read_retries": Option(5, type=valid_int_f1),
"common_retries": Option(5, type=valid_int_f1),
@@ -418,4 +421,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
reset_required = (1 if response[1] & 0b01000000 else 0)
self.__state_flags.update(online=1, busy=reset_required, status=status)
if reset_required:
self.__reset_required_event.set()
if self.__reset_self:
time.sleep(1) # Pico перезагружается сам вскоре после ответа
else:
self.__reset_required_event.set()

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

@@ -100,6 +100,7 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes
chip: int,
hw_cs: bool,
sw_cs_pin: int,
sw_cs_per_byte: bool,
max_freq: int,
block_usec: int,
read_timeout: float,
@@ -110,6 +111,7 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes
self.__chip = chip
self.__hw_cs = hw_cs
self.__sw_cs_pin = sw_cs_pin
self.__sw_cs_per_byte = sw_cs_per_byte
self.__max_freq = max_freq
self.__block_usec = block_usec
self.__read_timeout = read_timeout
@@ -125,7 +127,7 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes
spi.no_cs = (not self.__hw_cs)
spi.max_speed_hz = self.__max_freq
def xfer(data: bytes) -> bytes:
def inner_xfer(data: bytes) -> bytes:
try:
if sw_cs_line is not None:
sw_cs_line.set_value(0)
@@ -134,6 +136,17 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes
if sw_cs_line is not None:
sw_cs_line.set_value(1)
if self.__sw_cs_per_byte:
# Режим для Pico, когда CS должен взводиться для отдельных байтов
def xfer(data: bytes) -> bytes:
got: list[int] = []
for byte in data:
got.extend(inner_xfer(byte.to_bytes(1, "big")))
return bytes(got)
else:
# Режим для Arduino, когда CS взводится для целого блока данных
xfer = inner_xfer
yield _SpiPhyConnection(
xfer=xfer,
read_timeout=self.__read_timeout,
@@ -167,11 +180,12 @@ class Plugin(BaseMcuHid):
@classmethod
def __get_phy_options(cls) -> dict:
return {
"bus": Option(-1, type=valid_int_f0),
"chip": Option(-1, type=valid_int_f0),
"hw_cs": Option(False, type=valid_bool),
"sw_cs_pin": Option(-1, type=valid_gpio_pin_optional),
"max_freq": Option(100000, type=valid_int_f1),
"block_usec": Option(1, type=valid_int_f0),
"read_timeout": Option(0.5, type=valid_float_f01),
"bus": Option(-1, type=valid_int_f0),
"chip": Option(-1, type=valid_int_f0),
"hw_cs": Option(False, type=valid_bool),
"sw_cs_pin": Option(-1, type=valid_gpio_pin_optional),
"sw_cs_per_byte": Option(False, type=valid_bool),
"max_freq": Option(100000, type=valid_int_f1),
"block_usec": Option(1, type=valid_int_f0),
"read_timeout": Option(0.5, type=valid_float_f01),
}