| 14 | return wrapper |
| 15 | |
| 16 | class USB3: |
| 17 | @staticmethod |
| 18 | @functools.cache |
| 19 | def ctx(): |
| 20 | ctx = c.init_c_var(ctypes.POINTER(libusb.struct_libusb_context), checked(libusb.libusb_init)) |
| 21 | if DEBUG >= 6: checked(libusb.libusb_set_option)(ctx, libusb.LIBUSB_OPTION_LOG_LEVEL, 4) |
| 22 | return ctx |
| 23 | |
| 24 | @classmethod |
| 25 | @functools.cache |
| 26 | def list_devices(cls, vendor:int, dev:int) -> list[tuple[c.POINTER[libusb.struct_libusb_device], str]]: |
| 27 | ret = [] |
| 28 | for i in range(checked(libusb.libusb_get_device_list)(cls.ctx(), devs:=ctypes.POINTER(ctypes.POINTER(libusb.struct_libusb_device))())): |
| 29 | desc = c.init_c_var(libusb.struct_libusb_device_descriptor, lambda x: checked(libusb.libusb_get_device_descriptor)(devs[i], x)) |
| 30 | if (desc.idVendor, desc.idProduct) == (vendor, dev): |
| 31 | ret.append((libusb.libusb_ref_device(devs[i]), f"usb:{libusb.libusb_get_bus_number(devs[i])}-{libusb.libusb_get_device_address(devs[i])}")) |
| 32 | libusb.libusb_free_device_list(devs, 1) |
| 33 | return ret |
| 34 | |
| 35 | def __init__(self, dev:c.POINTER[libusb.struct_libusb_device], ep_data_in:int, ep_stat_in:int, ep_data_out:int, ep_cmd_out:int, |
| 36 | max_streams:int=31, use_bot=False): |
| 37 | self.ep_data_in, self.ep_stat_in, self.ep_data_out, self.ep_cmd_out = ep_data_in, ep_stat_in, ep_data_out, ep_cmd_out |
| 38 | self.max_streams, self.use_bot = max_streams, use_bot |
| 39 | self._transferred = ctypes.c_int(0) |
| 40 | self._bulk_in_buf, self._bulk_in_mv = alloc_cbuffer(4 << 20) |
| 41 | self._bulk_out_buf, self._bulk_out_mv = alloc_cbuffer(4 << 20) |
| 42 | |
| 43 | self.handle = c.init_c_var(c.POINTER[libusb.struct_libusb_device_handle], lambda x: checked(libusb.libusb_open)(dev, x)) |
| 44 | |
| 45 | # Read product string descriptor |
| 46 | _buf = (ctypes.c_ubyte * 256)() |
| 47 | _desc = libusb.struct_libusb_device_descriptor() |
| 48 | checked(libusb.libusb_get_device_descriptor)(libusb.libusb_get_device(self.handle), ctypes.byref(_desc)) |
| 49 | _ret = checked(libusb.libusb_get_string_descriptor_ascii)(self.handle, _desc.iProduct, _buf, 256) |
| 50 | self.product = bytes(_buf[:_ret]).decode("ascii", errors="replace") |
| 51 | self.is_custom = self.product.startswith("custom") |
| 52 | if self.is_custom: self.use_bot = use_bot = True |
| 53 | |
| 54 | # Detach kernel driver if needed |
| 55 | if checked(libusb.libusb_kernel_driver_active)(self.handle, 0): |
| 56 | checked(libusb.libusb_detach_kernel_driver)(self.handle, 0) |
| 57 | checked(libusb.libusb_reset_device)(self.handle) |
| 58 | |
| 59 | # Set configuration and claim interface |
| 60 | checked(libusb.libusb_set_configuration)(self.handle, 1) |
| 61 | checked(libusb.libusb_claim_interface)(self.handle, 0) |
| 62 | |
| 63 | if use_bot: |
| 64 | checked(libusb.libusb_set_interface_alt_setting)(self.handle, 0, 0) |
| 65 | self._tag = 0 |
| 66 | else: |
| 67 | checked(libusb.libusb_set_interface_alt_setting)(self.handle, 0, 1) |
| 68 | |
| 69 | # Clear any stalled endpoints |
| 70 | all_eps = (self.ep_data_out, self.ep_data_in, self.ep_stat_in, self.ep_cmd_out) |
| 71 | for ep in all_eps: checked(libusb.libusb_clear_halt)(self.handle, ep) |
| 72 | |
| 73 | # Allocate streams |