Improved USB hid adapter.

Author: David Cedar

programmor_adapters/shared/api.py

@@ -128,7 +128,7 @@ def run(self) -> None:
             self._process_scheduled_messages()
 
             # Sleep the thread
-            sleep(0.0001)  # 0.1ms
+            sleep(0.001)  # 0.1ms
 
     def _process_scheduled_messages(self) -> None:
         """Process Scheduled Messages

programmor_adapters/shared/comm.py

@@ -63,11 +63,14 @@ def run(self) -> None:
             # Process incoming messages
             self.process_incoming_frames()
 
+            # Sleep the thread
+            sleep(0.005)  # 5ms
+
             # Process outgoing messages
             self.process_outgoing_frames()
 
             # Sleep the thread
-            sleep(0.0001)  # 0.1ms
+            sleep(0.005)  # 5ms
 
     def process_incoming_frames(self) -> ProcessState:
         """Processes the incoming frames.
@@ -185,6 +188,7 @@ def process_outgoing_frames(self) -> ProcessState:
             # Write data
             frame_data = frame.to_bytes()
             self.write(frame_data)
+            sleep(0.001)  # 1ms
 
         return ProcessState.OK
 

programmor_adapters/shared/socket_endpoint.py

@@ -35,7 +35,7 @@ class SocketEndpoint(Endpoint):
     class ApiNamespace(WebSocketEndpoint):
         api: API
         ws_connections: List[WebSocket]
-        emit: Callable
+        emit: Callable  # type: ignore
 
         async def get_devices(self, _):
             """Get Devices
@@ -219,7 +219,7 @@ def __init__(self, api: API, port: int) -> None:
         self.ApiNamespace.emit = self.emit
         api.register_callback(lambda data: self.emit_data(data))
         self.app = Starlette(routes=routes, middleware=middleware, on_startup=[self.start_thread])
-        self.message_queue = queue.Queue()
+        self.message_queue: queue.Queue[ResponseType] = queue.Queue()
         self.stop_event = threading.Event()
         logger.debug('Websocket Endpoint Initialised')
 
@@ -268,7 +268,7 @@ def patch_log_started_message(listeners):
             orig_log_started_message(listeners)
             server_started(server)
 
-        server._log_started_message = patch_log_started_message
+        server._log_started_message = patch_log_started_message  # type: ignore
         logger.debug('Starting socket endpoint')
         server.run()
 
@@ -277,4 +277,4 @@ def stop(self) -> None:
         """
         self.stop_event.set()
         for ws in self.ws_connections:
-            ws.close()
+            ws.close()  # type: ignore

programmor_adapters/test_adapter/__main__.py

@@ -2,4 +2,4 @@
 
 
 # Run Application
-main()
+main()

programmor_adapters/test_adapter/main.py

@@ -91,7 +91,7 @@ def main():
             time.sleep(0.1)  # Sleep to keep the main thread alive
     except KeyboardInterrupt:
         logger.info("User stopping Application")
-        
+
     # Stopping Adapter
     logger.info("Stopping Application")
     api.stop()

programmor_adapters/test_adapter/test_device.py

@@ -146,7 +146,7 @@ def process_data(self, data: bytes) -> None:
         elif inMessage.action == transaction_pb2.TransactionMessage.SHARE_PUBLISH:  # type: ignore
             if inMessage.shareId == 1:
                 # Share1 publish
-                inData: bytes = inMessage.data[0:inMessage.dataLength]
+                inData: bytes = inMessage.data[0:inMessage.dataLength]  # type: ignore
                 testMessage: test_pb2.TestMessage = test_pb2.Share1()  # type: ignore
                 try:
                     testMessage.ParseFromString(bytes(inData[0:DATA_MAX_SIZE]))
@@ -159,7 +159,7 @@ def process_data(self, data: bytes) -> None:
 
             elif inMessage.shareId == 2:
                 # Share2 publish
