summaryrefslogtreecommitdiff
path: root/kvmd/plugins/hid
diff options
context:
space:
mode:
Diffstat (limited to 'kvmd/plugins/hid')
-rw-r--r--kvmd/plugins/hid/__init__.py12
-rw-r--r--kvmd/plugins/hid/_mcu/__init__.py251
-rw-r--r--kvmd/plugins/hid/_mcu/proto.py202
-rw-r--r--kvmd/plugins/hid/bt/__init__.py9
-rw-r--r--kvmd/plugins/hid/otg/__init__.py6
-rw-r--r--kvmd/plugins/hid/serial.py5
-rw-r--r--kvmd/plugins/hid/spi.py10
7 files changed, 321 insertions, 174 deletions
diff --git a/kvmd/plugins/hid/__init__.py b/kvmd/plugins/hid/__init__.py
index f57379aa..8baceb28 100644
--- a/kvmd/plugins/hid/__init__.py
+++ b/kvmd/plugins/hid/__init__.py
@@ -57,14 +57,22 @@ class BaseHid(BasePlugin):
raise NotImplementedError
def send_mouse_move_event(self, to_x: int, to_y: int) -> None:
- raise NotImplementedError
+ _ = to_x
+ _ = to_y
def send_mouse_relative_event(self, delta_x: int, delta_y: int) -> None:
- raise NotImplementedError
+ _ = delta_x
+ _ = delta_y
def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None:
raise NotImplementedError
+ def set_keyboard_output(self, output: str) -> None:
+ _ = output
+
+ def set_mouse_output(self, output: str) -> None:
+ _ = output
+
def clear_events(self) -> None:
raise NotImplementedError
diff --git a/kvmd/plugins/hid/_mcu/__init__.py b/kvmd/plugins/hid/_mcu/__init__.py
index b46c876d..6e0f902e 100644
--- a/kvmd/plugins/hid/_mcu/__init__.py
+++ b/kvmd/plugins/hid/_mcu/__init__.py
@@ -22,10 +22,8 @@
import os
import multiprocessing
-import dataclasses
import contextlib
import queue
-import struct
import time
from typing import Tuple
@@ -37,8 +35,6 @@ from typing import AsyncGenerator
from ....logging import get_logger
-from ....keyboard.mappings import KEYMAP
-
from .... import tools
from .... import aiotools
from .... import aiomulti
@@ -56,6 +52,22 @@ from .. import BaseHid
from .gpio import Gpio
+from .proto import REQUEST_PING
+from .proto import REQUEST_REPEAT
+from .proto import RESPONSE_LEGACY_OK
+from .proto import BaseEvent
+from .proto import SetKeyboardOutputEvent
+from .proto import SetMouseOutputEvent
+from .proto import ClearEvent
+from .proto import KeyEvent
+from .proto import MouseButtonEvent
+from .proto import MouseMoveEvent
+from .proto import MouseRelativeEvent
+from .proto import MouseWheelEvent
+from .proto import get_active_keyboard
+from .proto import get_active_mouse
+from .proto import check_response
+
# =====
class _RequestError(Exception):
@@ -73,84 +85,6 @@ class _TempRequestError(_RequestError):
# =====
-class _BaseEvent:
- def make_command(self) -> bytes:
- raise NotImplementedError
-
-
-class _ClearEvent(_BaseEvent):
- def make_command(self) -> bytes:
- return b"\x10\x00\x00\x00\x00"
-
-
[email protected](frozen=True)
-class _KeyEvent(_BaseEvent):
- name: str
- state: bool
-
- def __post_init__(self) -> None:
- assert self.name in KEYMAP
-
- def make_command(self) -> bytes:
- code = KEYMAP[self.name].mcu.code
- return struct.pack(">BBBxx", 0x11, code, int(self.state))
-
-
[email protected](frozen=True)
-class _MouseButtonEvent(_BaseEvent):
- name: str
- state: bool
-
- def __post_init__(self) -> None:
- assert self.name in ["left", "right", "middle", "up", "down"]
-
- def make_command(self) -> bytes:
- (code, state_pressed, is_main) = {
- "left": (0b10000000, 0b00001000, True),
- "right": (0b01000000, 0b00000100, True),
- "middle": (0b00100000, 0b00000010, True),
- "up": (0b10000000, 0b00001000, False), # Back
- "down": (0b01000000, 0b00000100, False), # Forward
- }[self.name]
- if self.state:
- code |= state_pressed
- if is_main:
- main_code = code
- extra_code = 0
- else:
- main_code = 0
- extra_code = code
- return struct.pack(">BBBxx", 0x13, main_code, extra_code)
-
-
[email protected](frozen=True)
-class _MouseMoveEvent(_BaseEvent):
- to_x: int
- to_y: int
-
- def __post_init__(self) -> None:
- assert -32768 <= self.to_x <= 32767
- assert -32768 <= self.to_y <= 32767
-
- def make_command(self) -> bytes:
- return struct.pack(">Bhh", 0x12, self.to_x, self.to_y)
-
-
[email protected](frozen=True)
-class _MouseWheelEvent(_BaseEvent):
- delta_x: int
- delta_y: int
-
- def __post_init__(self) -> None:
- assert -127 <= self.delta_x <= 127
- assert -127 <= self.delta_y <= 127
-
- def make_command(self) -> bytes:
- # Горизонтальная прокрутка пока не поддерживается
- return struct.pack(">Bxbxx", 0x14, self.delta_y)
-
-
-# =====
class BasePhyConnection:
def send(self, request: bytes) -> bytes:
raise NotImplementedError
@@ -192,16 +126,13 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
self.__phy = phy
self.__gpio = Gpio(reset_pin, reset_inverted, reset_delay)
- self.__events_queue: "multiprocessing.Queue[_BaseEvent]" = multiprocessing.Queue()
+ self.__events_queue: "multiprocessing.Queue[BaseEvent]" = multiprocessing.Queue()
self.__notifier = aiomulti.AioProcessNotifier()
self.__state_flags = aiomulti.AioSharedFlags({
- "keyboard_online": True,
- "mouse_online": True,
- "caps": False,
- "scroll": False,
- "num": False,
- }, self.__notifier)
+ "online": 0,
+ "status": 0,
+ }, self.__notifier, type=int)
self.__stop_event = multiprocessing.Event()
@@ -226,19 +157,51 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
async def get_state(self) -> Dict:
state = await self.__state_flags.get()
+ online = bool(state["online"])
+ pong = (state["status"] >> 16) & 0xFF
+ outputs = (state["status"] >> 8) & 0xFF
+ features = state["status"] & 0xFF
+
+ absolute = True
+ active_mouse = get_active_mouse(outputs)
+ if online and active_mouse in ["usb_rel", "ps2"]:
+ absolute = False
+
+ keyboard_outputs: Dict = {"available": {}, "active": ""}
+ mouse_outputs: Dict = {"available": {}, "active": ""}
+
+ if outputs & 0b10000000: # Dynamic
+ if features & 0b00000001: # USB
+ keyboard_outputs["available"]["usb"] = {"name": "USB"}
+ mouse_outputs["available"]["usb"] = {"name": "USB", "absolute": True}
+ mouse_outputs["available"]["usb_rel"] = {"name": "USB Relative", "absolute": False}
+
+ if features & 0b00000010: # PS/2
+ keyboard_outputs["available"]["ps2"] = {"name": "PS/2"}
+ mouse_outputs["available"]["ps2"] = {"name": "PS/2"}
+
+ active_keyboard = get_active_keyboard(outputs)
+ if active_keyboard in keyboard_outputs["available"]:
+ keyboard_outputs["active"] = active_keyboard
+
+ if active_mouse in mouse_outputs["available"]:
+ mouse_outputs["active"] = active_mouse
+
return {
- "online": (state["keyboard_online"] and state["mouse_online"]),
+ "online": online,
"keyboard": {
- "online": state["keyboard_online"],
+ "online": (online and not (pong & 0b00001000)),
"leds": {
- "caps": state["caps"],
- "scroll": state["scroll"],
- "num": state["num"],
+ "caps": bool(pong & 0b00000001),
+ "scroll": bool(pong & 0b00000010),
+ "num": bool(pong & 0b00000100),
},
+ "outputs": keyboard_outputs,
},
"mouse": {
- "online": state["mouse_online"],
- "absolute": True,
+ "online": (online and not (pong & 0b00010000)),
+ "absolute": absolute,
+ "outputs": mouse_outputs,
},
}
@@ -268,7 +231,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
get_logger().info("Clearing HID events ...")
try:
with self.__phy.connected() as conn:
- self.__process_command(conn, b"\x10\x00\x00\x00\x00")
+ self.__process_request(conn, ClearEvent().make_request())
except Exception:
logger.exception("Can't clear HID events")
finally:
@@ -278,30 +241,36 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
def send_key_events(self, keys: Iterable[Tuple[str, bool]]) -> None:
for (key, state) in keys:
- self.__queue_event(_KeyEvent(key, state))
+ self.__queue_event(KeyEvent(key, state))
def send_mouse_button_event(self, button: str, state: bool) -> None:
- self.__queue_event(_MouseButtonEvent(button, state))
+ self.__queue_event(MouseButtonEvent(button, state))
def send_mouse_move_event(self, to_x: int, to_y: int) -> None:
- self.__queue_event(_MouseMoveEvent(to_x, to_y))
+ self.__queue_event(MouseMoveEvent(to_x, to_y))
def send_mouse_relative_event(self, delta_x: int, delta_y: int) -> None:
- _ = delta_x # No relative events yet
- _ = delta_y
+ self.__queue_event(MouseRelativeEvent(delta_x, delta_y))
def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None:
- self.__queue_event(_MouseWheelEvent(delta_x, delta_y))
+ self.__queue_event(MouseWheelEvent(delta_x, delta_y))
+
+ def set_keyboard_output(self, output: str) -> None:
+ self.__queue_event(SetKeyboardOutputEvent(output), clear=True)
+
+ def set_mouse_output(self, output: str) -> None:
+ self.__queue_event(SetMouseOutputEvent(output), clear=True)
def clear_events(self) -> None:
- # FIXME: Если очистка производится со стороны процесса хида, то возможна гонка между
- # очисткой и добавлением события _ClearEvent. Неприятно, но не смертельно.
- # Починить блокировкой после перехода на асинхронные очереди.
- tools.clear_queue(self.__events_queue)
- self.__queue_event(_ClearEvent())
+ self.__queue_event(ClearEvent(), clear=True)
- def __queue_event(self, event: _BaseEvent) -> None:
+ def __queue_event(self, event: BaseEvent, clear: bool=False) -> None:
if not self.__stop_event.is_set():
+ if clear:
+ # FIXME: Если очистка производится со стороны процесса хида, то возможна гонка между
+ # очисткой и добавлением нового события. Неприятно, но не смертельно.
+ # Починить блокировкой после перехода на асинхронные очереди.
+ tools.clear_queue(self.__events_queue)
self.__events_queue.put_nowait(event)
def run(self) -> None: # pylint: disable=too-many-branches
@@ -319,9 +288,9 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
try:
event = self.__events_queue.get(timeout=0.1)
except queue.Empty:
- self.__process_command(conn, b"\x01\x00\x00\x00\x00") # Ping
+ self.__process_request(conn, REQUEST_PING)
else:
- if not self.__process_command(conn, event.make_command()):
+ if not self.__process_request(conn, event.make_request()):
self.clear_events()
else:
logger.error("Missing HID device")
@@ -331,9 +300,6 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
logger.exception("Unexpected HID error")
time.sleep(1)
- def __process_command(self, conn: BasePhyConnection, command: bytes) -> bool:
- return self.__process_request(conn, self.__make_request(command))
-
def __process_request(self, conn: BasePhyConnection, request: bytes) -> bool: # pylint: disable=too-many-branches
logger = get_logger()
error_messages: List[str] = []
@@ -344,15 +310,14 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
error_retval = False
while common_retries and read_retries:
- response = self.__send_request(conn, request)
+ response = (RESPONSE_LEGACY_OK if self.__noop else conn.send(request))
try:
if len(response) < 4:
read_retries -= 1
raise _TempRequestError(f"No response from HID: request={request!r}")
- assert len(response) == 4, response
- if self.__make_crc16(response[-4:-2]) != struct.unpack(">H", response[-2:])[0]:
- request = self.__make_request(b"\x02\x00\x00\x00\x00") # Repeat an answer
+ if not check_response(response):
+ request = REQUEST_REPEAT
raise _TempRequestError("Invalid response CRC; requesting response again ...")
code = response[1]
@@ -368,9 +333,9 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
self.__set_state_online(True)
return True
elif code & 0x80: # Pong/Done with state
- self.__set_state_code(code)
+ self.__set_state_pong(response)
return True
- raise _TempRequestError(f"Invalid response from HID: request={request!r}; code=0x{code:02X}")
+ raise _TempRequestError(f"Invalid response from HID: request={request!r}, response=0x{response!r}")
except _RequestError as err:
common_retries -= 1
@@ -401,42 +366,10 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many-
return error_retval
def __set_state_online(self, online: bool) -> None:
- self.__state_flags.update(
- keyboard_online=online,
- mouse_online=online,
- )
-
- def __set_state_code(self, code: int) -> None:
- self.__state_flags.update(
- keyboard_online=(not (code & 0b00001000)),
- mouse_online=(not (code & 0b00010000)),
- caps=bool(code & 0b00000001),
- scroll=bool(code & 0b00000010),
- num=bool(code & 0b00000100),
- )
-
- def __send_request(self, conn: BasePhyConnection, request: bytes) -> bytes:
- if not self.__noop:
- response = conn.send(request)
- else:
- response = b"\x33\x20" # Magic + OK
- response += struct.pack(">H", self.__make_crc16(response))
- return response
-
- def __make_request(self, command: bytes) -> bytes:
- request = b"\x33" + command
- request += struct.pack(">H", self.__make_crc16(request))
- assert len(request) == 8, (request, command)
- return request
-
- def __make_crc16(self, data: bytes) -> int:
- crc = 0xFFFF
- for byte in data:
- crc = crc ^ byte
- for _ in range(8):
- if crc & 0x0001 == 0:
- crc = crc >> 1
- else:
- crc = crc >> 1
- crc = crc ^ 0xA001
- return crc
+ self.__state_flags.update(online=int(online))
+
+ def __set_state_pong(self, response: bytes) -> None:
+ status = response[1] << 16
+ if len(response) > 4:
+ status |= (response[2] << 8) | response[3]
+ self.__state_flags.update(online=1, status=status)
diff --git a/kvmd/plugins/hid/_mcu/proto.py b/kvmd/plugins/hid/_mcu/proto.py
new file mode 100644
index 00000000..7cfc3fd7
--- /dev/null
+++ b/kvmd/plugins/hid/_mcu/proto.py
@@ -0,0 +1,202 @@
+# ========================================================================== #
+# #
+# KVMD - The main Pi-KVM daemon. #
+# #
+# Copyright (C) 2018 Maxim Devaev <[email protected]> #
+# #
+# 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/>. #
+# #
+# ========================================================================== #
+
+
+import dataclasses
+import struct
+
+from ....keyboard.mappings import KEYMAP
+
+from .... import tools
+
+
+# =====
+class BaseEvent:
+ def make_request(self) -> bytes:
+ raise NotImplementedError
+
+
+# =====
+_KEYBOARD_NAMES_TO_CODES = {
+ "usb": 0b00000001,
+ "ps2": 0b00000011,
+}
+_KEYBOARD_CODES_TO_NAMES = tools.swapped_kvs(_KEYBOARD_NAMES_TO_CODES)
+
+
+def get_active_keyboard(outputs: int) -> str:
+ return _KEYBOARD_CODES_TO_NAMES.get(outputs & 0b00000111, "")
+
+
[email protected](frozen=True)
+class SetKeyboardOutputEvent(BaseEvent):
+ keyboard: str
+
+ def __post_init__(self) -> None:
+ assert not self.keyboard or self.keyboard in _KEYBOARD_NAMES_TO_CODES
+
+ def make_request(self) -> bytes:
+ code = _KEYBOARD_NAMES_TO_CODES.get(self.keyboard, 0)
+ return _make_request(struct.pack(">BBxxx", 0x03, code))
+
+
+# =====
+_MOUSE_NAMES_TO_CODES = {
+ "usb": 0b00001000,
+ "usb_rel": 0b00010000,
+ "ps2": 0b00011000,
+}
+_MOUSE_CODES_TO_NAMES = tools.swapped_kvs(_MOUSE_NAMES_TO_CODES)
+
+
+def get_active_mouse(outputs: int) -> str:
+ return _MOUSE_CODES_TO_NAMES.get(outputs & 0b00111000, "")
+
+
[email protected](frozen=True)
+class SetMouseOutputEvent(BaseEvent):
+ mouse: str
+
+ def __post_init__(self) -> None:
+ assert not self.mouse or self.mouse in _MOUSE_NAMES_TO_CODES
+
+ def make_request(self) -> bytes:
+ return _make_request(struct.pack(">BBxxx", 0x04, _MOUSE_NAMES_TO_CODES.get(self.mouse, 0)))
+
+
+# =====
+class ClearEvent(BaseEvent):
+ def make_request(self) -> bytes:
+ return _make_request(b"\x10\x00\x00\x00\x00")
+
+
[email protected](frozen=True)
+class KeyEvent(BaseEvent):
+ name: str
+ state: bool
+
+ def __post_init__(self) -> None:
+ assert self.name in KEYMAP
+
+ def make_request(self) -> bytes:
+ code = KEYMAP[self.name].mcu.code
+ return _make_request(struct.pack(">BBBxx", 0x11, code, int(self.state)))
+
+
[email protected](frozen=True)
+class MouseButtonEvent(BaseEvent):
+ name: str
+ state: bool
+
+ def __post_init__(self) -> None:
+ assert self.name in ["left", "right", "middle", "up", "down"]
+
+ def make_request(self) -> bytes:
+ (code, state_pressed, is_main) = {
+ "left": (0b10000000, 0b00001000, True),
+ "right": (0b01000000, 0b00000100, True),
+ "middle": (0b00100000, 0b00000010, True),
+ "up": (0b10000000, 0b00001000, False), # Back
+ "down": (0b01000000, 0b00000100, False), # Forward
+ }[self.name]
+ if self.state:
+ code |= state_pressed
+ if is_main:
+ main_code = code
+ extra_code = 0
+ else:
+ main_code = 0
+ extra_code = code
+ return _make_request(struct.pack(">BBBxx", 0x13, main_code, extra_code))
+
+
[email protected](frozen=True)
+class MouseMoveEvent(BaseEvent):
+ to_x: int
+ to_y: int
+
+ def __post_init__(self) -> None:
+ assert -32768 <= self.to_x <= 32767
+ assert -32768 <= self.to_y <= 32767
+
+ def make_request(self) -> bytes:
+ return _make_request(struct.pack(">Bhh", 0x12, self.to_x, self.to_y))
+
+
[email protected](frozen=True)
+class MouseRelativeEvent(BaseEvent):
+ delta_x: int
+ delta_y: int
+
+ def __post_init__(self) -> None:
+ assert -127 <= self.delta_x <= 127
+ assert -127 <= self.delta_y <= 127
+
+ def make_request(self) -> bytes:
+ return _make_request(struct.pack(">Bbbxx", 0x15, self.delta_x, self.delta_y))
+
+
[email protected](frozen=True)
+class MouseWheelEvent(BaseEvent):
+ delta_x: int
+ delta_y: int
+
+ def __post_init__(self) -> None:
+ assert -127 <= self.delta_x <= 127
+ assert -127 <= self.delta_y <= 127
+
+ def make_request(self) -> bytes:
+ # Горизонтальная прокрутка пока не поддерживается
+ return _make_request(struct.pack(">Bxbxx", 0x14, self.delta_y))
+
+
+# =====
+def check_response(response: bytes) -> bool:
+ assert len(response) in (4, 8), response
+ return (_make_crc16(response[:-2]) == struct.unpack(">H", response[-2:])[0])
+
+
+def _make_request(command: bytes) -> bytes:
+ assert len(command) == 5, command
+ request = b"\x33" + command
+ request += struct.pack(">H", _make_crc16(request))
+ assert len(request) == 8, request
+ return request
+
+
+def _make_crc16(data: bytes) -> int:
+ crc = 0xFFFF
+ for byte in data:
+ crc = crc ^ byte
+ for _ in range(8):
+ if crc & 0x0001 == 0:
+ crc = crc >> 1
+ else:
+ crc = crc >> 1
+ crc = crc ^ 0xA001
+ return crc
+
+
+# =====
+REQUEST_PING = _make_request(b"\x01\x00\x00\x00\x00")
+REQUEST_REPEAT = _make_request(b"\x02\x00\x00\x00\x00")
+
+RESPONSE_LEGACY_OK = b"\x33\x20" + struct.pack(">H", _make_crc16(b"\x33\x20"))
diff --git a/kvmd/plugins/hid/bt/__init__.py b/kvmd/plugins/hid/bt/__init__.py
index 6c8dc46b..c13e2a34 100644
--- a/kvmd/plugins/hid/bt/__init__.py
+++ b/kvmd/plugins/hid/bt/__init__.py
@@ -131,8 +131,9 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes
async def get_state(self) -> Dict:
state = await self.__server.get_state()
+ outputs: Dict = {"available": {}, "active": ""}
return {
- "online": state["online"],
+ "online": True,
"keyboard": {
"online": state["online"],
"leds": {
@@ -140,10 +141,12 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes
"scroll": state["scroll"],
"num": state["num"],
},
+ "outputs": outputs,
},
"mouse": {
"online": state["online"],
"absolute": False,
+ "outputs": outputs,
},
}
@@ -178,10 +181,6 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes
def send_mouse_button_event(self, button: str, state: bool) -> None:
self.__server.queue_event(MouseButtonEvent(button, state))
- def send_mouse_move_event(self, to_x: int, to_y: int) -> None:
- _ = to_x # No absolute events
- _ = to_y
-
def send_mouse_relative_event(self, delta_x: int, delta_y: int) -> None:
self.__server.queue_event(MouseRelativeEvent(delta_x, delta_y))
diff --git a/kvmd/plugins/hid/otg/__init__.py b/kvmd/plugins/hid/otg/__init__.py
index aa3d79b2..e685226b 100644
--- a/kvmd/plugins/hid/otg/__init__.py
+++ b/kvmd/plugins/hid/otg/__init__.py
@@ -89,8 +89,9 @@ class Plugin(BaseHid):
async def get_state(self) -> Dict:
keyboard_state = await self.__keyboard_proc.get_state()
mouse_state = await self.__mouse_proc.get_state()
+ outputs: Dict = {"available": {}, "active": ""}
return {
- "online": (keyboard_state["online"] and mouse_state["online"]),
+ "online": True,
"keyboard": {
"online": keyboard_state["online"],
"leds": {
@@ -98,8 +99,9 @@ class Plugin(BaseHid):
"scroll": keyboard_state["scroll"],
"num": keyboard_state["num"],
},
+ "outputs": outputs,
},
- "mouse": mouse_state,
+ "mouse": {**mouse_state, "outputs": outputs},
}
async def poll_state(self) -> AsyncGenerator[Dict, None]:
diff --git a/kvmd/plugins/hid/serial.py b/kvmd/plugins/hid/serial.py
index dee992b6..da61f2c6 100644
--- a/kvmd/plugins/hid/serial.py
+++ b/kvmd/plugins/hid/serial.py
@@ -51,7 +51,10 @@ class _SerialPhyConnection(BasePhyConnection):
if self.__tty.in_waiting:
self.__tty.read_all()
assert self.__tty.write(request) == 8
- return self.__tty.read(4)
+ data = self.__tty.read(4)
+ if data[0] == 0x34: # New response protocol
+ data += self.__tty.read(4)
+ return data
class _SerialPhy(BasePhy):
diff --git a/kvmd/plugins/hid/spi.py b/kvmd/plugins/hid/spi.py
index 7dadac0e..785ca779 100644
--- a/kvmd/plugins/hid/spi.py
+++ b/kvmd/plugins/hid/spi.py
@@ -67,7 +67,7 @@ class _SpiPhyConnection(BasePhyConnection):
assert request[0] == 0x33
deadline_ts = time.monotonic() + self.__read_timeout
- dummy = b"\x00" * 8
+ dummy = b"\x00" * 10
while time.monotonic() < deadline_ts:
if bytes(self.__xfer(dummy)) == dummy:
break
@@ -81,15 +81,15 @@ class _SpiPhyConnection(BasePhyConnection):
deadline_ts = time.monotonic() + self.__read_timeout
found = False
while time.monotonic() < deadline_ts:
- for byte in self.__xfer(b"\x00" * (5 - len(response))):
+ for byte in self.__xfer(b"\x00" * (9 - len(response))):
if not found:
- if byte != 0x33:
+ if byte == 0:
continue
found = True
response.append(byte)
- if len(response) == 4:
+ if len(response) == 8:
break
- if len(response) == 4:
+ if len(response) == 8:
break
else:
get_logger(0).error("SPI timeout reached while responce waiting")
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-24 17:29:01 +0000