removed UsbDeviceController() class

kvmd/plugins/hid/otg/__init__.py

@@ -27,6 +27,8 @@ from typing import AsyncGenerator
 from typing import Optional
 from typing import Any
 
+from ....logging import get_logger
+
 from .... import aiomulti
 from .... import usb
 
@@ -54,16 +56,13 @@ class Plugin(BaseHid):  # pylint: disable=too-many-instance-attributes
         udc: str,  # XXX: Not from options, see /kvmd/apps/kvmd/__init__.py for details
     ) -> None:
 
+        self.__udc = udc
+
         self.__notifier = aiomulti.AioProcessNotifier()
 
-        self.__udc = usb.UsbDeviceController(udc)
-
         win98_fix = mouse.pop("absolute_win98_fix")
-        common = {
-            "udc": self.__udc,
-            "noop": noop,
-            "notifier": self.__notifier,
-        }
+        common = {"notifier": self.__notifier, "noop": noop}
+
         self.__keyboard_proc = KeyboardProcess(**common, **keyboard)
         self.__mouse_current = self.__mouse_proc = MouseProcess(**common, **mouse)
 
@@ -115,11 +114,12 @@ class Plugin(BaseHid):  # pylint: disable=too-many-instance-attributes
         }
 
     def sysprep(self) -> None:
-        self.__udc.find()
-        self.__keyboard_proc.start()
-        self.__mouse_proc.start()
+        udc = usb.find_udc(self.__udc)
+        get_logger().info("Using UDC %s", udc)
+        self.__keyboard_proc.start(udc)
+        self.__mouse_proc.start(udc)
         if self.__mouse_alt_proc:
-            self.__mouse_alt_proc.start()
+            self.__mouse_alt_proc.start(udc)
 
     async def get_state(self) -> Dict:
         keyboard_state = await self.__keyboard_proc.get_state()

kvmd/plugins/hid/otg/device.py

@@ -32,10 +32,10 @@ from typing import Generator
 
 from ....logging import get_logger
 
+from .... import env
 from .... import tools
 from .... import aiomulti
 from .... import aioproc
-from .... import usb
 
 from .events import BaseEvent
 
@@ -49,8 +49,6 @@ class BaseDeviceProcess(multiprocessing.Process):  # pylint: disable=too-many-in
         initial_state: Dict,
         notifier: aiomulti.AioProcessNotifier,
 
-        udc: usb.UsbDeviceController,
-
         device_path: str,
         select_timeout: float,
         queue_timeout: float,
@@ -63,19 +61,22 @@ class BaseDeviceProcess(multiprocessing.Process):  # pylint: disable=too-many-in
         self.__name = name
         self.__read_size = read_size
 
-        self.__udc = udc
-
         self.__device_path = device_path
         self.__select_timeout = select_timeout
         self.__queue_timeout = queue_timeout
         self.__write_retries = write_retries
         self.__noop = noop
 
+        self.__udc_state_path = ""
         self.__fd = -1
         self.__events_queue: "multiprocessing.Queue[BaseEvent]" = multiprocessing.Queue()
         self.__state_flags = aiomulti.AioSharedFlags({"online": True, **initial_state}, notifier)
         self.__stop_event = multiprocessing.Event()
 
+    def start(self, udc: str) -> None:  # type: ignore  # pylint: disable=arguments-differ
+        self.__udc_state_path = os.path.join(f"{env.SYSFS_PREFIX}/sys/class/udc", udc, "state")
+        super().start()
+
     def run(self) -> None:  # pylint: disable=too-many-branches
         logger = aioproc.settle(f"HID-{self.__name}", f"hid-{self.__name}")
         report = b""
@@ -95,7 +96,7 @@ class BaseDeviceProcess(multiprocessing.Process):  # pylint: disable=too-many-in
                         #    - https://github.com/raspberrypi/linux/pull/3151
                         # Так что нам нужно проверять состояние контроллера, чтобы не спамить
                         # в устройство и отслеживать его состояние.
-                        if not self.__udc.can_operate():
+                        if not self.__is_udc_configured():
                             self.__state_flags.update(online=False)
                     else:
                         # Посылка свежих репортов важнее старого
@@ -160,6 +161,10 @@ class BaseDeviceProcess(multiprocessing.Process):  # pylint: disable=too-many-in
 
     # =====
 
+    def __is_udc_configured(self) -> bool:
+        with open(self.__udc_state_path) as udc_state_file:
+            return (udc_state_file.read().strip().lower() == "configured")
+
     def __write_report(self, report: bytes) -> bool:
         assert report
 

kvmd/usb.py

@@ -22,8 +22,6 @@
 
 import os
 
-from .logging import get_logger
-
 from . import env
 
 
@@ -38,20 +36,3 @@ def find_udc(udc: str) -> str:
     elif udc not in candidates:
         raise RuntimeError(f"Can't find selected UDC: {udc}")
     return udc  # fe980000.usb
-
-
-class UsbDeviceController:
-    def __init__(self, udc: str) -> None:
-        self.__udc = udc
-        self.__state_path = ""
-
-    def find(self) -> None:
-        udc = find_udc(self.__udc)
-        self.__state_path = os.path.join(f"{env.SYSFS_PREFIX}/sys/class/udc", udc, "state")
-        get_logger().info("Using UDC %s", udc)
-
-    def can_operate(self) -> bool:
-        assert self.__state_path
-        with open(self.__state_path, "r") as state_file:
-            # https://www.maxlinear.com/Files/Documents/an213_033111.pdf
-            return (state_file.read().strip().lower() == "configured")