pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

pyusb: Collect vendorId and productId as metadata

Browse Source
This commit is contained in:
Charlie Boutier
2024-03-15 17:27:53 +00:00
committed by Slvr
parent 16d5cf6770
commit e8ed69fb09
1 changed files with 6 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
bumble/transport/pyusb.py
View File

@@ -113,9 +113,10 @@ async def open_pyusb_transport(spec: str) -> Transport:
self.loop.call_soon_threadsafe(self.stop_event.set) self.loop.call_soon_threadsafe(self.stop_event.set)
class UsbPacketSource(asyncio.Protocol, ParserSource): class UsbPacketSource(asyncio.Protocol, ParserSource):
def __init__(self, device, sco_enabled): def __init__(self, device, metadata, sco_enabled):
super().__init__() super().__init__()
self.device = device self.device = device
self.metadata = metadata
self.loop = asyncio.get_running_loop() self.loop = asyncio.get_running_loop()
self.queue = asyncio.Queue() self.queue = asyncio.Queue()
self.dequeue_task = None self.dequeue_task = None
@@ -244,6 +245,9 @@ async def open_pyusb_transport(spec: str) -> Transport:
raise ValueError('device not found') raise ValueError('device not found')
logger.debug(f'USB Device: {device}') logger.debug(f'USB Device: {device}')
# Collect the metadata
device_metadata = {'vendor_id': device.idVendor, 'product_id': device.idProduct}
# Detach the kernel driver if needed # Detach the kernel driver if needed
if device.is_kernel_driver_active(0): if device.is_kernel_driver_active(0):
logger.debug("detaching kernel driver") logger.debug("detaching kernel driver")
@@ -298,7 +302,7 @@ async def open_pyusb_transport(spec: str) -> Transport:
# except usb.USBError: # except usb.USBError:
# logger.warning('failed to set alternate setting') # logger.warning('failed to set alternate setting')
packet_source = UsbPacketSource(device, sco_enabled) packet_source = UsbPacketSource(device, device_metadata, sco_enabled)
packet_sink = UsbPacketSink(device) packet_sink = UsbPacketSink(device)
packet_source.start() packet_source.start()
packet_sink.start() packet_sink.start()