refactoring

2 files changed, 182 insertions, 144 deletions

diff --git a/kvmd/plugins/hid/_mcu/__init__.py b/kvmd/plugins/hid/_mcu/__init__.py
index 678b4887..62c5870a 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,18 @@ 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 ClearEvent
+from .proto import KeyEvent
+from .proto import MouseButtonEvent
+from .proto import MouseMoveEvent
+from .proto import MouseRelativeEvent
+from .proto import MouseWheelEvent
+from .proto import check_response
+
 
 # =====
 class _RequestError(Exception):
@@ -73,97 +81,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 _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_command(self) -> bytes:
-        return 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_command(self) -> bytes:
-        # Горизонтальная прокрутка пока не поддерживается
-        return struct.pack(">Bxbxx", 0x14, self.delta_y)
-
-
-# =====
 class BasePhyConnection:
     def send(self, request: bytes) -> bytes:
         raise NotImplementedError
@@ -205,7 +122,7 @@ 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({
@@ -310,7 +227,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:
@@ -320,28 +237,28 @@ 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:
-        self.__queue_event(_MouseRelativeEvent(delta_x, 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 clear_events(self) -> None:
         # FIXME: Если очистка производится со стороны процесса хида, то возможна гонка между
         # очисткой и добавлением события _ClearEvent. Неприятно, но не смертельно.
         # Починить блокировкой после перехода на асинхронные очереди.
         tools.clear_queue(self.__events_queue)
-        self.__queue_event(_ClearEvent())
+        self.__queue_event(ClearEvent())
 
-    def __queue_event(self, event: _BaseEvent) -> None:
+    def __queue_event(self, event: BaseEvent) -> None:
         if not self.__stop_event.is_set():
             self.__events_queue.put_nowait(event)
 
@@ -360,9 +277,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")
@@ -372,9 +289,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] = []
@@ -385,15 +299,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) in (4, 8), response
-                if self.__make_crc16(response[:-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]
@@ -411,7 +324,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process):  # pylint: disable=too-many-
                 elif code & 0x80:  # Pong/Done with state
                     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
@@ -444,34 +357,8 @@ class BaseMcuHid(BaseHid, multiprocessing.Process):  # pylint: disable=too-many-
     def __set_state_online(self, online: bool) -> None:
         self.__state_flags.update(online=int(online))
 
-    def __set_state_pong(self, data: bytes) -> None:
-        status = data[1] << 16
-        if len(data) > 4:
-            status |= (data[2] << 8) | data[3]
+    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)
-
-    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
diff --git a/kvmd/plugins/hid/_mcu/proto.py b/kvmd/plugins/hid/_mcu/proto.py
new file mode 100644
index 00000000..fa76932a
--- /dev/null
+++ b/kvmd/plugins/hid/_mcu/proto.py
@@ -0,0 +1,151 @@
+# ========================================================================== #
+#                                                                            #
+#    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
+
+
+# =====
+class BaseEvent:
+    def make_request(self) -> bytes:
+        raise NotImplementedError
+
+
+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"))