use HCI_LE_Rand

add ASHA L2CAP and Event Emitter

bumble/device.py

@@ -46,6 +46,7 @@ from .hci import (
     HCI_LE_CODED_PHY_LE_SUPPORTED_FEATURE,
     HCI_LE_EXTENDED_ADVERTISING_LE_SUPPORTED_FEATURE,
     HCI_LE_EXTENDED_CREATE_CONNECTION_COMMAND,
+    HCI_LE_RAND_COMMAND,
     HCI_LE_READ_PHY_COMMAND,
     HCI_MITM_NOT_REQUIRED_GENERAL_BONDING_AUTHENTICATION_REQUIREMENTS,
     HCI_MITM_NOT_REQUIRED_NO_BONDING_AUTHENTICATION_REQUIREMENTS,
@@ -78,6 +79,7 @@ from .hci import (
     HCI_LE_Enable_Encryption_Command,
     HCI_LE_Extended_Advertising_Report_Event,
     HCI_LE_Extended_Create_Connection_Command,
+    HCI_LE_Rand_Command,
     HCI_LE_Read_PHY_Command,
     HCI_LE_Set_Advertising_Data_Command,
     HCI_LE_Set_Advertising_Enable_Command,
@@ -1113,10 +1115,35 @@ class Device(CompositeEventEmitter):
 
         if self.le_enabled:
             # Set the controller address
-            await self.send_command(
-                HCI_LE_Set_Random_Address_Command(random_address=self.random_address),
-                check_result=True,
-            )
+            if self.random_address == Address.ANY_RANDOM:
+                # Try to use an address generated at random by the controller
+                if self.host.supports_command(HCI_LE_RAND_COMMAND):
+                    # Get 8 random bytes
+                    response = await self.send_command(
+                        HCI_LE_Rand_Command(), check_result=True
+                    )
+
+                    # Ensure the address bytes can be a static random address
+                    address_bytes = response.return_parameters.random_number[
+                        :5
+                    ] + bytes([response.return_parameters.random_number[5] | 0xC0])
+
+                    # Create a static random address from the random bytes
+                    self.random_address = Address(address_bytes)
+
+            if self.random_address != Address.ANY_RANDOM:
+                logger.debug(
+                    color(
+                        f'LE Random Address: {self.random_address}',
+                        'yellow',
+                    )
+                )
+                await self.send_command(
+                    HCI_LE_Set_Random_Address_Command(
+                        random_address=self.random_address
+                    ),
+                    check_result=True,
+                )
 
             # Load the address resolving list
             if self.keystore and self.host.supports_command(

bumble/hci.py

@@ -1638,8 +1638,8 @@ class HCI_Object:
             # Map the value if needed
             if value_mappers:
                 value_mapper = value_mappers.get(key, value_mapper)
-                if value_mapper is not None:
-                    value = value_mapper(value)
+            if value_mapper is not None:
+                value = value_mapper(value)
 
             # Get the string representation of the value
             value_str = HCI_Object.format_field_value(
@@ -1807,6 +1807,7 @@ class Address:
 # Predefined address values
 Address.NIL = Address(b"\xff\xff\xff\xff\xff\xff", Address.PUBLIC_DEVICE_ADDRESS)
 Address.ANY = Address(b"\x00\x00\x00\x00\x00\x00", Address.PUBLIC_DEVICE_ADDRESS)
+Address.ANY_RANDOM = Address(b"\x00\x00\x00\x00\x00\x00", Address.RANDOM_DEVICE_ADDRESS)
 
 # -----------------------------------------------------------------------------
 class OwnAddressType:
@@ -3104,6 +3105,16 @@ class HCI_LE_Read_Remote_Features_Command(HCI_Command):
     '''
 
 
+# -----------------------------------------------------------------------------
+@HCI_Command.command(
+    return_parameters_fields=[("status", STATUS_SPEC), ("random_number", 8)]
+)
+class HCI_LE_Rand_Command(HCI_Command):
+    """
+    See Bluetooth spec @ 7.8.23 LE Rand Command
+    """
+
+
 # -----------------------------------------------------------------------------
 @HCI_Command.command(
     [

bumble/profiles/asha_service.py

@@ -31,6 +31,7 @@ from ..gatt import (
     Characteristic,
     CharacteristicValue,
 )
+from ..device import Device
 
 # -----------------------------------------------------------------------------
 # Logging
@@ -50,9 +51,13 @@ class AshaService(TemplateService):
     SUPPORTED_CODEC_ID = [0x02, 0x01]  # Codec IDs [G.722 at 16 kHz]
     RENDER_DELAY = [00, 00]
 
-    def __init__(self, capability: int, hisyncid: List[int]):
+    def __init__(self, capability: int, hisyncid: List[int], device: Device, psm=0):
         self.hisyncid = hisyncid
         self.capability = capability  # Device Capabilities [Left, Monaural]
+        self.device = device
+        self.emitted_data_name = 'ASHA_data_' + str(self.capability)
+        self.audio_out_data = b''
+        self.psm = psm  # a non-zero psm is mainly for testing purpose
 
         # Handler for volume control
         def on_volume_write(_connection, value):
@@ -116,9 +121,18 @@ class AshaService(TemplateService):
             CharacteristicValue(write=on_volume_write),
         )
 
-        # TODO add real psm value
-        self.psm = 0x0080
-        # self.psm = device.register_l2cap_channel_server(0, on_coc, 8)
+        # Register an L2CAP CoC server
+        def on_coc(channel):
+            def on_data(data):
+                logging.debug(f'<<< data received:{data}')
+
+                self.emit(self.emitted_data_name, data)
+                self.audio_out_data += data
+
+            channel.sink = on_data
+
+        # let the server find a free PSM
+        self.psm = self.device.register_l2cap_channel_server(self.psm, on_coc, 8)
         self.le_psm_out_characteristic = Characteristic(
             GATT_ASHA_LE_PSM_OUT_CHARACTERISTIC,
             Characteristic.READ,

bumble/transport/hci_socket.py

@@ -97,7 +97,7 @@ async def open_hci_socket_transport(spec):
             super().__init__()
             self.socket = hci_socket
             asyncio.get_running_loop().add_reader(
-                socket.fileno(), self.recv_until_would_block
+                self.socket.fileno(), self.recv_until_would_block
             )
 
         def recv_until_would_block(self):
@@ -140,7 +140,7 @@ async def open_hci_socket_transport(spec):
                 if not self.writer_added:
                     asyncio.get_running_loop().add_writer(
                         # pylint: disable=no-member
-                        socket.fileno(),
+                        self.socket.fileno(),
                         self.send_until_would_block,
                     )
                     self.writer_added = True