-                inData: bytes = inMessage.data[0:inMessage.dataLength]
+                inData: bytes = inMessage.data[0:inMessage.dataLength]  # type: ignore
                 testMessage: test_pb2.TestMessage = test_pb2.Share2()  # type: ignore
                 try:
                     testMessage.ParseFromString(bytes(inData[0:DATA_MAX_SIZE]))
@@ -175,7 +175,7 @@ def process_data(self, data: bytes) -> None:
 
             elif inMessage.shareId == 4:
                 # Share4 publish
-                inData: bytes = inMessage.data[0:inMessage.dataLength]
+                inData: bytes = inMessage.data[0:inMessage.dataLength]  # type: ignore
                 testMessage: test_pb2.TestMessage = test_pb2.Share4()  # type: ignore
                 try:
                     testMessage.ParseFromString(bytes(inData[0:DATA_MAX_SIZE]))
@@ -187,7 +187,7 @@ def process_data(self, data: bytes) -> None:
 
             elif inMessage.shareId == 5:
                 # Share5 publish
-                inData: bytes = inMessage.data[0:inMessage.dataLength]
+                inData: bytes = inMessage.data[0:inMessage.dataLength]  # type: ignore
                 testMessage: test_pb2.TestMessage = test_pb2.Share5()  # type: ignore
                 try:
                     testMessage.ParseFromString(bytes(inData[0:DATA_MAX_SIZE]))
@@ -204,7 +204,7 @@ def process_data(self, data: bytes) -> None:
 
             elif inMessage.shareId == 6:
                 # Share6 publish
-                inData: bytes = inMessage.data[0:inMessage.dataLength]
+                inData: bytes = inMessage.data[0:inMessage.dataLength]  # type: ignore
                 testMessage: test_pb2.TestMessage = test_pb2.Share6()  # type: ignore
                 try:
                     testMessage.ParseFromString(bytes(inData[0:DATA_MAX_SIZE]))

programmor_adapters/usb_adapter/usb.py

@@ -1,10 +1,12 @@
-from typing import Optional
 import usb.core
 import usb.util
 import libusb_package
+from typing import Optional
 
 from shared.comm import Comm
 from usb_adapter.helper import get_device_endpoints
+from usb_adapter.usb_read import USBRead
+from usb_adapter.usb_write import USBWrite
 
 import logging
 logger = logging.getLogger(__name__)
@@ -24,27 +26,24 @@ def __init__(self, device_id_vender: int, device_id_product: int) -> None:
         # meta data
         # self.device_manufacturer: str = None
         # self.device_product: str = None
+        # Endpoints.
+        self.read_endpoint: Optional[USBRead] = None
+        self.write_endpoint: Optional[USBWrite] = None
 
     def __str__(self) -> str:
         return f"USB(id: vender_id: {self.device_id_vender} product_id: {self.device_id_product})"
 
     def read(self) -> bytes:
         # Return no bytes if device is not ready
-        if self.device is None:
-            return bytes(0)
-        try:
-            if self.device_endpoint_in is not None:
-                return bytes(self.device_endpoint_in.read(64, 100))
-            else:
-                return bytes(0)
-        except Exception:
+        if self.device is None or self.read_endpoint is None:
             return bytes(0)
+        return self.read_endpoint.read()
 
     def write(self, buffer: bytes) -> None:
         if len(buffer) > 0 and self.device is None:
             return  # Not connected to device
-        if self.device_endpoint_out is not None:
-            self.device_endpoint_out.write(bytes(buffer), 100)
+        if self.write_endpoint:
+            self.write_endpoint.write(buffer)
 
     def connect(self) -> bool:
         self.device: Optional[usb.core.Device] = libusb_package.find(idVendor=self.device_id_vender, idProduct=self.device_id_product)  # type: ignore
@@ -53,6 +52,10 @@ def connect(self) -> bool:
         self.device_endpoint_in, self.device_endpoint_out = get_device_endpoints(self.device)
         if self.device_endpoint_in is None or self.device_endpoint_out is None:
             return False
