From ecb8693f5d2255ed833ce7dd0c0e7cfe5799a090 Mon Sep 17 00:00:00 2001 From: dccote Date: Wed, 21 Oct 2026 11:04:24 -0400 Subject: [PATCH] USBPort: add readStream() for high-throughput bulk IN, plus serial matching and endpoint-by-address readData() reads one wMaxPacketSize packet at a time into an internal buffer, which is right for a command/response dialog but far too slow to drain a high-bandwidth bulk IN stream: a device that bursts faster than the host reads (a camera reading out its sensor) overruns its FIFO and the data comes back corrupted. - readStream(length, endPoint, chunkSize, timeout): read `length` bytes in large chunks (default wMaxPacketSize*16 = 16 KB, a SuperSpeed burst) in a tight loop, trimming the final short read. Independent of the internal buffer. - open() now honors serialNumber (via custom_match) so a specific unit can be selected when several identical devices are present; previously it was ignored. - endPointForAddress(address): resolve an endpoint by its USB address (e.g. 0x83), clearer than an interface index on multi-endpoint devices. readData/writeData/ flush now accept an endpoint object as well as an index (backward compatible). Validated on a ThorLabs Kiralux CS505MUP1 (5 MP polarization camera): command dialog on EP 0x02/0x82, and readStream on EP 0x83 delivering a full 7.5 MB frame at ~61 MB/s with no corruption, where packet-sized readData could not keep up. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Cwn4LmnBsBpXbQKRh6ByZr --- hardwarelibrary/communication/usbport.py | 76 ++++++++++++++++++++++-- 1 file changed, 72 insertions(+), 4 deletions(-) diff --git a/hardwarelibrary/communication/usbport.py b/hardwarelibrary/communication/usbport.py index 8f6f3578..b8780ab6 100644 --- a/hardwarelibrary/communication/usbport.py +++ b/hardwarelibrary/communication/usbport.py @@ -64,7 +64,16 @@ def open(self): self._internalBuffer = bytearray() - self.device = usb.core.find(idVendor=self.idVendor, idProduct=self.idProduct) + if self.serialNumber is not None: + def matchesSerial(device): + try: + return device.serial_number == self.serialNumber + except (ValueError, usb.core.USBError): + return False # serial unreadable without opening: not a match + self.device = usb.core.find(idVendor=self.idVendor, idProduct=self.idProduct, + custom_match=matchesSerial) + else: + self.device = usb.core.find(idVendor=self.idVendor, idProduct=self.idProduct) if self.device is None: raise IOError("Can't find device") @@ -106,8 +115,10 @@ def flush(self, endPoint=None): if endPoint is None: inputEndPoint = self.defaultInputEndPoint - else: + elif isinstance(endPoint, int): inputEndPoint = self.interface[endPoint] + else: + inputEndPoint = endPoint time.sleep(0.1) @@ -127,8 +138,10 @@ def readData(self, length, endPoint=None) -> bytearray: if endPoint is None: inputEndPoint = self.defaultInputEndPoint - else: + elif isinstance(endPoint, int): inputEndPoint = self.interface[endPoint] + else: + inputEndPoint = endPoint with self.portLock: while length > len(self._internalBuffer): @@ -141,14 +154,69 @@ def readData(self, length, endPoint=None) -> bytearray: return data + def endPointForAddress(self, address): + """Return the interface endpoint whose bEndpointAddress equals `address`. + + Lets a caller target an endpoint by its USB address (e.g. 0x83) rather than + by its index in the interface, which is clearer on devices with several + endpoints. Returns None if there is no such endpoint. + """ + for endPoint in self.interface: + if endPoint.bEndpointAddress == address: + return endPoint + return None + + def readStream(self, length, endPoint=None, chunkSize=None, timeout=None) -> bytearray: + """Read `length` bytes from a bulk IN endpoint at full throughput. + + Unlike readData(), which reads one wMaxPacketSize packet at a time into an + internal buffer for a command/response dialog, readStream() reads in large + chunks in a tight loop, which sustains the endpoint's full (SuperSpeed burst) + rate -- fast enough for an image or data stream, e.g. a multi-megapixel + camera frame, without the device's FIFO overrunning and corrupting the data. + + `endPoint` is the default input endpoint when None, an endpoint index (int, + as elsewhere in this class), or an endpoint object from endPointForAddress(). + `chunkSize` defaults to the endpoint's maximum burst (wMaxPacketSize * 16, + i.e. 16 KB for a SuperSpeed bulk endpoint); the final short read trims the + result to exactly `length`. This path does not touch the internal buffer. + """ + if not self.isOpen: + self.open() + + if endPoint is None: + inputEndPoint = self.defaultInputEndPoint + elif isinstance(endPoint, int): + inputEndPoint = self.interface[endPoint] + else: + inputEndPoint = endPoint + + if chunkSize is None: + chunkSize = inputEndPoint.wMaxPacketSize * 16 + if timeout is None: + timeout = self.defaultTimeout + + data = bytearray() + buffer = usb.util.create_buffer(chunkSize) + with self.portLock: + while len(data) < length: + nBytesRead = inputEndPoint.read(size_or_buffer=buffer, timeout=timeout) + if nBytesRead == 0: + break + data += bytearray(buffer[:nBytesRead]) + + return data[:length] + def writeData(self, data, endPoint=None) -> int: if not self.isOpen: self.open() if endPoint is None: outputEndPoint = self.defaultOutputEndPoint - else: + elif isinstance(endPoint, int): outputEndPoint = self.interface[endPoint] + else: + outputEndPoint = endPoint with self.portLock: nBytesWritten = outputEndPoint.write(data, timeout=self.defaultTimeout)