Merge pull request #198 from ap--/dynamically-limit-max-ep-size

Dynamically limit max ep size

src/seabreeze/pyseabreeze/transport.py

@@ -161,6 +161,24 @@ def open_device(self, device: USBTransportHandle) -> None:
             raise USBTransportError.from_usberror(err)
         else:
             self._opened = True
+
+        # configure the default_read_size according to pyusb info
+        ep_max_packet_size = {}
+        for intf in pyusb_device.get_active_configuration():
+            for ep in intf.endpoints():
+                ep_max_packet_size[ep.bEndpointAddress] = ep.wMaxPacketSize
+
+        for mode_name, endpoint_map_name in self._read_endpoints.items():
+            ep_int = getattr(self._endpoint_map, endpoint_map_name, None)
+            if ep_int is None:
+                continue
+            try:
+                max_size = ep_max_packet_size[ep_int]
+            except KeyError:
+                continue
+            cur_size = self._default_read_size[mode_name]
+            self._default_read_size[mode_name] = min(cur_size, max_size)
+
         # This will initialize the communication protocol
         if self._opened:
             self._protocol = self._protocol_cls(self)

tests/test_spectrometers.py

@@ -71,12 +71,13 @@ def _aquire_connected_usb_spectrometers(timeout=10.0):
     return params, ids, len(ids), supports
 
 
+_timeout = int(os.environ.get("SEABREEZE_SPECTROMETER_TEST_CONNECTED_TIMEOUT", "10"))
 (
     _SPEC_PARAMS,
     _SPEC_IDS,
     _SPEC_NUM,
     _SPEC_SUPPORTS,
-) = _aquire_connected_usb_spectrometers()
+) = _aquire_connected_usb_spectrometers(_timeout)
 
 
 @pytest.fixture(scope="function", params=_SPEC_PARAMS, ids=_SPEC_IDS)