+        self.read_endpoint = USBRead(self.device_endpoint_in, 1)
+        self.write_endpoint = USBWrite(self.device_endpoint_out, 1)
+        self.read_endpoint.start()
+        self.write_endpoint.start()
         logger.debug(f"Connected {self.__str__()}")
         return True
 
@@ -62,6 +65,12 @@ def close(self) -> None:
             # self.device.reset()
             usb.util.dispose_resources(self.device)
             self.device = None
+            if self.read_endpoint:
+                self.read_endpoint.stop()
+            if self.write_endpoint:
+                self.write_endpoint.stop()
+            self.read_endpoint = None
+            self.write_endpoint = None
             self.device_endpoint_in = None
             self.device_endpoint_out = None
             logger.debug(f"Closed {self.__str__()}")

programmor_adapters/usb_adapter/usb_endpoint.py

@@ -0,0 +1,54 @@
+from time import sleep
+from typing import Optional
+from collections import deque
+import threading
+import usb.core
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+class USBEndpoint(threading.Thread):
+    """
+    USBEndpoint.
+    A generic USB endpoint. Extend this class to implement the reader/writer.
+    """
+    def __init__(self, endpoint: Optional[usb.core.Endpoint], interval: int) -> None:
+        """Constructor method
+        """
+        threading.Thread.__init__(self)
+        self.stop_flag: bool = False
+        self.endpoint = endpoint
+        self.interval = interval
+        self.queue: deque[bytes] = deque()
+
+    def start(self) -> None:
+        """Starts the Endpoint
+        """
+        threading.Thread.start(self)
+        self.stop_flag = False
+        logger.debug("Starting USBEndpoint Thread")
+
+    def stop(self) -> None:
+        """Stops the thread
+        """
+        logger.debug("Stopping USBEndpoint Thread")
+        self.stop_flag = True
+
+    def run(self) -> None:
+        """Run method used by python threading
+        """
+        while True:
+            # Stop thread
+            if self.stop_flag:
+                logger.debug("Stopped USBEndpoint Thread")
+                break
+
+            # Process incoming messages
+            self.action()
+
+            # Sleep the thread as per the interval.
+            sleep(self.interval/1000)
+
+    def action(self):
+        raise NotImplementedError("Action method not implemented")

programmor_adapters/usb_adapter/usb_manager.py

@@ -41,7 +41,9 @@ def get_devices(self) -> List[str]:
             # Check if device is available
             if dev is None:
                 continue
-            # Check if meta data is available
+            # Skip keyboards and mice
+            if dev.bDeviceClass == 0x03:
+                continue
             try:
                 assert dev.manufacturer is not None
                 assert dev.product is not None
@@ -50,7 +52,7 @@ def get_devices(self) -> List[str]:
             # Skip STM32 Link V3
             if dev.idVendor == 1155 and dev.idProduct == 14159:
                 continue
-            # print(f"{dev.manufacturer} {dev.product}")
+            print(f"{dev.manufacturer} {dev.product}")
             # Check if the device is already in use
             device_id = self.get_device_id_from_vender_product(dev.idVendor, dev.idProduct)
             if device_id:

programmor_adapters/usb_adapter/usb_read.py

@@ -0,0 +1,27 @@
+from usb_adapter.usb_endpoint import USBEndpoint
+from typing import Optional
+import usb.core
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+class USBRead(USBEndpoint):
+    def __init__(self, endpoint: Optional[usb.core.Endpoint], interval: int) -> None:
+        super().__init__(endpoint, interval)
+
+    def action(self) -> None:
+        try:
+            if self.endpoint is not None:
+                self.queue.append(bytes(self.endpoint.read(64, 1)))
+                # print("Read")
+        except Exception:
+            # print("Read Error")
+            # print(e)
+            pass
+
+    def read(self) -> bytes:
+        try:
+            return self.queue.popleft()
+        except Exception:
+            return bytes(0)