A2DP example: Codec selection based on file type

Currently support SBC and AAC

.github/ci-gradle.properties

@@ -0,0 +1,26 @@
+#
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.gradle.configureondemand=true
+org.gradle.caching=true
+org.gradle.parallel=true
+
+# Declare we support AndroidX
+android.useAndroidX=true
+
+org.gradle.jvmargs=-Xmx4608m -XX:MaxMetaspaceSize=1536m -XX:+HeapDumpOnOutOfMemoryError
+
+kotlin.compiler.execution.strategy=in-process

.github/workflows/code-check.yml

@@ -33,7 +33,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install ".[build,test,development]"
+        python -m pip install ".[build,examples,test,development]"
     - name: Check
       run: |
         invoke project.pre-commit

.github/workflows/gradle-btbench.yml

@@ -0,0 +1,33 @@
+name: Gradle Android Build & test
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+    paths:
+      - 'extras/android/BtBench/**'
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    timeout-minutes: 40
+
+    steps:
+      - name: Check out from Git
+        uses: actions/checkout@v3
+
+      - name: Set up JDK
+        uses: actions/setup-java@v4
+        with:
+          distribution: 'zulu'
+          java-version: 17
+
+      - name: Setup Gradle
+        uses: gradle/actions/setup-gradle@v3
+
+      - name: Build with Gradle
+        run: cd extras/android/BtBench && ./gradlew build

apps/bench.py

@@ -121,9 +121,9 @@ def print_connection(connection):
 
         params.append(
             'Parameters='
-            f'{connection.parameters.connection_interval * 1.25:.2f}/'
+            f'{connection.parameters.connection_interval:.2f}/'
             f'{connection.parameters.peripheral_latency}/'
-            f'{connection.parameters.supervision_timeout * 10} '
+            f'{connection.parameters.supervision_timeout:.2f} '
         )
 
         params.append(f'MTU={connection.att_mtu}')
@@ -1256,6 +1256,7 @@ class Central(Connection.Listener):
             self.device.classic_enabled = self.classic
 
             # Set up a pairing config factory with minimal requirements.
+            self.device.config.keystore = "JsonKeyStore"
             self.device.pairing_config_factory = lambda _: PairingConfig(
                 sc=False, mitm=False, bonding=False
             )
@@ -1408,6 +1409,7 @@ class Peripheral(Device.Listener, Connection.Listener):
             self.device.classic_enabled = self.classic
 
             # Set up a pairing config factory with minimal requirements.
+            self.device.config.keystore = "JsonKeyStore"
             self.device.pairing_config_factory = lambda _: PairingConfig(
                 sc=False, mitm=False, bonding=False
             )

apps/console.py

@@ -335,9 +335,9 @@ class ConsoleApp:
             elif self.connected_peer:
                 connection = self.connected_peer.connection
                 connection_parameters = (
-                    f'{connection.parameters.connection_interval}/'
+                    f'{connection.parameters.connection_interval:.2f}/'
                     f'{connection.parameters.peripheral_latency}/'
-                    f'{connection.parameters.supervision_timeout}'
+                    f'{connection.parameters.supervision_timeout:.2f}'
                 )
                 if self.connection_phy is not None:
                     phy_state = (

apps/pair.py

@@ -18,9 +18,12 @@
 import asyncio
 import os
 import logging
+import struct
+
 import click
 from prompt_toolkit.shortcuts import PromptSession
 
+from bumble.a2dp import make_audio_sink_service_sdp_records
 from bumble.colors import color
 from bumble.device import Device, Peer
 from bumble.transport import open_transport_or_link
@@ -30,16 +33,20 @@ from bumble.smp import error_name as smp_error_name
 from bumble.keys import JsonKeyStore
 from bumble.core import (
     AdvertisingData,
+    Appearance,
     ProtocolError,
     PhysicalTransport,
+    UUID,
 )
 from bumble.gatt import (
     GATT_DEVICE_NAME_CHARACTERISTIC,
     GATT_GENERIC_ACCESS_SERVICE,
+    GATT_HEART_RATE_SERVICE,
+    GATT_HEART_RATE_MEASUREMENT_CHARACTERISTIC,
     Service,
     Characteristic,
-    CharacteristicValue,
 )
+from bumble.hci import OwnAddressType
 from bumble.att import (
     ATT_Error,
     ATT_INSUFFICIENT_AUTHENTICATION_ERROR,
@@ -62,7 +69,7 @@ class Waiter:
         self.linger = linger
 
     def terminate(self):
-        if not self.linger:
+        if not self.linger and not self.done.done:
             self.done.set_result(None)
 
     async def wait_until_terminated(self):
@@ -193,7 +200,7 @@ class Delegate(PairingDelegate):
 
 # -----------------------------------------------------------------------------
 async def get_peer_name(peer, mode):
-    if mode == 'classic':
+    if peer.connection.transport == PhysicalTransport.BR_EDR:
         return await peer.request_name()
 
     # Try to get the peer name from GATT
@@ -225,13 +232,14 @@ def read_with_error(connection):
     raise ATT_Error(ATT_INSUFFICIENT_AUTHENTICATION_ERROR)
 
 
-def write_with_error(connection, _value):
-    if not connection.is_encrypted:
-        raise ATT_Error(ATT_INSUFFICIENT_ENCRYPTION_ERROR)
-
-    if not AUTHENTICATION_ERROR_RETURNED[1]:
-        AUTHENTICATION_ERROR_RETURNED[1] = True
-        raise ATT_Error(ATT_INSUFFICIENT_AUTHENTICATION_ERROR)
+# -----------------------------------------------------------------------------
+def sdp_records():
+    service_record_handle = 0x00010001
+    return {
+        service_record_handle: make_audio_sink_service_sdp_records(
+            service_record_handle
+        )
+    }
 
 
 # -----------------------------------------------------------------------------
@@ -239,15 +247,19 @@ def on_connection(connection, request):
     print(color(f'<<< Connection: {connection}', 'green'))
 
     # Listen for pairing events
-    connection.on('pairing_start', on_pairing_start)
-    connection.on('pairing', lambda keys: on_pairing(connection, keys))
+    connection.on(connection.EVENT_PAIRING_START, on_pairing_start)
+    connection.on(connection.EVENT_PAIRING, lambda keys: on_pairing(connection, keys))
     connection.on(
-        'pairing_failure', lambda reason: on_pairing_failure(connection, reason)
+        connection.EVENT_CLASSIC_PAIRING, lambda: on_classic_pairing(connection)
+    )
+    connection.on(
+        connection.EVENT_PAIRING_FAILURE,
+        lambda reason: on_pairing_failure(connection, reason),
     )
 
     # Listen for encryption changes
     connection.on(
-        'connection_encryption_change',
+        connection.EVENT_CONNECTION_ENCRYPTION_CHANGE,
         lambda: on_connection_encryption_change(connection),
     )
 
@@ -288,6 +300,20 @@ async def on_pairing(connection, keys):
     Waiter.instance.terminate()
 
 
+# -----------------------------------------------------------------------------
+@AsyncRunner.run_in_task()
+async def on_classic_pairing(connection):
+    print(color('***-----------------------------------', 'cyan'))
+    print(
+        color(
+            f'*** Paired [Classic]! (peer identity={connection.peer_address})', 'cyan'
+        )
+    )
+    print(color('***-----------------------------------', 'cyan'))
+    await asyncio.sleep(POST_PAIRING_DELAY)
+    Waiter.instance.terminate()
+
+
 # -----------------------------------------------------------------------------
 @AsyncRunner.run_in_task()
 async def on_pairing_failure(connection, reason):
@@ -305,6 +331,7 @@ async def pair(
     mitm,
     bond,
     ctkd,
+    advertising_address,
     identity_address,
     linger,
     io,
@@ -313,6 +340,8 @@ async def pair(
     request,
     print_keys,
     keystore_file,
+    advertise_service_uuids,
+    advertise_appearance,
     device_config,
     hci_transport,
     address_or_name,
@@ -328,29 +357,33 @@ async def pair(
 
         # Expose a GATT characteristic that can be used to trigger pairing by
         # responding with an authentication error when read
-        if mode == 'le':
-            device.le_enabled = True
+        if mode in ('le', 'dual'):
             device.add_service(
                 Service(
-                    '50DB505C-8AC4-4738-8448-3B1D9CC09CC5',
+                    GATT_HEART_RATE_SERVICE,
                     [
                         Characteristic(
-                            '552957FB-CF1F-4A31-9535-E78847E1A714',
-                            Characteristic.Properties.READ
-                            | Characteristic.Properties.WRITE,
-                            Characteristic.READABLE | Characteristic.WRITEABLE,
-                            CharacteristicValue(
-                                read=read_with_error, write=write_with_error
-                            ),
+                            GATT_HEART_RATE_MEASUREMENT_CHARACTERISTIC,
+                            Characteristic.Properties.READ,
+                            Characteristic.READ_REQUIRES_AUTHENTICATION,
+                            bytes(1),
                         )
                     ],
                 )
             )
 
-        # Select LE or Classic
-        if mode == 'classic':
+        # LE and Classic support
+        if mode in ('classic', 'dual'):
             device.classic_enabled = True
             device.classic_smp_enabled = ctkd
+        if mode in ('le', 'dual'):
+            device.le_enabled = True
+        if mode == 'dual':
+            device.le_simultaneous_enabled = True
+
+        # Setup SDP
+        if mode in ('classic', 'dual'):
+            device.sdp_service_records = sdp_records()
 
         # Get things going
         await device.power_on()
@@ -436,13 +469,109 @@ async def pair(
                     print(color(f'Pairing failed: {error}', 'red'))
 
         else:
-            if mode == 'le':
-                # Advertise so that peers can find us and connect
-                await device.start_advertising(auto_restart=True)
-            else:
+            if mode in ('le', 'dual'):
+                # Advertise so that peers can find us and connect.
+                # Include the heart rate service UUID in the advertisement data
+                # so that devices like iPhones can show this device in their
+                # Bluetooth selector.
+                service_uuids_16 = []
+                service_uuids_32 = []
+                service_uuids_128 = []
+                if advertise_service_uuids:
+                    for uuid in advertise_service_uuids:
+                        uuid = uuid.replace("-", "")
+                        if len(uuid) == 4:
+                            service_uuids_16.append(UUID(uuid))
+                        elif len(uuid) == 8:
+                            service_uuids_32.append(UUID(uuid))
+                        elif len(uuid) == 32:
+                            service_uuids_128.append(UUID(uuid))
+                        else:
+                            print(color('Invalid UUID format', 'red'))
+                            return
+                else:
+                    service_uuids_16.append(GATT_HEART_RATE_SERVICE)
+
+                flags = AdvertisingData.Flags.LE_LIMITED_DISCOVERABLE_MODE
+                if mode == 'le':
+                    flags |= AdvertisingData.Flags.BR_EDR_NOT_SUPPORTED
+                if mode == 'dual':
+                    flags |= AdvertisingData.Flags.SIMULTANEOUS_LE_BR_EDR_CAPABLE
+
+                ad_structs = [
+                    (
+                        AdvertisingData.FLAGS,
+                        bytes([flags]),
+                    ),
+                    (AdvertisingData.COMPLETE_LOCAL_NAME, 'Bumble'.encode()),
+                ]
+                if service_uuids_16:
+                    ad_structs.append(
+                        (
+                            AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
+                            b"".join(bytes(uuid) for uuid in service_uuids_16),
+                        )
+                    )
+                if service_uuids_32:
+                    ad_structs.append(
+                        (
+                            AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
+                            b"".join(bytes(uuid) for uuid in service_uuids_32),
+                        )
+                    )
+                if service_uuids_128:
+                    ad_structs.append(
+                        (
+                            AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
+                            b"".join(bytes(uuid) for uuid in service_uuids_128),
+                        )
+                    )
+
+                if advertise_appearance:
+                    advertise_appearance = advertise_appearance.upper()
+                    try:
+                        advertise_appearance_int = int(advertise_appearance)
+                    except ValueError:
+                        category, subcategory = advertise_appearance.split('/')
+                        try:
+                            category_enum = Appearance.Category[category]
+                        except ValueError:
+                            print(
+                                color(f'Invalid appearance category {category}', 'red')
+                            )
+                            return
+                        subcategory_class = Appearance.SUBCATEGORY_CLASSES[
+                            category_enum
+                        ]
+                        try:
+                            subcategory_enum = subcategory_class[subcategory]
+                        except ValueError:
+                            print(color(f'Invalid subcategory {subcategory}', 'red'))
+                            return
+                        advertise_appearance_int = int(
+                            Appearance(category_enum, subcategory_enum)
+                        )
+                    ad_structs.append(
+                        (
+                            AdvertisingData.APPEARANCE,
+                            struct.pack('<H', advertise_appearance_int),
+                        )
+                    )
+                device.advertising_data = bytes(AdvertisingData(ad_structs))
+                await device.start_advertising(
+                    auto_restart=True,
+                    own_address_type=(
+                        OwnAddressType.PUBLIC
+                        if advertising_address == 'public'
+                        else OwnAddressType.RANDOM
+                    ),
+                )
+
+            if mode in ('classic', 'dual'):
                 # Become discoverable and connectable
                 await device.set_discoverable(True)
                 await device.set_connectable(True)
+                print(color('Ready for connections on', 'blue'), device.public_address)
 
         # Run until the user asks to exit
         await Waiter.instance.wait_until_terminated()
@@ -462,7 +591,10 @@ class LogHandler(logging.Handler):
 # -----------------------------------------------------------------------------
 @click.command()
 @click.option(
-    '--mode', type=click.Choice(['le', 'classic']), default='le', show_default=True
+    '--mode',
+    type=click.Choice(['le', 'classic', 'dual']),
+    default='le',
+    show_default=True,
 )
 @click.option(
     '--sc',
@@ -484,6 +616,10 @@ class LogHandler(logging.Handler):
     help='Enable CTKD',
     show_default=True,
 )
+@click.option(
+    '--advertising-address',
+    type=click.Choice(['random', 'public']),
+)
 @click.option(
     '--identity-address',
     type=click.Choice(['random', 'public']),
@@ -512,9 +648,20 @@ class LogHandler(logging.Handler):
 @click.option('--print-keys', is_flag=True, help='Print the bond keys before pairing')
 @click.option(
     '--keystore-file',
-    metavar='<filename>',
+    metavar='FILENAME',
     help='File in which to store the pairing keys',
 )
+@click.option(
+    '--advertise-service-uuid',
+    metavar="UUID",
+    multiple=True,
+    help="Advertise a GATT service UUID (may be specified more than once)",
+)
+@click.option(
+    '--advertise-appearance',
+    metavar='APPEARANCE',
+    help='Advertise an Appearance ID (int value or string)',
+)
 @click.argument('device-config')
 @click.argument('hci_transport')
 @click.argument('address-or-name', required=False)
@@ -524,6 +671,7 @@ def main(
     mitm,
     bond,
     ctkd,
+    advertising_address,
     identity_address,
     linger,
     io,
@@ -532,6 +680,8 @@ def main(
     request,
     print_keys,
     keystore_file,
+    advertise_service_uuid,
+    advertise_appearance,
     device_config,
     hci_transport,
     address_or_name,
@@ -550,6 +700,7 @@ def main(
             mitm,
             bond,
             ctkd,
+            advertising_address,
             identity_address,
             linger,
             io,
@@ -558,6 +709,8 @@ def main(
             request,
             print_keys,
             keystore_file,
+            advertise_service_uuid,
+            advertise_appearance,
             device_config,
             hci_transport,
             address_or_name,

bumble/att.py

@@ -836,6 +836,9 @@ class Attribute(utils.EventEmitter, Generic[_T]):
     READ_REQUIRES_AUTHORIZATION = Permissions.READ_REQUIRES_AUTHORIZATION
     WRITE_REQUIRES_AUTHORIZATION = Permissions.WRITE_REQUIRES_AUTHORIZATION
 
+    EVENT_READ = "read"
+    EVENT_WRITE = "write"
+
     value: Union[AttributeValue[_T], _T, None]
 
     def __init__(
@@ -906,7 +909,7 @@ class Attribute(utils.EventEmitter, Generic[_T]):
         else:
             value = self.value
 
-        self.emit('read', connection, b'' if value is None else value)
+        self.emit(self.EVENT_READ, connection, b'' if value is None else value)
 
         return b'' if value is None else self.encode_value(value)
 
@@ -947,7 +950,7 @@ class Attribute(utils.EventEmitter, Generic[_T]):
         else:
             self.value = decoded_value
 
-        self.emit('write', connection, decoded_value)
+        self.emit(self.EVENT_WRITE, connection, decoded_value)
 
     def __repr__(self):
         if isinstance(self.value, bytes):

bumble/avctp.py

@@ -166,8 +166,8 @@ class Protocol:
 
         # Register to receive PDUs from the channel
         l2cap_channel.sink = self.on_pdu
-        l2cap_channel.on("open", self.on_l2cap_channel_open)
-        l2cap_channel.on("close", self.on_l2cap_channel_close)
+        l2cap_channel.on(l2cap_channel.EVENT_OPEN, self.on_l2cap_channel_open)
+        l2cap_channel.on(l2cap_channel.EVENT_CLOSE, self.on_l2cap_channel_close)
 
     def on_l2cap_channel_open(self):
         logger.debug(color("<<< AVCTP channel open", "magenta"))

bumble/avdtp.py

@@ -896,7 +896,7 @@ class Set_Configuration_Reject(Message):
         self.service_category = self.payload[0]
         self.error_code = self.payload[1]
 
-    def __init__(self, service_category, error_code):
+    def __init__(self, error_code: int, service_category: int = 0) -> None:
         super().__init__(payload=bytes([service_category, error_code]))
         self.service_category = service_category
         self.error_code = error_code
@@ -1132,6 +1132,14 @@ class Security_Control_Command(Message):
     See Bluetooth AVDTP spec - 8.17.1 Security Control Command
     '''
 
+    def init_from_payload(self):
+        # pylint: disable=attribute-defined-outside-init
+        self.acp_seid = self.payload[0] >> 2
+        self.data = self.payload[1:]
+
+    def __str__(self) -> str:
+        return self.to_string([f'ACP_SEID: {self.acp_seid}', f'data:    {self.data}'])
+
 
 # -----------------------------------------------------------------------------
 @Message.subclass
@@ -1200,6 +1208,9 @@ class Protocol(utils.EventEmitter):
     transaction_results: List[Optional[asyncio.Future[Message]]]
     channel_connector: Callable[[], Awaitable[l2cap.ClassicChannel]]
 
+    EVENT_OPEN = "open"
+    EVENT_CLOSE = "close"
+
     class PacketType(enum.IntEnum):
         SINGLE_PACKET = 0
         START_PACKET = 1
@@ -1239,8 +1250,8 @@ class Protocol(utils.EventEmitter):
 
         # Register to receive PDUs from the channel
         l2cap_channel.sink = self.on_pdu
-        l2cap_channel.on('open', self.on_l2cap_channel_open)
-        l2cap_channel.on('close', self.on_l2cap_channel_close)
+        l2cap_channel.on(l2cap_channel.EVENT_OPEN, self.on_l2cap_channel_open)
+        l2cap_channel.on(l2cap_channel.EVENT_CLOSE, self.on_l2cap_channel_close)
 
     def get_local_endpoint_by_seid(self, seid: int) -> Optional[LocalStreamEndPoint]:
         if 0 < seid <= len(self.local_endpoints):
@@ -1410,20 +1421,20 @@ class Protocol(utils.EventEmitter):
             self.transaction_results[transaction_label] = None
             self.transaction_semaphore.release()
 
-    def on_l2cap_connection(self, channel):
+    def on_l2cap_connection(self, channel: l2cap.ClassicChannel) -> None:
         # Forward the channel to the endpoint that's expecting it
         if self.channel_acceptor is None:
             logger.warning(color('!!! l2cap connection with no acceptor', 'red'))
             return
         self.channel_acceptor.on_l2cap_connection(channel)
 
-    def on_l2cap_channel_open(self):
+    def on_l2cap_channel_open(self) -> None:
         logger.debug(color('<<< L2CAP channel open', 'magenta'))
-        self.emit('open')
+        self.emit(self.EVENT_OPEN)
 
-    def on_l2cap_channel_close(self):
+    def on_l2cap_channel_close(self) -> None:
         logger.debug(color('<<< L2CAP channel close', 'magenta'))
-        self.emit('close')
+        self.emit(self.EVENT_CLOSE)
 
     def send_message(self, transaction_label: int, message: Message) -> None:
         logger.debug(
@@ -1541,28 +1552,34 @@ class Protocol(utils.EventEmitter):
     async def abort(self, seid: int) -> Abort_Response:
         return await self.send_command(Abort_Command(seid))
 
-    def on_discover_command(self, _command):
+    def on_discover_command(self, command: Discover_Command) -> Optional[Message]:
         endpoint_infos = [
             EndPointInfo(endpoint.seid, 0, endpoint.media_type, endpoint.tsep)
             for endpoint in self.local_endpoints
         ]
         return Discover_Response(endpoint_infos)
 
-    def on_get_capabilities_command(self, command):
+    def on_get_capabilities_command(
+        self, command: Get_Capabilities_Command
+    ) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return Get_Capabilities_Reject(AVDTP_BAD_ACP_SEID_ERROR)
 
         return Get_Capabilities_Response(endpoint.capabilities)
 
-    def on_get_all_capabilities_command(self, command):
+    def on_get_all_capabilities_command(
+        self, command: Get_All_Capabilities_Command
+    ) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return Get_All_Capabilities_Reject(AVDTP_BAD_ACP_SEID_ERROR)
 
         return Get_All_Capabilities_Response(endpoint.capabilities)
 
-    def on_set_configuration_command(self, command):
+    def on_set_configuration_command(
+        self, command: Set_Configuration_Command
+    ) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return Set_Configuration_Reject(AVDTP_BAD_ACP_SEID_ERROR)
@@ -1578,7 +1595,9 @@ class Protocol(utils.EventEmitter):
         result = stream.on_set_configuration_command(command.capabilities)
         return result or Set_Configuration_Response()
 
-    def on_get_configuration_command(self, command):
+    def on_get_configuration_command(
+        self, command: Get_Configuration_Command
+    ) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return Get_Configuration_Reject(AVDTP_BAD_ACP_SEID_ERROR)
@@ -1587,7 +1606,7 @@ class Protocol(utils.EventEmitter):
 
         return endpoint.stream.on_get_configuration_command()
 
-    def on_reconfigure_command(self, command):
+    def on_reconfigure_command(self, command: Reconfigure_Command) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return Reconfigure_Reject(0, AVDTP_BAD_ACP_SEID_ERROR)
@@ -1597,7 +1616,7 @@ class Protocol(utils.EventEmitter):
         result = endpoint.stream.on_reconfigure_command(command.capabilities)
         return result or Reconfigure_Response()
 
-    def on_open_command(self, command):
+    def on_open_command(self, command: Open_Command) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return Open_Reject(AVDTP_BAD_ACP_SEID_ERROR)
@@ -1607,25 +1626,26 @@ class Protocol(utils.EventEmitter):
         result = endpoint.stream.on_open_command()
         return result or Open_Response()
 
-    def on_start_command(self, command):
+    def on_start_command(self, command: Start_Command) -> Optional[Message]:
         for seid in command.acp_seids:
             endpoint = self.get_local_endpoint_by_seid(seid)
             if endpoint is None:
                 return Start_Reject(seid, AVDTP_BAD_ACP_SEID_ERROR)
             if endpoint.stream is None:
-                return Start_Reject(AVDTP_BAD_STATE_ERROR)
+                return Start_Reject(seid, AVDTP_BAD_STATE_ERROR)
 
         # Start all streams
         # TODO: deal with partial failures
         for seid in command.acp_seids:
             endpoint = self.get_local_endpoint_by_seid(seid)
-            result = endpoint.stream.on_start_command()
-            if result is not None:
+            if not endpoint or not endpoint.stream:
+                raise InvalidStateError("Should already be checked!")
+            if (result := endpoint.stream.on_start_command()) is not None:
                 return result
 
         return Start_Response()
 
-    def on_suspend_command(self, command):
+    def on_suspend_command(self, command: Suspend_Command) -> Optional[Message]:
         for seid in command.acp_seids:
             endpoint = self.get_local_endpoint_by_seid(seid)
             if endpoint is None:
@@ -1637,13 +1657,14 @@ class Protocol(utils.EventEmitter):
         # TODO: deal with partial failures
         for seid in command.acp_seids:
             endpoint = self.get_local_endpoint_by_seid(seid)
-            result = endpoint.stream.on_suspend_command()
-            if result is not None:
+            if not endpoint or not endpoint.stream:
+                raise InvalidStateError("Should already be checked!")
+            if (result := endpoint.stream.on_suspend_command()) is not None:
                 return result
 
         return Suspend_Response()
 
-    def on_close_command(self, command):
+    def on_close_command(self, command: Close_Command) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return Close_Reject(AVDTP_BAD_ACP_SEID_ERROR)
@@ -1653,7 +1674,7 @@ class Protocol(utils.EventEmitter):
         result = endpoint.stream.on_close_command()
         return result or Close_Response()
 
-    def on_abort_command(self, command):
+    def on_abort_command(self, command: Abort_Command) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None or endpoint.stream is None:
             return Abort_Response()
@@ -1661,15 +1682,17 @@ class Protocol(utils.EventEmitter):
         endpoint.stream.on_abort_command()
         return Abort_Response()
 
-    def on_security_control_command(self, command):
+    def on_security_control_command(
+        self, command: Security_Control_Command
+    ) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return Security_Control_Reject(AVDTP_BAD_ACP_SEID_ERROR)
 
-        result = endpoint.on_security_control_command(command.payload)
+        result = endpoint.on_security_control_command(command.data)
         return result or Security_Control_Response()
 
-    def on_delayreport_command(self, command):
+    def on_delayreport_command(self, command: DelayReport_Command) -> Optional[Message]:
         endpoint = self.get_local_endpoint_by_seid(command.acp_seid)
         if endpoint is None:
             return DelayReport_Reject(AVDTP_BAD_ACP_SEID_ERROR)
@@ -1682,6 +1705,8 @@ class Protocol(utils.EventEmitter):
 class Listener(utils.EventEmitter):
     servers: Dict[int, Protocol]
 
+    EVENT_CONNECTION = "connection"
+
     @staticmethod
     def create_registrar(device: device.Device):
         warnings.warn("Please use Listener.for_device()", DeprecationWarning)
@@ -1716,7 +1741,7 @@ class Listener(utils.EventEmitter):
         l2cap_server = device.create_l2cap_server(
             spec=l2cap.ClassicChannelSpec(psm=AVDTP_PSM)
         )
-        l2cap_server.on('connection', listener.on_l2cap_connection)
+        l2cap_server.on(l2cap_server.EVENT_CONNECTION, listener.on_l2cap_connection)
         return listener
 
     def on_l2cap_connection(self, channel: l2cap.ClassicChannel) -> None:
@@ -1732,14 +1757,14 @@ class Listener(utils.EventEmitter):
                 logger.debug('setting up new Protocol for the connection')
                 server = Protocol(channel, self.version)
                 self.set_server(channel.connection, server)
-                self.emit('connection', server)
+                self.emit(self.EVENT_CONNECTION, server)
 
             def on_channel_close():
                 logger.debug('removing Protocol for the connection')
                 self.remove_server(channel.connection)
 
-            channel.on('open', on_channel_open)
-            channel.on('close', on_channel_close)
+            channel.on(channel.EVENT_OPEN, on_channel_open)
+            channel.on(channel.EVENT_CLOSE, on_channel_close)
 
 
 # -----------------------------------------------------------------------------
@@ -1788,6 +1813,7 @@ class Stream:
         )
 
     async def start(self) -> None:
+        """[Source] Start streaming."""
         # Auto-open if needed
         if self.state == AVDTP_CONFIGURED_STATE:
             await self.open()
@@ -1804,6 +1830,7 @@ class Stream:
         self.change_state(AVDTP_STREAMING_STATE)
 
     async def stop(self) -> None:
+        """[Source] Stop streaming and transit to OPEN state."""
         if self.state != AVDTP_STREAMING_STATE:
             raise InvalidStateError('current state is not STREAMING')
 
@@ -1816,6 +1843,7 @@ class Stream:
         self.change_state(AVDTP_OPEN_STATE)
 
     async def close(self) -> None:
+        """[Source] Close channel and transit to IDLE state."""
         if self.state not in (AVDTP_OPEN_STATE, AVDTP_STREAMING_STATE):
             raise InvalidStateError('current state is not OPEN or STREAMING')
 
@@ -1847,7 +1875,7 @@ class Stream:
         self.change_state(AVDTP_CONFIGURED_STATE)
         return None
 
-    def on_get_configuration_command(self, configuration):
+    def on_get_configuration_command(self):
         if self.state not in (
             AVDTP_CONFIGURED_STATE,
             AVDTP_OPEN_STATE,
@@ -1855,7 +1883,7 @@ class Stream:
         ):
             return Get_Configuration_Reject(AVDTP_BAD_STATE_ERROR)
 
-        return self.local_endpoint.on_get_configuration_command(configuration)
+        return self.local_endpoint.on_get_configuration_command()
 
     def on_reconfigure_command(self, configuration):
         if self.state != AVDTP_OPEN_STATE:
@@ -1935,20 +1963,20 @@ class Stream:
             # Wait for the RTP channel to be closed
             self.change_state(AVDTP_ABORTING_STATE)
 
-    def on_l2cap_connection(self, channel):
+    def on_l2cap_connection(self, channel: l2cap.ClassicChannel) -> None:
         logger.debug(color('<<< stream channel connected', 'magenta'))
         self.rtp_channel = channel
-        channel.on('open', self.on_l2cap_channel_open)
-        channel.on('close', self.on_l2cap_channel_close)
+        channel.on(channel.EVENT_OPEN, self.on_l2cap_channel_open)
+        channel.on(channel.EVENT_CLOSE, self.on_l2cap_channel_close)
 
         # We don't need more channels
         self.protocol.channel_acceptor = None
 
-    def on_l2cap_channel_open(self):
+    def on_l2cap_channel_open(self) -> None:
         logger.debug(color('<<< stream channel open', 'magenta'))
         self.local_endpoint.on_rtp_channel_open()
 
-    def on_l2cap_channel_close(self):
+    def on_l2cap_channel_close(self) -> None:
         logger.debug(color('<<< stream channel closed', 'magenta'))
         self.local_endpoint.on_rtp_channel_close()
         self.local_endpoint.in_use = 0
@@ -2065,6 +2093,19 @@ class DiscoveredStreamEndPoint(StreamEndPoint, StreamEndPointProxy):
 class LocalStreamEndPoint(StreamEndPoint, utils.EventEmitter):
     stream: Optional[Stream]
 
+    EVENT_CONFIGURATION = "configuration"
+    EVENT_OPEN = "open"
+    EVENT_START = "start"
+    EVENT_STOP = "stop"
+    EVENT_RTP_PACKET = "rtp_packet"
+    EVENT_SUSPEND = "suspend"
+    EVENT_CLOSE = "close"
+    EVENT_ABORT = "abort"
+    EVENT_DELAY_REPORT = "delay_report"
+    EVENT_SECURITY_CONTROL = "security_control"
+    EVENT_RTP_CHANNEL_OPEN = "rtp_channel_open"
+    EVENT_RTP_CHANNEL_CLOSE = "rtp_channel_close"
+
     def __init__(
         self,
         protocol: Protocol,
@@ -2080,52 +2121,65 @@ class LocalStreamEndPoint(StreamEndPoint, utils.EventEmitter):
         self.configuration = configuration if configuration is not None else []
         self.stream = None
 
-    async def start(self):
-        pass
+    async def start(self) -> None:
+        """[Source Only] Handles when receiving start command."""
 
-    async def stop(self):
-        pass
+    async def stop(self) -> None:
+        """[Source Only] Handles when receiving stop command."""
 
-    async def close(self):
-        pass
+    async def close(self) -> None:
+        """[Source Only] Handles when receiving close command."""
 
-    def on_reconfigure_command(self, command):
-        pass
+    def on_reconfigure_command(self, command) -> Optional[Message]:
+        return None
 
-    def on_set_configuration_command(self, configuration):
+    def on_set_configuration_command(self, configuration) -> Optional[Message]:
         logger.debug(
             '<<< received configuration: '
             f'{",".join([str(capability) for capability in configuration])}'
         )
         self.configuration = configuration
-        self.emit('configuration')
+        self.emit(self.EVENT_CONFIGURATION)
+        return None
 
-    def on_get_configuration_command(self):
+    def on_get_configuration_command(self) -> Optional[Message]:
         return Get_Configuration_Response(self.configuration)
 
-    def on_open_command(self):
-        self.emit('open')
+    def on_open_command(self) -> Optional[Message]:
+        self.emit(self.EVENT_OPEN)
+        return None
 
-    def on_start_command(self):
-        self.emit('start')
+    def on_start_command(self) -> Optional[Message]:
+        self.emit(self.EVENT_START)
+        return None
 
-    def on_suspend_command(self):
-        self.emit('suspend')
+    def on_suspend_command(self) -> Optional[Message]:
+        self.emit(self.EVENT_SUSPEND)
+        return None
 
-    def on_close_command(self):
-        self.emit('close')
+    def on_close_command(self) -> Optional[Message]:
+        self.emit(self.EVENT_CLOSE)
+        return None
 
-    def on_abort_command(self):
-        self.emit('abort')
+    def on_abort_command(self) -> Optional[Message]:
+        self.emit(self.EVENT_ABORT)
+        return None
 
-    def on_delayreport_command(self, delay: int):
-        self.emit('delay_report', delay)
+    def on_delayreport_command(self, delay: int) -> Optional[Message]:
+        self.emit(self.EVENT_DELAY_REPORT, delay)
+        return None
 
-    def on_rtp_channel_open(self):
-        self.emit('rtp_channel_open')
+    def on_security_control_command(self, data: bytes) -> Optional[Message]:
+        self.emit(self.EVENT_SECURITY_CONTROL, data)
+        return None
 
-    def on_rtp_channel_close(self):
-        self.emit('rtp_channel_close')
+    def on_rtp_channel_open(self) -> None:
+        self.emit(self.EVENT_RTP_CHANNEL_OPEN)
+        return None
+
+    def on_rtp_channel_close(self) -> None:
+        self.emit(self.EVENT_RTP_CHANNEL_CLOSE)
+        return None
 
 
 # -----------------------------------------------------------------------------
@@ -2156,13 +2210,13 @@ class LocalSource(LocalStreamEndPoint):
         if self.packet_pump and self.stream and self.stream.rtp_channel:
             return await self.packet_pump.start(self.stream.rtp_channel)
 
-        self.emit('start')
+        self.emit(self.EVENT_START)
 
     async def stop(self) -> None:
         if self.packet_pump:
             return await self.packet_pump.stop()
 
-        self.emit('stop')
+        self.emit(self.EVENT_STOP)
 
     def on_start_command(self):
         asyncio.create_task(self.start())
@@ -2203,4 +2257,4 @@ class LocalSink(LocalStreamEndPoint):
             f'{color("<<< RTP Packet:", "green")} '
             f'{rtp_packet} {rtp_packet.payload[:16].hex()}'
         )
-        self.emit('rtp_packet', rtp_packet)
+        self.emit(self.EVENT_RTP_PACKET, rtp_packet)

bumble/avrcp.py

@@ -996,6 +996,10 @@ class Delegate:
 class Protocol(utils.EventEmitter):
     """AVRCP Controller and Target protocol."""
 
+    EVENT_CONNECTION = "connection"
+    EVENT_START = "start"
+    EVENT_STOP = "stop"
+
     class PacketType(enum.IntEnum):
         SINGLE = 0b00
         START = 0b01
@@ -1456,9 +1460,11 @@ class Protocol(utils.EventEmitter):
 
     def _on_avctp_connection(self, l2cap_channel: l2cap.ClassicChannel) -> None:
         logger.debug("AVCTP connection established")
-        l2cap_channel.on("open", lambda: self._on_avctp_channel_open(l2cap_channel))
+        l2cap_channel.on(
+            l2cap_channel.EVENT_OPEN, lambda: self._on_avctp_channel_open(l2cap_channel)
+        )
 
-        self.emit("connection")
+        self.emit(self.EVENT_CONNECTION)
 
     def _on_avctp_channel_open(self, l2cap_channel: l2cap.ClassicChannel) -> None:
         logger.debug("AVCTP channel open")
@@ -1473,15 +1479,15 @@ class Protocol(utils.EventEmitter):
         self.avctp_protocol.register_response_handler(
             AVRCP_PID, self._on_avctp_response
         )
-        l2cap_channel.on("close", self._on_avctp_channel_close)
+        l2cap_channel.on(l2cap_channel.EVENT_CLOSE, self._on_avctp_channel_close)
 
-        self.emit("start")
+        self.emit(self.EVENT_START)
 
     def _on_avctp_channel_close(self) -> None:
         logger.debug("AVCTP channel closed")
         self.avctp_protocol = None
 
-        self.emit("stop")
+        self.emit(self.EVENT_STOP)
 
     def _on_avctp_command(
         self, transaction_label: int, command: avc.CommandFrame

bumble/core.py

@@ -809,7 +809,7 @@ class Appearance:
         STICK_PC = 0x0F
 
     class WatchSubcategory(utils.OpenIntEnum):
-        GENENERIC_WATCH = 0x00
+        GENERIC_WATCH = 0x00
         SPORTS_WATCH = 0x01
         SMARTWATCH = 0x02
 
@@ -1127,7 +1127,7 @@ class Appearance:
         TURNTABLE = 0x05
         CD_PLAYER = 0x06
         DVD_PLAYER = 0x07
-        BLUERAY_PLAYER = 0x08
+        BLURAY_PLAYER = 0x08
         OPTICAL_DISC_PLAYER = 0x09
         SET_TOP_BOX = 0x0A
 
@@ -1351,6 +1351,12 @@ class AdvertisingData:
         THREE_D_INFORMATION_DATA                            = 0x3D
         MANUFACTURER_SPECIFIC_DATA                          = 0xFF
 
+    class Flags(enum.IntFlag):
+        LE_LIMITED_DISCOVERABLE_MODE = 1 << 0
+        LE_GENERAL_DISCOVERABLE_MODE = 1 << 1
+        BR_EDR_NOT_SUPPORTED = 1 << 2
+        SIMULTANEOUS_LE_BR_EDR_CAPABLE = 1 << 3
+
     # For backward-compatibility
     FLAGS                                            = Type.FLAGS
     INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS    = Type.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS
@@ -1407,11 +1413,11 @@ class AdvertisingData:
     THREE_D_INFORMATION_DATA                         = Type.THREE_D_INFORMATION_DATA
     MANUFACTURER_SPECIFIC_DATA                       = Type.MANUFACTURER_SPECIFIC_DATA
 
-    LE_LIMITED_DISCOVERABLE_MODE_FLAG = 0x01
-    LE_GENERAL_DISCOVERABLE_MODE_FLAG = 0x02
-    BR_EDR_NOT_SUPPORTED_FLAG         = 0x04
-    BR_EDR_CONTROLLER_FLAG            = 0x08
-    BR_EDR_HOST_FLAG                  = 0x10
+    LE_LIMITED_DISCOVERABLE_MODE_FLAG = Flags.LE_LIMITED_DISCOVERABLE_MODE
+    LE_GENERAL_DISCOVERABLE_MODE_FLAG = Flags.LE_GENERAL_DISCOVERABLE_MODE
+    BR_EDR_NOT_SUPPORTED_FLAG         = Flags.BR_EDR_NOT_SUPPORTED
+    BR_EDR_CONTROLLER_FLAG            = Flags.SIMULTANEOUS_LE_BR_EDR_CAPABLE
+    BR_EDR_HOST_FLAG                  = 0x10 # Deprecated
 
     ad_structures: list[tuple[int, bytes]]
 

bumble/gatt.py

@@ -448,6 +448,8 @@ class Characteristic(Attribute[_T]):
     uuid: UUID
     properties: Characteristic.Properties
 
+    EVENT_SUBSCRIPTION = "subscription"
+
     class Properties(enum.IntFlag):
         """Property flags"""
 

bumble/gatt_client.py

@@ -202,6 +202,8 @@ class CharacteristicProxy(AttributeProxy[_T]):
     descriptors: List[DescriptorProxy]
     subscribers: Dict[Any, Callable[[_T], Any]]
 
+    EVENT_UPDATE = "update"
+
     def __init__(
         self,
         client: Client,
@@ -308,7 +310,7 @@ class Client:
         self.services = []
         self.cached_values = {}
 
-        connection.on('disconnection', self.on_disconnection)
+        connection.on(connection.EVENT_DISCONNECTION, self.on_disconnection)
 
     def send_gatt_pdu(self, pdu: bytes) -> None:
         self.connection.send_l2cap_pdu(ATT_CID, pdu)
@@ -1142,7 +1144,7 @@ class Client:
             if callable(subscriber):
                 subscriber(notification.attribute_value)
             else:
-                subscriber.emit('update', notification.attribute_value)
+                subscriber.emit(subscriber.EVENT_UPDATE, notification.attribute_value)
 
     def on_att_handle_value_indication(self, indication):
         # Call all subscribers
@@ -1157,7 +1159,7 @@ class Client:
             if callable(subscriber):
                 subscriber(indication.attribute_value)
             else:
-                subscriber.emit('update', indication.attribute_value)
+                subscriber.emit(subscriber.EVENT_UPDATE, indication.attribute_value)
 
         # Confirm that we received the indication
         self.send_confirmation(ATT_Handle_Value_Confirmation())

bumble/gatt_server.py

@@ -110,6 +110,8 @@ class Server(utils.EventEmitter):
     indication_semaphores: defaultdict[int, asyncio.Semaphore]
     pending_confirmations: defaultdict[int, Optional[asyncio.futures.Future]]
 
+    EVENT_CHARACTERISTIC_SUBSCRIPTION = "characteristic_subscription"
+
     def __init__(self, device: Device) -> None:
         super().__init__()
         self.device = device
@@ -347,10 +349,13 @@ class Server(utils.EventEmitter):
         notify_enabled = value[0] & 0x01 != 0
         indicate_enabled = value[0] & 0x02 != 0
         characteristic.emit(
-            'subscription', connection, notify_enabled, indicate_enabled
+            characteristic.EVENT_SUBSCRIPTION,
+            connection,
+            notify_enabled,
+            indicate_enabled,
         )
         self.emit(
-            'characteristic_subscription',
+            self.EVENT_CHARACTERISTIC_SUBSCRIPTION,
             connection,
             characteristic,
             notify_enabled,

bumble/hci.py

@@ -29,13 +29,12 @@ from typing_extensions import Self
 from bumble import crypto
 from bumble.colors import color
 from bumble.core import (
-    PhysicalTransport,
     AdvertisingData,
     DeviceClass,
     InvalidArgumentError,
     InvalidPacketError,
-    ProtocolError,
     PhysicalTransport,
+    ProtocolError,
     bit_flags_to_strings,
     name_or_number,
     padded_bytes,
@@ -225,6 +224,7 @@ HCI_CONNECTIONLESS_PERIPHERAL_BROADCAST_CHANNEL_MAP_CHANGE_EVENT = 0X55
 HCI_INQUIRY_RESPONSE_NOTIFICATION_EVENT                          = 0X56
 HCI_AUTHENTICATED_PAYLOAD_TIMEOUT_EXPIRED_EVENT                  = 0X57
 HCI_SAM_STATUS_CHANGE_EVENT                                      = 0X58
+HCI_ENCRYPTION_CHANGE_V2_EVENT                                   = 0x59
 
 HCI_VENDOR_EVENT = 0xFF
 
@@ -3364,6 +3364,20 @@ class HCI_Set_Event_Mask_Page_2_Command(HCI_Command):
     See Bluetooth spec @ 7.3.69 Set Event Mask Page 2 Command
     '''
 
+    @staticmethod
+    def mask(event_codes: Iterable[int]) -> bytes:
+        '''
+        Compute the event mask value for a list of events.
+        '''
+        # NOTE: this implementation takes advantage of the fact that as of version 6.0
+        # of the core specification, the bit number for each event code is equal to 64
+        # less than the event code.
+        # If future versions of the specification deviate from that, a different
+        # implementation would be needed.
+        return sum((1 << event_code - 64) for event_code in event_codes).to_bytes(
+            8, 'little'
+        )
+
 
 # -----------------------------------------------------------------------------
 @HCI_Command.command(
@@ -5971,6 +5985,33 @@ class HCI_LE_Enhanced_Connection_Complete_Event(HCI_LE_Meta_Event):
     '''
 
 
+# -----------------------------------------------------------------------------
+@HCI_LE_Meta_Event.event(
+    [
+        ('status', STATUS_SPEC),
+        ('connection_handle', 2),
+        (
+            'role',
+            {'size': 1, 'mapper': lambda x: 'CENTRAL' if x == 0 else 'PERIPHERAL'},
+        ),
+        ('peer_address_type', Address.ADDRESS_TYPE_SPEC),
+        ('peer_address', Address.parse_address_preceded_by_type),
+        ('local_resolvable_private_address', Address.parse_random_address),
+        ('peer_resolvable_private_address', Address.parse_random_address),
+        ('connection_interval', 2),
+        ('peripheral_latency', 2),
+        ('supervision_timeout', 2),
+        ('central_clock_accuracy', 1),
+        ('advertising_handle', 1),
+        ('sync_handle', 2),
+    ]
+)
+class HCI_LE_Enhanced_Connection_Complete_V2_Event(HCI_LE_Meta_Event):
+    '''
+    See Bluetooth spec @ 7.7.65.10 LE Enhanced Connection Complete Event
+    '''
+
+
 # -----------------------------------------------------------------------------
 @HCI_LE_Meta_Event.event(
     [
@@ -6950,6 +6991,30 @@ class HCI_Encryption_Change_Event(HCI_Event):
         )
 
 
+# -----------------------------------------------------------------------------
+@HCI_Event.event(
+    [
+        ('status', STATUS_SPEC),
+        ('connection_handle', 2),
+        (
+            'encryption_enabled',
+            {
+                'size': 1,
+                # pylint: disable-next=unnecessary-lambda
+                'mapper': lambda x: HCI_Encryption_Change_Event.encryption_enabled_name(
+                    x
+                ),
+            },
+        ),
+        ('encryption_key_size', 1),
+    ]
+)
+class HCI_Encryption_Change_V2_Event(HCI_Event):
+    '''
+    See Bluetooth spec @ 7.7.8 Encryption Change Event
+    '''
+
+
 # -----------------------------------------------------------------------------
 @HCI_Event.event(
     [('status', STATUS_SPEC), ('connection_handle', 2), ('lmp_features', 8)]

bumble/hfp.py

@@ -720,6 +720,14 @@ class HfProtocol(utils.EventEmitter):
                 vrec: VoiceRecognitionState
     """
 
+    EVENT_CODEC_NEGOTIATION = "codec_negotiation"
+    EVENT_AG_INDICATOR = "ag_indicator"
+    EVENT_SPEAKER_VOLUME = "speaker_volume"
+    EVENT_MICROPHONE_VOLUME = "microphone_volume"
+    EVENT_RING = "ring"
+    EVENT_CLI_NOTIFICATION = "cli_notification"
+    EVENT_VOICE_RECOGNITION = "voice_recognition"
+
     class HfLoopTermination(HfpProtocolError):
         """Termination signal for run() loop."""
 
@@ -777,7 +785,8 @@ class HfProtocol(utils.EventEmitter):
         self.dlc.sink = self._read_at
         # Stop the run() loop when L2CAP is closed.
         self.dlc.multiplexer.l2cap_channel.on(
-            'close', lambda: self.unsolicited_queue.put_nowait(None)
+            self.dlc.multiplexer.l2cap_channel.EVENT_CLOSE,
+            lambda: self.unsolicited_queue.put_nowait(None),
         )
 
     def supports_hf_feature(self, feature: HfFeature) -> bool:
@@ -1034,7 +1043,7 @@ class HfProtocol(utils.EventEmitter):
         # ID. The HF shall be ready to accept the synchronous connection
         # establishment as soon as it has sent the AT commands AT+BCS=<Codec ID>.
         self.active_codec = AudioCodec(codec_id)
-        self.emit('codec_negotiation', self.active_codec)
+        self.emit(self.EVENT_CODEC_NEGOTIATION, self.active_codec)
 
         logger.info("codec connection setup completed")
 
@@ -1095,7 +1104,7 @@ class HfProtocol(utils.EventEmitter):
         # CIEV is in 1-index, while ag_indicators is in 0-index.
         ag_indicator = self.ag_indicators[index - 1]
         ag_indicator.current_status = value
-        self.emit('ag_indicator', ag_indicator)
+        self.emit(self.EVENT_AG_INDICATOR, ag_indicator)
         logger.info(f"AG indicator updated: {ag_indicator.indicator}, {value}")
 
     async def handle_unsolicited(self):
@@ -1110,19 +1119,21 @@ class HfProtocol(utils.EventEmitter):
                 int(result.parameters[0]), int(result.parameters[1])
             )
         elif result.code == "+VGS":
-            self.emit('speaker_volume', int(result.parameters[0]))
+            self.emit(self.EVENT_SPEAKER_VOLUME, int(result.parameters[0]))
         elif result.code == "+VGM":
-            self.emit('microphone_volume', int(result.parameters[0]))
+            self.emit(self.EVENT_MICROPHONE_VOLUME, int(result.parameters[0]))
         elif result.code == "RING":
-            self.emit('ring')
+            self.emit(self.EVENT_RING)
         elif result.code == "+CLIP":
             self.emit(
-                'cli_notification', CallLineIdentification.parse_from(result.parameters)
+                self.EVENT_CLI_NOTIFICATION,
+                CallLineIdentification.parse_from(result.parameters),
             )
         elif result.code == "+BVRA":
             # TODO: Support Enhanced Voice Recognition.
             self.emit(
-                'voice_recognition', VoiceRecognitionState(int(result.parameters[0]))
+                self.EVENT_VOICE_RECOGNITION,
+                VoiceRecognitionState(int(result.parameters[0])),
             )
         else:
             logging.info(f"unhandled unsolicited response {result.code}")
@@ -1179,6 +1190,19 @@ class AgProtocol(utils.EventEmitter):
                 volume: Int
     """
 
+    EVENT_SLC_COMPLETE = "slc_complete"
+    EVENT_SUPPORTED_AUDIO_CODECS = "supported_audio_codecs"
+    EVENT_CODEC_NEGOTIATION = "codec_negotiation"
+    EVENT_VOICE_RECOGNITION = "voice_recognition"
+    EVENT_CALL_HOLD = "call_hold"
+    EVENT_HF_INDICATOR = "hf_indicator"
+    EVENT_CODEC_CONNECTION_REQUEST = "codec_connection_request"
+    EVENT_ANSWER = "answer"
+    EVENT_DIAL = "dial"
+    EVENT_HANG_UP = "hang_up"
+    EVENT_SPEAKER_VOLUME = "speaker_volume"
+    EVENT_MICROPHONE_VOLUME = "microphone_volume"
+
     supported_hf_features: int
     supported_hf_indicators: Set[HfIndicator]
     supported_audio_codecs: List[AudioCodec]
@@ -1371,7 +1395,7 @@ class AgProtocol(utils.EventEmitter):
 
     def _check_remained_slc_commands(self) -> None:
         if not self._remained_slc_setup_features:
-            self.emit('slc_complete')
+            self.emit(self.EVENT_SLC_COMPLETE)
 
     def _on_brsf(self, hf_features: bytes) -> None:
         self.supported_hf_features = int(hf_features)
@@ -1390,17 +1414,17 @@ class AgProtocol(utils.EventEmitter):
 
     def _on_bac(self, *args) -> None:
         self.supported_audio_codecs = [AudioCodec(int(value)) for value in args]
-        self.emit('supported_audio_codecs', self.supported_audio_codecs)
+        self.emit(self.EVENT_SUPPORTED_AUDIO_CODECS, self.supported_audio_codecs)
         self.send_ok()
 
     def _on_bcs(self, codec: bytes) -> None:
         self.active_codec = AudioCodec(int(codec))
         self.send_ok()
-        self.emit('codec_negotiation', self.active_codec)
+        self.emit(self.EVENT_CODEC_NEGOTIATION, self.active_codec)
 
     def _on_bvra(self, vrec: bytes) -> None:
         self.send_ok()
-        self.emit('voice_recognition', VoiceRecognitionState(int(vrec)))
+        self.emit(self.EVENT_VOICE_RECOGNITION, VoiceRecognitionState(int(vrec)))
 
     def _on_chld(self, operation_code: bytes) -> None:
         call_index: Optional[int] = None
@@ -1427,7 +1451,7 @@ class AgProtocol(utils.EventEmitter):
         # Real three-way calls have more complicated situations, but this is not a popular issue - let users to handle the remaining :)
 
         self.send_ok()
-        self.emit('call_hold', operation, call_index)
+        self.emit(self.EVENT_CALL_HOLD, operation, call_index)
 
     def _on_chld_test(self) -> None:
         if not self.supports_ag_feature(AgFeature.THREE_WAY_CALLING):
@@ -1553,7 +1577,7 @@ class AgProtocol(utils.EventEmitter):
             return
 
         self.hf_indicators[index].current_status = int(value_bytes)
-        self.emit('hf_indicator', self.hf_indicators[index])
+        self.emit(self.EVENT_HF_INDICATOR, self.hf_indicators[index])
         self.send_ok()
 
     def _on_bia(self, *args) -> None:
@@ -1562,21 +1586,21 @@ class AgProtocol(utils.EventEmitter):
         self.send_ok()
 
     def _on_bcc(self) -> None:
-        self.emit('codec_connection_request')
+        self.emit(self.EVENT_CODEC_CONNECTION_REQUEST)
         self.send_ok()
 
     def _on_a(self) -> None:
         """ATA handler."""
-        self.emit('answer')
+        self.emit(self.EVENT_ANSWER)
         self.send_ok()
 
     def _on_d(self, number: bytes) -> None:
         """ATD handler."""
-        self.emit('dial', number.decode())
+        self.emit(self.EVENT_DIAL, number.decode())
         self.send_ok()
 
     def _on_chup(self) -> None:
-        self.emit('hang_up')
+        self.emit(self.EVENT_HANG_UP)
         self.send_ok()
 
     def _on_clcc(self) -> None:
@@ -1602,11 +1626,11 @@ class AgProtocol(utils.EventEmitter):
         self.send_ok()
 
     def _on_vgs(self, level: bytes) -> None:
-        self.emit('speaker_volume', int(level))
+        self.emit(self.EVENT_SPEAKER_VOLUME, int(level))
         self.send_ok()
 
     def _on_vgm(self, level: bytes) -> None:
-        self.emit('microphone_volume', int(level))
+        self.emit(self.EVENT_MICROPHONE_VOLUME, int(level))
         self.send_ok()
 
 

bumble/hid.py

@@ -201,6 +201,13 @@ class HID(ABC, utils.EventEmitter):
     l2cap_intr_channel: Optional[l2cap.ClassicChannel] = None
     connection: Optional[device.Connection] = None
 
+    EVENT_INTERRUPT_DATA = "interrupt_data"
+    EVENT_CONTROL_DATA = "control_data"
+    EVENT_SUSPEND = "suspend"
+    EVENT_EXIT_SUSPEND = "exit_suspend"
+    EVENT_VIRTUAL_CABLE_UNPLUG = "virtual_cable_unplug"
+    EVENT_HANDSHAKE = "handshake"
+
     class Role(enum.IntEnum):
         HOST = 0x00
         DEVICE = 0x01
@@ -215,7 +222,7 @@ class HID(ABC, utils.EventEmitter):
         device.register_l2cap_server(HID_CONTROL_PSM, self.on_l2cap_connection)
         device.register_l2cap_server(HID_INTERRUPT_PSM, self.on_l2cap_connection)
 
-        device.on('connection', self.on_device_connection)
+        device.on(device.EVENT_CONNECTION, self.on_device_connection)
 
     async def connect_control_channel(self) -> None:
         # Create a new L2CAP connection - control channel
@@ -258,15 +265,20 @@ class HID(ABC, utils.EventEmitter):
     def on_device_connection(self, connection: device.Connection) -> None:
         self.connection = connection
         self.remote_device_bd_address = connection.peer_address
-        connection.on('disconnection', self.on_device_disconnection)
+        connection.on(connection.EVENT_DISCONNECTION, self.on_device_disconnection)
 
     def on_device_disconnection(self, reason: int) -> None:
         self.connection = None
 
     def on_l2cap_connection(self, l2cap_channel: l2cap.ClassicChannel) -> None:
         logger.debug(f'+++ New L2CAP connection: {l2cap_channel}')
-        l2cap_channel.on('open', lambda: self.on_l2cap_channel_open(l2cap_channel))
-        l2cap_channel.on('close', lambda: self.on_l2cap_channel_close(l2cap_channel))
+        l2cap_channel.on(
+            l2cap_channel.EVENT_OPEN, lambda: self.on_l2cap_channel_open(l2cap_channel)
+        )
+        l2cap_channel.on(
+            l2cap_channel.EVENT_CLOSE,
+            lambda: self.on_l2cap_channel_close(l2cap_channel),
+        )
 
     def on_l2cap_channel_open(self, l2cap_channel: l2cap.ClassicChannel) -> None:
         if l2cap_channel.psm == HID_CONTROL_PSM:
@@ -290,7 +302,7 @@ class HID(ABC, utils.EventEmitter):
 
     def on_intr_pdu(self, pdu: bytes) -> None:
         logger.debug(f'<<< HID INTERRUPT PDU: {pdu.hex()}')
-        self.emit("interrupt_data", pdu)
+        self.emit(self.EVENT_INTERRUPT_DATA, pdu)
 
     def send_pdu_on_ctrl(self, msg: bytes) -> None:
         assert self.l2cap_ctrl_channel
@@ -363,17 +375,17 @@ class Device(HID):
             self.handle_set_protocol(pdu)
         elif message_type == Message.MessageType.DATA:
             logger.debug('<<< HID CONTROL DATA')
-            self.emit('control_data', pdu)
+            self.emit(self.EVENT_CONTROL_DATA, pdu)
         elif message_type == Message.MessageType.CONTROL:
             if param == Message.ControlCommand.SUSPEND:
                 logger.debug('<<< HID SUSPEND')
-                self.emit('suspend')
+                self.emit(self.EVENT_SUSPEND)
             elif param == Message.ControlCommand.EXIT_SUSPEND:
                 logger.debug('<<< HID EXIT SUSPEND')
-                self.emit('exit_suspend')
+                self.emit(self.EVENT_EXIT_SUSPEND)
             elif param == Message.ControlCommand.VIRTUAL_CABLE_UNPLUG:
                 logger.debug('<<< HID VIRTUAL CABLE UNPLUG')
-                self.emit('virtual_cable_unplug')
+                self.emit(self.EVENT_VIRTUAL_CABLE_UNPLUG)
             else:
                 logger.debug('<<< HID CONTROL OPERATION UNSUPPORTED')
         else:
@@ -538,14 +550,14 @@ class Host(HID):
         message_type = pdu[0] >> 4
         if message_type == Message.MessageType.HANDSHAKE:
             logger.debug(f'<<< HID HANDSHAKE: {Message.Handshake(param).name}')
-            self.emit('handshake', Message.Handshake(param))
+            self.emit(self.EVENT_HANDSHAKE, Message.Handshake(param))
         elif message_type == Message.MessageType.DATA:
             logger.debug('<<< HID CONTROL DATA')
-            self.emit('control_data', pdu)
+            self.emit(self.EVENT_CONTROL_DATA, pdu)
         elif message_type == Message.MessageType.CONTROL:
             if param == Message.ControlCommand.VIRTUAL_CABLE_UNPLUG:
                 logger.debug('<<< HID VIRTUAL CABLE UNPLUG')
-                self.emit('virtual_cable_unplug')
+                self.emit(self.EVENT_VIRTUAL_CABLE_UNPLUG)
             else:
                 logger.debug('<<< HID CONTROL OPERATION UNSUPPORTED')
         else:

bumble/host.py

@@ -435,6 +435,14 @@ class Host(utils.EventEmitter):
                 )
             )
         )
+        if self.supports_command(hci.HCI_SET_EVENT_MASK_PAGE_2_COMMAND):
+            await self.send_command(
+                hci.HCI_Set_Event_Mask_Page_2_Command(
+                    event_mask_page_2=hci.HCI_Set_Event_Mask_Page_2_Command.mask(
+                        [hci.HCI_ENCRYPTION_CHANGE_V2_EVENT]
+                    )
+                )
+            )
 
         if (
             self.local_version is not None
@@ -456,6 +464,7 @@ class Host(utils.EventEmitter):
                     hci.HCI_LE_READ_LOCAL_P_256_PUBLIC_KEY_COMPLETE_EVENT,
                     hci.HCI_LE_GENERATE_DHKEY_COMPLETE_EVENT,
                     hci.HCI_LE_ENHANCED_CONNECTION_COMPLETE_EVENT,
+                    hci.HCI_LE_ENHANCED_CONNECTION_COMPLETE_V2_EVENT,
                     hci.HCI_LE_DIRECTED_ADVERTISING_REPORT_EVENT,
                     hci.HCI_LE_PHY_UPDATE_COMPLETE_EVENT,
                     hci.HCI_LE_EXTENDED_ADVERTISING_REPORT_EVENT,
@@ -1383,6 +1392,21 @@ class Host(utils.EventEmitter):
                 'connection_encryption_change',
                 event.connection_handle,
                 event.encryption_enabled,
+                0,
+            )
+        else:
+            self.emit(
+                'connection_encryption_failure', event.connection_handle, event.status
+            )
+
+    def on_hci_encryption_change_v2_event(self, event):
+        # Notify the client
+        if event.status == hci.HCI_SUCCESS:
+            self.emit(
+                'connection_encryption_change',
+                event.connection_handle,
+                event.encryption_enabled,
+                event.encryption_key_size,
             )
         else:
             self.emit(

bumble/keys.py

@@ -22,14 +22,15 @@
 # -----------------------------------------------------------------------------
 from __future__ import annotations
 import asyncio
+import dataclasses
 import logging
 import os
 import json
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type, Any
 from typing_extensions import Self
 
 from bumble.colors import color
-from bumble.hci import Address
+from bumble import hci
 
 if TYPE_CHECKING:
     from bumble.device import Device
@@ -42,16 +43,17 @@ logger = logging.getLogger(__name__)
 
 
 # -----------------------------------------------------------------------------
+@dataclasses.dataclass
 class PairingKeys:
+    @dataclasses.dataclass
     class Key:
-        def __init__(self, value, authenticated=False, ediv=None, rand=None):
-            self.value = value
-            self.authenticated = authenticated
-            self.ediv = ediv
-            self.rand = rand
+        value: bytes
+        authenticated: bool = False
+        ediv: Optional[int] = None
+        rand: Optional[bytes] = None
 
         @classmethod
-        def from_dict(cls, key_dict):
+        def from_dict(cls, key_dict: dict[str, Any]) -> PairingKeys.Key:
             value = bytes.fromhex(key_dict['value'])
             authenticated = key_dict.get('authenticated', False)
             ediv = key_dict.get('ediv')
@@ -61,7 +63,7 @@ class PairingKeys:
 
             return cls(value, authenticated, ediv, rand)
 
-        def to_dict(self):
+        def to_dict(self) -> dict[str, Any]:
             key_dict = {'value': self.value.hex(), 'authenticated': self.authenticated}
             if self.ediv is not None:
                 key_dict['ediv'] = self.ediv
@@ -70,39 +72,42 @@ class PairingKeys:
 
             return key_dict
 
-    def __init__(self):
-        self.address_type = None
-        self.ltk = None
-        self.ltk_central = None
-        self.ltk_peripheral = None
-        self.irk = None
-        self.csrk = None
-        self.link_key = None  # Classic
+    address_type: Optional[hci.AddressType] = None
+    ltk: Optional[Key] = None
+    ltk_central: Optional[Key] = None
+    ltk_peripheral: Optional[Key] = None
+    irk: Optional[Key] = None
+    csrk: Optional[Key] = None
+    link_key: Optional[Key] = None  # Classic
+    link_key_type: Optional[int] = None  # Classic
 
-    @staticmethod
-    def key_from_dict(keys_dict, key_name):
+    @classmethod
+    def key_from_dict(cls, keys_dict: dict[str, Any], key_name: str) -> Optional[Key]:
         key_dict = keys_dict.get(key_name)
         if key_dict is None:
             return None
 
         return PairingKeys.Key.from_dict(key_dict)
 
-    @staticmethod
-    def from_dict(keys_dict):
-        keys = PairingKeys()
+    @classmethod
+    def from_dict(cls, keys_dict: dict[str, Any]) -> PairingKeys:
+        return PairingKeys(
+            address_type=(
+                hci.AddressType(t)
+                if (t := keys_dict.get('address_type')) is not None
+                else None
+            ),
+            ltk=PairingKeys.key_from_dict(keys_dict, 'ltk'),
+            ltk_central=PairingKeys.key_from_dict(keys_dict, 'ltk_central'),
+            ltk_peripheral=PairingKeys.key_from_dict(keys_dict, 'ltk_peripheral'),
+            irk=PairingKeys.key_from_dict(keys_dict, 'irk'),
+            csrk=PairingKeys.key_from_dict(keys_dict, 'csrk'),
+            link_key=PairingKeys.key_from_dict(keys_dict, 'link_key'),
+            link_key_type=keys_dict.get('link_key_type'),
+        )
 
-        keys.address_type = keys_dict.get('address_type')
-        keys.ltk = PairingKeys.key_from_dict(keys_dict, 'ltk')
-        keys.ltk_central = PairingKeys.key_from_dict(keys_dict, 'ltk_central')
-        keys.ltk_peripheral = PairingKeys.key_from_dict(keys_dict, 'ltk_peripheral')
-        keys.irk = PairingKeys.key_from_dict(keys_dict, 'irk')
-        keys.csrk = PairingKeys.key_from_dict(keys_dict, 'csrk')
-        keys.link_key = PairingKeys.key_from_dict(keys_dict, 'link_key')
-
-        return keys
-
-    def to_dict(self):
-        keys = {}
+    def to_dict(self) -> dict[str, Any]:
+        keys: dict[str, Any] = {}
 
         if self.address_type is not None:
             keys['address_type'] = self.address_type
@@ -125,9 +130,12 @@ class PairingKeys:
         if self.link_key is not None:
             keys['link_key'] = self.link_key.to_dict()
 
+        if self.link_key_type is not None:
+            keys['link_key_type'] = self.link_key_type
+
         return keys
 
-    def print(self, prefix=''):
+    def print(self, prefix: str = '') -> None:
         keys_dict = self.to_dict()
         for container_property, value in keys_dict.items():
             if isinstance(value, dict):
@@ -156,20 +164,28 @@ class KeyStore:
         all_keys = await self.get_all()
         await asyncio.gather(*(self.delete(name) for (name, _) in all_keys))
 
-    async def get_resolving_keys(self):
+    async def get_resolving_keys(self) -> list[tuple[bytes, hci.Address]]:
         all_keys = await self.get_all()
         resolving_keys = []
         for name, keys in all_keys:
             if keys.irk is not None:
-                if keys.address_type is None:
-                    address_type = Address.RANDOM_DEVICE_ADDRESS
-                else:
-                    address_type = keys.address_type
-                resolving_keys.append((keys.irk.value, Address(name, address_type)))
+                resolving_keys.append(
+                    (
+                        keys.irk.value,
+                        hci.Address(
+                            name,
+                            (
+                                keys.address_type
+                                if keys.address_type is not None
+                                else hci.Address.RANDOM_DEVICE_ADDRESS
+                            ),
+                        ),
+                    )
+                )
 
         return resolving_keys
 
-    async def print(self, prefix=''):
+    async def print(self, prefix: str = '') -> None:
         entries = await self.get_all()
         separator = ''
         for name, keys in entries:
@@ -177,8 +193,8 @@ class KeyStore:
             keys.print(prefix=prefix + '  ')
             separator = '\n'
 
-    @staticmethod
-    def create_for_device(device: Device) -> KeyStore:
+    @classmethod
+    def create_for_device(cls, device: Device) -> KeyStore:
         if device.config.keystore is None:
             return MemoryKeyStore()
 
@@ -266,9 +282,9 @@ class JsonKeyStore(KeyStore):
                     filename = params[0]
 
         # Use a namespace based on the device address
-        if device.public_address not in (Address.ANY, Address.ANY_RANDOM):
+        if device.public_address not in (hci.Address.ANY, hci.Address.ANY_RANDOM):
             namespace = str(device.public_address)
-        elif device.random_address != Address.ANY_RANDOM:
+        elif device.random_address != hci.Address.ANY_RANDOM:
             namespace = str(device.random_address)
         else:
             namespace = JsonKeyStore.DEFAULT_NAMESPACE

bumble/l2cap.py

@@ -744,6 +744,9 @@ class ClassicChannel(utils.EventEmitter):
         WAIT_FINAL_RSP = 0x16
         WAIT_CONTROL_IND = 0x17
 
+    EVENT_OPEN = "open"
+    EVENT_CLOSE = "close"
+
     connection_result: Optional[asyncio.Future[None]]
     disconnection_result: Optional[asyncio.Future[None]]
     response: Optional[asyncio.Future[bytes]]
@@ -847,7 +850,7 @@ class ClassicChannel(utils.EventEmitter):
     def abort(self) -> None:
         if self.state == self.State.OPEN:
             self._change_state(self.State.CLOSED)
-            self.emit('close')
+            self.emit(self.EVENT_CLOSE)
 
     def send_configure_request(self) -> None:
         options = L2CAP_Control_Frame.encode_configuration_options(
@@ -940,7 +943,7 @@ class ClassicChannel(utils.EventEmitter):
             if self.connection_result:
                 self.connection_result.set_result(None)
                 self.connection_result = None
-            self.emit('open')
+            self.emit(self.EVENT_OPEN)
         elif self.state == self.State.WAIT_CONFIG_REQ_RSP:
             self._change_state(self.State.WAIT_CONFIG_RSP)
 
@@ -956,7 +959,7 @@ class ClassicChannel(utils.EventEmitter):
                 if self.connection_result:
                     self.connection_result.set_result(None)
                     self.connection_result = None
-                self.emit('open')
+                self.emit(self.EVENT_OPEN)
             else:
                 logger.warning(color('invalid state', 'red'))
         elif (
@@ -991,7 +994,7 @@ class ClassicChannel(utils.EventEmitter):
                 )
             )
             self._change_state(self.State.CLOSED)
-            self.emit('close')
+            self.emit(self.EVENT_CLOSE)
             self.manager.on_channel_closed(self)
         else:
             logger.warning(color('invalid state', 'red'))
@@ -1012,7 +1015,7 @@ class ClassicChannel(utils.EventEmitter):
         if self.disconnection_result:
             self.disconnection_result.set_result(None)
             self.disconnection_result = None
-        self.emit('close')
+        self.emit(self.EVENT_CLOSE)
         self.manager.on_channel_closed(self)
 
     def __str__(self) -> str:
@@ -1047,6 +1050,9 @@ class LeCreditBasedChannel(utils.EventEmitter):
     connection: Connection
     sink: Optional[Callable[[bytes], Any]]
 
+    EVENT_OPEN = "open"
+    EVENT_CLOSE = "close"
+
     def __init__(
         self,
         manager: ChannelManager,
@@ -1098,9 +1104,9 @@ class LeCreditBasedChannel(utils.EventEmitter):
         self.state = new_state
 
         if new_state == self.State.CONNECTED:
-            self.emit('open')
+            self.emit(self.EVENT_OPEN)
         elif new_state == self.State.DISCONNECTED:
-            self.emit('close')
+            self.emit(self.EVENT_CLOSE)
 
     def send_pdu(self, pdu: Union[SupportsBytes, bytes]) -> None:
         self.manager.send_pdu(self.connection, self.destination_cid, pdu)
@@ -1381,6 +1387,8 @@ class LeCreditBasedChannel(utils.EventEmitter):
 
 # -----------------------------------------------------------------------------
 class ClassicChannelServer(utils.EventEmitter):
+    EVENT_CONNECTION = "connection"
+
     def __init__(
         self,
         manager: ChannelManager,
@@ -1395,7 +1403,7 @@ class ClassicChannelServer(utils.EventEmitter):
         self.mtu = mtu
 
     def on_connection(self, channel: ClassicChannel) -> None:
-        self.emit('connection', channel)
+        self.emit(self.EVENT_CONNECTION, channel)
         if self.handler:
             self.handler(channel)
 
@@ -1406,6 +1414,8 @@ class ClassicChannelServer(utils.EventEmitter):
 
 # -----------------------------------------------------------------------------
 class LeCreditBasedChannelServer(utils.EventEmitter):
+    EVENT_CONNECTION = "connection"
+
     def __init__(
         self,
         manager: ChannelManager,
@@ -1424,7 +1434,7 @@ class LeCreditBasedChannelServer(utils.EventEmitter):
         self.mps = mps
 
     def on_connection(self, channel: LeCreditBasedChannel) -> None:
-        self.emit('connection', channel)
+        self.emit(self.EVENT_CONNECTION, channel)
         if self.handler:
             self.handler(channel)
 

bumble/pandora/host.py

@@ -296,12 +296,12 @@ class HostService(HostServicer):
             def on_disconnection(_: None) -> None:
                 disconnection_future.set_result(None)
 
-            connection.on('disconnection', on_disconnection)
+            connection.on(connection.EVENT_DISCONNECTION, on_disconnection)
             try:
                 await disconnection_future
                 self.log.debug("Disconnected")
             finally:
-                connection.remove_listener('disconnection', on_disconnection)  # type: ignore
+                connection.remove_listener(connection.EVENT_DISCONNECTION, on_disconnection)  # type: ignore
 
         return empty_pb2.Empty()
 
@@ -383,7 +383,7 @@ class HostService(HostServicer):
                 ):
                     connections.put_nowait(connection)
 
-            self.device.on('connection', on_connection)
+            self.device.on(self.device.EVENT_CONNECTION, on_connection)
 
         try:
             # Advertise until RPC is canceled
@@ -501,7 +501,7 @@ class HostService(HostServicer):
                 ):
                     connections.put_nowait(connection)
 
-            self.device.on('connection', on_connection)
+            self.device.on(self.device.EVENT_CONNECTION, on_connection)
 
         try:
             while True:
@@ -531,7 +531,7 @@ class HostService(HostServicer):
                 await asyncio.sleep(1)
         finally:
             if request.connectable:
-                self.device.remove_listener('connection', on_connection)  # type: ignore
+                self.device.remove_listener(self.device.EVENT_CONNECTION, on_connection)  # type: ignore
 
             try:
                 self.log.debug('Stop advertising')
@@ -557,7 +557,7 @@ class HostService(HostServicer):
             scanning_phys = [int(Phy.LE_1M), int(Phy.LE_CODED)]
 
         scan_queue: asyncio.Queue[Advertisement] = asyncio.Queue()
-        handler = self.device.on('advertisement', scan_queue.put_nowait)
+        handler = self.device.on(self.device.EVENT_ADVERTISEMENT, scan_queue.put_nowait)
         await self.device.start_scanning(
             legacy=request.legacy,
             active=not request.passive,
@@ -602,7 +602,7 @@ class HostService(HostServicer):
                 yield sr
 
         finally:
-            self.device.remove_listener('advertisement', handler)  # type: ignore
+            self.device.remove_listener(self.device.EVENT_ADVERTISEMENT, handler)  # type: ignore
             try:
                 self.log.debug('Stop scanning')
                 await bumble.utils.cancel_on_event(
@@ -621,10 +621,10 @@ class HostService(HostServicer):
             Optional[Tuple[Address, int, AdvertisingData, int]]
         ] = asyncio.Queue()
         complete_handler = self.device.on(
-            'inquiry_complete', lambda: inquiry_queue.put_nowait(None)
+            self.device.EVENT_INQUIRY_COMPLETE, lambda: inquiry_queue.put_nowait(None)
         )
         result_handler = self.device.on(  # type: ignore
-            'inquiry_result',
+            self.device.EVENT_INQUIRY_RESULT,
             lambda address, class_of_device, eir_data, rssi: inquiry_queue.put_nowait(  # type: ignore
                 (address, class_of_device, eir_data, rssi)  # type: ignore
             ),
@@ -643,8 +643,8 @@ class HostService(HostServicer):
                 )
 
         finally:
-            self.device.remove_listener('inquiry_complete', complete_handler)  # type: ignore
-            self.device.remove_listener('inquiry_result', result_handler)  # type: ignore
+            self.device.remove_listener(self.device.EVENT_INQUIRY_COMPLETE, complete_handler)  # type: ignore
+            self.device.remove_listener(self.device.EVENT_INQUIRY_RESULT, result_handler)  # type: ignore
             try:
                 self.log.debug('Stop inquiry')
                 await bumble.utils.cancel_on_event(

bumble/pandora/l2cap.py

@@ -83,7 +83,7 @@ class L2CAPService(L2CAPServicer):
             close_future.set_result(None)
 
         l2cap_channel.sink = on_channel_sdu
-        l2cap_channel.on('close', on_close)
+        l2cap_channel.on(l2cap_channel.EVENT_CLOSE, on_close)
 
         return ChannelContext(close_future, sdu_queue)
 
@@ -151,7 +151,7 @@ class L2CAPService(L2CAPServicer):
                 spec=spec, handler=on_l2cap_channel
             )
         else:
-            l2cap_server.on('connection', on_l2cap_channel)
+            l2cap_server.on(l2cap_server.EVENT_CONNECTION, on_l2cap_channel)
 
         try:
             self.log.debug('Waiting for a channel connection.')

bumble/pandora/security.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 import asyncio
 import contextlib
+from collections.abc import Awaitable
 import grpc
 import logging
 
@@ -24,6 +25,7 @@ from bumble import hci
 from bumble.core import (
     PhysicalTransport,
     ProtocolError,
+    InvalidArgumentError,
 )
 import bumble.utils
 from bumble.device import Connection as BumbleConnection, Device
@@ -188,35 +190,6 @@ class PairingDelegate(BasePairingDelegate):
         self.service.event_queue.put_nowait(event)
 
 
-BR_LEVEL_REACHED: Dict[SecurityLevel, Callable[[BumbleConnection], bool]] = {
-    LEVEL0: lambda connection: True,
-    LEVEL1: lambda connection: connection.encryption == 0 or connection.authenticated,
-    LEVEL2: lambda connection: connection.encryption != 0 and connection.authenticated,
-    LEVEL3: lambda connection: connection.encryption != 0
-    and connection.authenticated
-    and connection.link_key_type
-    in (
-        hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_192_TYPE,
-        hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE,
-    ),
-    LEVEL4: lambda connection: connection.encryption
-    == hci.HCI_Encryption_Change_Event.AES_CCM
-    and connection.authenticated
-    and connection.link_key_type
-    == hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE,
-}
-
-LE_LEVEL_REACHED: Dict[LESecurityLevel, Callable[[BumbleConnection], bool]] = {
-    LE_LEVEL1: lambda connection: True,
-    LE_LEVEL2: lambda connection: connection.encryption != 0,
-    LE_LEVEL3: lambda connection: connection.encryption != 0
-    and connection.authenticated,
-    LE_LEVEL4: lambda connection: connection.encryption != 0
-    and connection.authenticated
-    and connection.sc,
-}
-
-
 class SecurityService(SecurityServicer):
     def __init__(self, device: Device, config: Config) -> None:
         self.log = utils.BumbleServerLoggerAdapter(
@@ -248,6 +221,59 @@ class SecurityService(SecurityServicer):
 
         self.device.pairing_config_factory = pairing_config_factory
 
+    async def _classic_level_reached(
+        self, level: SecurityLevel, connection: BumbleConnection
+    ) -> bool:
+        if level == LEVEL0:
+            return True
+        if level == LEVEL1:
+            return connection.encryption == 0 or connection.authenticated
+        if level == LEVEL2:
+            return connection.encryption != 0 and connection.authenticated
+
+        link_key_type: Optional[int] = None
+        if (keystore := connection.device.keystore) and (
+            keys := await keystore.get(str(connection.peer_address))
+        ):
+            link_key_type = keys.link_key_type
+        self.log.debug("link_key_type: %d", link_key_type)
+
+        if level == LEVEL3:
+            return (
+                connection.encryption != 0
+                and connection.authenticated
+                and link_key_type
+                in (
+                    hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_192_TYPE,
+                    hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE,
+                )
+            )
+        if level == LEVEL4:
+            return (
+                connection.encryption == hci.HCI_Encryption_Change_Event.AES_CCM
+                and connection.authenticated
+                and link_key_type
+                == hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE
+            )
+        raise InvalidArgumentError(f"Unexpected level {level}")
+
+    def _le_level_reached(
+        self, level: LESecurityLevel, connection: BumbleConnection
+    ) -> bool:
+        if level == LE_LEVEL1:
+            return True
+        if level == LE_LEVEL2:
+            return connection.encryption != 0
+        if level == LE_LEVEL3:
+            return connection.encryption != 0 and connection.authenticated
+        if level == LE_LEVEL4:
+            return (
+                connection.encryption != 0
+                and connection.authenticated
+                and connection.sc
+            )
+        raise InvalidArgumentError(f"Unexpected level {level}")
+
     @utils.rpc
     async def OnPairing(
         self, request: AsyncIterator[PairingEventAnswer], context: grpc.ServicerContext
@@ -290,7 +316,7 @@ class SecurityService(SecurityServicer):
         ] == oneof
 
         # security level already reached
-        if self.reached_security_level(connection, level):
+        if await self.reached_security_level(connection, level):
             return SecureResponse(success=empty_pb2.Empty())
 
         # trigger pairing if needed
@@ -302,15 +328,15 @@ class SecurityService(SecurityServicer):
 
                 with contextlib.closing(bumble.utils.EventWatcher()) as watcher:
 
-                    @watcher.on(connection, 'pairing')
+                    @watcher.on(connection, connection.EVENT_PAIRING)
                     def on_pairing(*_: Any) -> None:
                         security_result.set_result('success')
 
-                    @watcher.on(connection, 'pairing_failure')
+                    @watcher.on(connection, connection.EVENT_PAIRING_FAILURE)
                     def on_pairing_failure(*_: Any) -> None:
                         security_result.set_result('pairing_failure')
 
-                    @watcher.on(connection, 'disconnection')
+                    @watcher.on(connection, connection.EVENT_DISCONNECTION)
                     def on_disconnection(*_: Any) -> None:
                         security_result.set_result('connection_died')
 
@@ -361,7 +387,7 @@ class SecurityService(SecurityServicer):
                 return SecureResponse(encryption_failure=empty_pb2.Empty())
 
         # security level has been reached ?
-        if self.reached_security_level(connection, level):
+        if await self.reached_security_level(connection, level):
             return SecureResponse(success=empty_pb2.Empty())
         return SecureResponse(not_reached=empty_pb2.Empty())
 
@@ -388,13 +414,10 @@ class SecurityService(SecurityServicer):
         pair_task: Optional[asyncio.Future[None]] = None
 
         async def authenticate() -> None:
-            assert connection
             if (encryption := connection.encryption) != 0:
                 self.log.debug('Disable encryption...')
-                try:
+                with contextlib.suppress(Exception):
                     await connection.encrypt(enable=False)
-                except:
-                    pass
                 self.log.debug('Disable encryption: done')
 
             self.log.debug('Authenticate...')
@@ -413,15 +436,13 @@ class SecurityService(SecurityServicer):
 
             return wrapper
 
-        def try_set_success(*_: Any) -> None:
-            assert connection
-            if self.reached_security_level(connection, level):
+        async def try_set_success(*_: Any) -> None:
+            if await self.reached_security_level(connection, level):
                 self.log.debug('Wait for security: done')
                 wait_for_security.set_result('success')
 
-        def on_encryption_change(*_: Any) -> None:
-            assert connection
-            if self.reached_security_level(connection, level):
+        async def on_encryption_change(*_: Any) -> None:
+            if await self.reached_security_level(connection, level):
                 self.log.debug('Wait for security: done')
                 wait_for_security.set_result('success')
             elif (
@@ -436,7 +457,7 @@ class SecurityService(SecurityServicer):
             if self.need_pairing(connection, level):
                 pair_task = asyncio.create_task(connection.pair())
 
-        listeners: Dict[str, Callable[..., None]] = {
+        listeners: Dict[str, Callable[..., Union[None, Awaitable[None]]]] = {
             'disconnection': set_failure('connection_died'),
             'pairing_failure': set_failure('pairing_failure'),
             'connection_authentication_failure': set_failure('authentication_failure'),
@@ -455,7 +476,7 @@ class SecurityService(SecurityServicer):
                 watcher.on(connection, event, listener)
 
             # security level already reached
-            if self.reached_security_level(connection, level):
+            if await self.reached_security_level(connection, level):
                 return WaitSecurityResponse(success=empty_pb2.Empty())
 
             self.log.debug('Wait for security...')
@@ -465,24 +486,20 @@ class SecurityService(SecurityServicer):
         # wait for `authenticate` to finish if any
         if authenticate_task is not None:
             self.log.debug('Wait for authentication...')
-            try:
+            with contextlib.suppress(Exception):
                 await authenticate_task  # type: ignore
-            except:
-                pass
             self.log.debug('Authenticated')
 
         # wait for `pair` to finish if any
         if pair_task is not None:
             self.log.debug('Wait for authentication...')
-            try:
+            with contextlib.suppress(Exception):
                 await pair_task  # type: ignore
-            except:
-                pass
             self.log.debug('paired')
 
         return WaitSecurityResponse(**kwargs)
 
-    def reached_security_level(
+    async def reached_security_level(
         self, connection: BumbleConnection, level: Union[SecurityLevel, LESecurityLevel]
     ) -> bool:
         self.log.debug(
@@ -492,15 +509,14 @@ class SecurityService(SecurityServicer):
                     'encryption': connection.encryption,
                     'authenticated': connection.authenticated,
                     'sc': connection.sc,
-                    'link_key_type': connection.link_key_type,
                 }
             )
         )
 
         if isinstance(level, LESecurityLevel):
-            return LE_LEVEL_REACHED[level](connection)
+            return self._le_level_reached(level, connection)
 
-        return BR_LEVEL_REACHED[level](connection)
+        return await self._classic_level_reached(level, connection)
 
     def need_pairing(self, connection: BumbleConnection, level: int) -> bool:
         if connection.transport == PhysicalTransport.LE:

bumble/profiles/ancs.py

@@ -250,6 +250,8 @@ class AncsClient(utils.EventEmitter):
     _expected_response_tuples: int
     _response_accumulator: bytes
 
+    EVENT_NOTIFICATION = "notification"
+
     def __init__(self, ancs_proxy: AncsProxy) -> None:
         super().__init__()
         self._ancs_proxy = ancs_proxy
@@ -284,7 +286,7 @@ class AncsClient(utils.EventEmitter):
 
     def _on_notification(self, notification: Notification) -> None:
         logger.debug(f"ANCS NOTIFICATION: {notification}")
-        self.emit("notification", notification)
+        self.emit(self.EVENT_NOTIFICATION, notification)
 
     def _on_data(self, data: bytes) -> None:
         logger.debug(f"ANCS DATA: {data.hex()}")

bumble/profiles/ascs.py

@@ -276,6 +276,8 @@ class AseStateMachine(gatt.Characteristic):
         DISABLING        = 0x05
         RELEASING        = 0x06
 
+    EVENT_STATE_CHANGE = "state_change"
+
     cis_link: Optional[device.CisLink] = None
 
     # Additional parameters in CODEC_CONFIGURED State
@@ -329,8 +331,12 @@ class AseStateMachine(gatt.Characteristic):
             value=gatt.CharacteristicValue(read=self.on_read),
         )
 
-        self.service.device.on('cis_request', self.on_cis_request)
-        self.service.device.on('cis_establishment', self.on_cis_establishment)
+        self.service.device.on(
+            self.service.device.EVENT_CIS_REQUEST, self.on_cis_request
+        )
+        self.service.device.on(
+            self.service.device.EVENT_CIS_ESTABLISHMENT, self.on_cis_establishment
+        )
 
     def on_cis_request(
         self,
@@ -356,7 +362,7 @@ class AseStateMachine(gatt.Characteristic):
             and cis_link.cis_id == self.cis_id
             and self.state == self.State.ENABLING
         ):
-            cis_link.on('disconnection', self.on_cis_disconnection)
+            cis_link.on(cis_link.EVENT_DISCONNECTION, self.on_cis_disconnection)
 
             async def post_cis_established():
                 await cis_link.setup_data_path(direction=self.role)
@@ -525,7 +531,7 @@ class AseStateMachine(gatt.Characteristic):
     def state(self, new_state: State) -> None:
         logger.debug(f'{self} state change -> {colors.color(new_state.name, "cyan")}')
         self._state = new_state
-        self.emit('state_change')
+        self.emit(self.EVENT_STATE_CHANGE)
 
     @property
     def value(self):

bumble/profiles/asha.py

@@ -88,6 +88,11 @@ class AudioStatus(utils.OpenIntEnum):
 class AshaService(gatt.TemplateService):
     UUID = gatt.GATT_ASHA_SERVICE
 
+    EVENT_STARTED = "started"
+    EVENT_STOPPED = "stopped"
+    EVENT_DISCONNECTED = "disconnected"
+    EVENT_VOLUME_CHANGED = "volume_changed"
+
     audio_sink: Optional[Callable[[bytes], Any]]
     active_codec: Optional[Codec] = None
     audio_type: Optional[AudioType] = None
@@ -211,14 +216,14 @@ class AshaService(gatt.TemplateService):
                 f'volume={self.volume}, '
                 f'other_state={self.other_state}'
             )
-            self.emit('started')
+            self.emit(self.EVENT_STARTED)
         elif opcode == OpCode.STOP:
             _logger.debug('### STOP')
             self.active_codec = None
             self.audio_type = None
             self.volume = None
             self.other_state = None
-            self.emit('stopped')
+            self.emit(self.EVENT_STOPPED)
         elif opcode == OpCode.STATUS:
             _logger.debug('### STATUS: %s', PeripheralStatus(value[1]).name)
 
@@ -231,7 +236,7 @@ class AshaService(gatt.TemplateService):
                 self.audio_type = None
                 self.volume = None
                 self.other_state = None
-                self.emit('disconnected')
+                self.emit(self.EVENT_DISCONNECTED)
 
             connection.once('disconnection', on_disconnection)
 
@@ -245,7 +250,7 @@ class AshaService(gatt.TemplateService):
     def _on_volume_write(self, connection: Optional[Connection], value: bytes) -> None:
         _logger.debug(f'--- VOLUME Write:{value[0]}')
         self.volume = value[0]
-        self.emit('volume_changed')
+        self.emit(self.EVENT_VOLUME_CHANGED)
 
     # Register an L2CAP CoC server
     def _on_connection(self, channel: l2cap.LeCreditBasedChannel) -> None:

bumble/profiles/hap.py

@@ -266,13 +266,13 @@ class HearingAccessService(gatt.TemplateService):
         # associate the lowest index as the current active preset at startup
         self.active_preset_index = sorted(self.preset_records.keys())[0]
 
-        @device.on('connection')  # type: ignore
+        @device.on(device.EVENT_CONNECTION)
         def on_connection(connection: Connection) -> None:
-            @connection.on('disconnection')  # type: ignore
+            @connection.on(connection.EVENT_DISCONNECTION)
             def on_disconnection(_reason) -> None:
                 self.currently_connected_clients.remove(connection)
 
-            @connection.on('pairing')  # type: ignore
+            @connection.on(connection.EVENT_PAIRING)
             def on_pairing(*_: Any) -> None:
                 self.on_incoming_paired_connection(connection)
 

bumble/profiles/mcp.py

@@ -338,6 +338,12 @@ class MediaControlServiceProxy(
         'content_control_id': gatt.GATT_CONTENT_CONTROL_ID_CHARACTERISTIC,
     }
 
+    EVENT_MEDIA_STATE = "media_state"
+    EVENT_TRACK_CHANGED = "track_changed"
+    EVENT_TRACK_TITLE = "track_title"
+    EVENT_TRACK_DURATION = "track_duration"
+    EVENT_TRACK_POSITION = "track_position"
+
     media_player_name: Optional[gatt_client.CharacteristicProxy[bytes]] = None
     media_player_icon_object_id: Optional[gatt_client.CharacteristicProxy[bytes]] = None
     media_player_icon_url: Optional[gatt_client.CharacteristicProxy[bytes]] = None
@@ -432,20 +438,20 @@ class MediaControlServiceProxy(
         self.media_control_point_notifications.put_nowait(data)
 
     def _on_media_state(self, data: bytes) -> None:
-        self.emit('media_state', MediaState(data[0]))
+        self.emit(self.EVENT_MEDIA_STATE, MediaState(data[0]))
 
     def _on_track_changed(self, data: bytes) -> None:
         del data
-        self.emit('track_changed')
+        self.emit(self.EVENT_TRACK_CHANGED)
 
     def _on_track_title(self, data: bytes) -> None:
-        self.emit('track_title', data.decode("utf-8"))
+        self.emit(self.EVENT_TRACK_TITLE, data.decode("utf-8"))
 
     def _on_track_duration(self, data: bytes) -> None:
-        self.emit('track_duration', struct.unpack_from('<i', data)[0])
+        self.emit(self.EVENT_TRACK_DURATION, struct.unpack_from('<i', data)[0])
 
     def _on_track_position(self, data: bytes) -> None:
-        self.emit('track_position', struct.unpack_from('<i', data)[0])
+        self.emit(self.EVENT_TRACK_POSITION, struct.unpack_from('<i', data)[0])
 
 
 class GenericMediaControlServiceProxy(MediaControlServiceProxy):

bumble/profiles/vcs.py

@@ -91,6 +91,8 @@ class VolumeState:
 class VolumeControlService(gatt.TemplateService):
     UUID = gatt.GATT_VOLUME_CONTROL_SERVICE
 
+    EVENT_VOLUME_STATE_CHANGE = "volume_state_change"
+
     volume_state: gatt.Characteristic[bytes]
     volume_control_point: gatt.Characteristic[bytes]
     volume_flags: gatt.Characteristic[bytes]
@@ -166,7 +168,7 @@ class VolumeControlService(gatt.TemplateService):
                 'disconnection',
                 connection.device.notify_subscribers(attribute=self.volume_state),
             )
-            self.emit('volume_state_change')
+            self.emit(self.EVENT_VOLUME_STATE_CHANGE)
 
     def _on_relative_volume_down(self) -> bool:
         old_volume = self.volume_setting

bumble/rfcomm.py

@@ -442,6 +442,9 @@ class RFCOMM_MCC_MSC:
 
 # -----------------------------------------------------------------------------
 class DLC(utils.EventEmitter):
+    EVENT_OPEN = "open"
+    EVENT_CLOSE = "close"
+
     class State(enum.IntEnum):
         INIT = 0x00
         CONNECTING = 0x01
@@ -529,7 +532,7 @@ class DLC(utils.EventEmitter):
         self.send_frame(RFCOMM_Frame.uih(c_r=self.c_r, dlci=0, information=mcc))
 
         self.change_state(DLC.State.CONNECTED)
-        self.emit('open')
+        self.emit(self.EVENT_OPEN)
 
     def on_ua_frame(self, _frame: RFCOMM_Frame) -> None:
         if self.state == DLC.State.CONNECTING:
@@ -550,7 +553,7 @@ class DLC(utils.EventEmitter):
                 self.disconnection_result.set_result(None)
                 self.disconnection_result = None
             self.multiplexer.on_dlc_disconnection(self)
-            self.emit('close')
+            self.emit(self.EVENT_CLOSE)
         else:
             logger.warning(
                 color(
@@ -733,7 +736,7 @@ class DLC(utils.EventEmitter):
             self.disconnection_result.cancel()
             self.disconnection_result = None
         self.change_state(DLC.State.RESET)
-        self.emit('close')
+        self.emit(self.EVENT_CLOSE)
 
     def __str__(self) -> str:
         return (
@@ -763,6 +766,8 @@ class Multiplexer(utils.EventEmitter):
         DISCONNECTED = 0x05
         RESET = 0x06
 
+    EVENT_DLC = "dlc"
+
     connection_result: Optional[asyncio.Future]
     disconnection_result: Optional[asyncio.Future]
     open_result: Optional[asyncio.Future]
@@ -785,7 +790,7 @@ class Multiplexer(utils.EventEmitter):
         # Become a sink for the L2CAP channel
         l2cap_channel.sink = self.on_pdu
 
-        l2cap_channel.on('close', self.on_l2cap_channel_close)
+        l2cap_channel.on(l2cap_channel.EVENT_CLOSE, self.on_l2cap_channel_close)
 
     def change_state(self, new_state: State) -> None:
         logger.debug(f'{self} state change -> {color(new_state.name, "cyan")}')
@@ -901,7 +906,7 @@ class Multiplexer(utils.EventEmitter):
                         self.dlcs[pn.dlci] = dlc
 
                         # Re-emit the handshake completion event
-                        dlc.on('open', lambda: self.emit('dlc', dlc))
+                        dlc.on(dlc.EVENT_OPEN, lambda: self.emit(self.EVENT_DLC, dlc))
 
                         # Respond to complete the handshake
                         dlc.accept()
@@ -1076,6 +1081,8 @@ class Client:
 
 # -----------------------------------------------------------------------------
 class Server(utils.EventEmitter):
+    EVENT_START = "start"
+
     def __init__(
         self, device: Device, l2cap_mtu: int = RFCOMM_DEFAULT_L2CAP_MTU
     ) -> None:
@@ -1122,7 +1129,9 @@ class Server(utils.EventEmitter):
 
     def on_connection(self, l2cap_channel: l2cap.ClassicChannel) -> None:
         logger.debug(f'+++ new L2CAP connection: {l2cap_channel}')
-        l2cap_channel.on('open', lambda: self.on_l2cap_channel_open(l2cap_channel))
+        l2cap_channel.on(
+            l2cap_channel.EVENT_OPEN, lambda: self.on_l2cap_channel_open(l2cap_channel)
+        )
 
     def on_l2cap_channel_open(self, l2cap_channel: l2cap.ClassicChannel) -> None:
         logger.debug(f'$$$ L2CAP channel open: {l2cap_channel}')
@@ -1130,10 +1139,10 @@ class Server(utils.EventEmitter):
         # Create a new multiplexer for the channel
         multiplexer = Multiplexer(l2cap_channel, Multiplexer.Role.RESPONDER)
         multiplexer.acceptor = self.accept_dlc
-        multiplexer.on('dlc', self.on_dlc)
+        multiplexer.on(multiplexer.EVENT_DLC, self.on_dlc)
 
         # Notify
-        self.emit('start', multiplexer)
+        self.emit(self.EVENT_START, multiplexer)
 
     def accept_dlc(self, channel_number: int) -> Optional[Tuple[int, int]]:
         return self.dlc_configs.get(channel_number)

bumble/smp.py

@@ -724,12 +724,13 @@ class Session:
         self.is_responder = not self.is_initiator
 
         # Listen for connection events
-        connection.on('disconnection', self.on_disconnection)
+        connection.on(connection.EVENT_DISCONNECTION, self.on_disconnection)
         connection.on(
-            'connection_encryption_change', self.on_connection_encryption_change
+            connection.EVENT_CONNECTION_ENCRYPTION_CHANGE,
+            self.on_connection_encryption_change,
         )
         connection.on(
-            'connection_encryption_key_refresh',
+            connection.EVENT_CONNECTION_ENCRYPTION_KEY_REFRESH,
             self.on_connection_encryption_key_refresh,
         )
 
@@ -1310,12 +1311,15 @@ class Session:
         )
 
     def on_disconnection(self, _: int) -> None:
-        self.connection.remove_listener('disconnection', self.on_disconnection)
         self.connection.remove_listener(
-            'connection_encryption_change', self.on_connection_encryption_change
+            self.connection.EVENT_DISCONNECTION, self.on_disconnection
         )
         self.connection.remove_listener(
-            'connection_encryption_key_refresh',
+            self.connection.EVENT_CONNECTION_ENCRYPTION_CHANGE,
+            self.on_connection_encryption_change,
+        )
+        self.connection.remove_listener(
+            self.connection.EVENT_CONNECTION_ENCRYPTION_KEY_REFRESH,
             self.on_connection_encryption_key_refresh,
         )
         self.manager.on_session_end(self)
@@ -1376,8 +1380,10 @@ class Session:
                 ediv=self.ltk_ediv,
                 rand=self.ltk_rand,
             )
+            if not self.peer_ltk:
+                logger.error("peer_ltk is None")
             peer_ltk_key = PairingKeys.Key(
-                value=self.peer_ltk,
+                value=self.peer_ltk or b'',
                 authenticated=authenticated,
                 ediv=self.peer_ediv,
                 rand=self.peer_rand,
@@ -1962,7 +1968,7 @@ class Manager(utils.EventEmitter):
     def on_smp_security_request_command(
         self, connection: Connection, request: SMP_Security_Request_Command
     ) -> None:
-        connection.emit('security_request', request.auth_req)
+        connection.emit(connection.EVENT_SECURITY_REQUEST, request.auth_req)
 
     def on_smp_pdu(self, connection: Connection, pdu: bytes) -> None:
         # Parse the L2CAP payload into an SMP Command object

examples/run_a2dp_sink.py

@@ -33,12 +33,6 @@ from bumble.avdtp import (
 from bumble.a2dp import (
     make_audio_sink_service_sdp_records,
     A2DP_SBC_CODEC_TYPE,
-    SBC_MONO_CHANNEL_MODE,
-    SBC_DUAL_CHANNEL_MODE,
-    SBC_SNR_ALLOCATION_METHOD,
-    SBC_LOUDNESS_ALLOCATION_METHOD,
-    SBC_STEREO_CHANNEL_MODE,
-    SBC_JOINT_STEREO_CHANNEL_MODE,
     SbcMediaCodecInformation,
 )
 

examples/run_a2dp_source.py

@@ -16,28 +16,43 @@
 # Imports
 # -----------------------------------------------------------------------------
 import asyncio
-import sys
-import os
 import logging
+import os
+import sys
+from dataclasses import dataclass
 
-from bumble.colors import color
-from bumble.device import Device
-from bumble.transport import open_transport_or_link
-from bumble.core import PhysicalTransport
+import ffmpeg
+
+from bumble.a2dp import (
+    A2DP_MPEG_2_4_AAC_CODEC_TYPE,
+    A2DP_SBC_CODEC_TYPE,
+    AacMediaCodecInformation,
+    AacPacketSource,
+    SbcMediaCodecInformation,
+    SbcPacketSource,
+    make_audio_source_service_sdp_records,
+)
 from bumble.avdtp import (
-    find_avdtp_service_with_connection,
     AVDTP_AUDIO_MEDIA_TYPE,
+    Listener,
     MediaCodecCapabilities,
     MediaPacketPump,
     Protocol,
-    Listener,
-)
-from bumble.a2dp import (
-    make_audio_source_service_sdp_records,
-    A2DP_SBC_CODEC_TYPE,
-    SbcMediaCodecInformation,
-    SbcPacketSource,
+    find_avdtp_service_with_connection,
 )
+from bumble.colors import color
+from bumble.core import PhysicalTransport
+from bumble.device import Device
+from bumble.transport import open_transport_or_link
+
+from typing import Dict, Union
+
+
+@dataclass
+class CodecCapabilities:
+    name: str
+    sample_rate: str
+    number_of_channels: str
 
 
 # -----------------------------------------------------------------------------
@@ -51,67 +66,147 @@ def sdp_records():
 
 
 # -----------------------------------------------------------------------------
-def codec_capabilities():
-    # NOTE: this shouldn't be hardcoded, but should be inferred from the input file
-    # instead
-    return MediaCodecCapabilities(
-        media_type=AVDTP_AUDIO_MEDIA_TYPE,
-        media_codec_type=A2DP_SBC_CODEC_TYPE,
-        media_codec_information=SbcMediaCodecInformation(
-            sampling_frequency=SbcMediaCodecInformation.SamplingFrequency.SF_44100,
-            channel_mode=SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
-            block_length=SbcMediaCodecInformation.BlockLength.BL_16,
-            subbands=SbcMediaCodecInformation.Subbands.S_8,
-            allocation_method=SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
-            minimum_bitpool_value=2,
-            maximum_bitpool_value=53,
-        ),
-    )
-
-
-# -----------------------------------------------------------------------------
-def on_avdtp_connection(read_function, protocol):
+def on_avdtp_connection(
+    read_function, protocol, codec_capabilities: MediaCodecCapabilities
+):
     packet_source = SbcPacketSource(read_function, protocol.l2cap_channel.peer_mtu)
     packet_pump = MediaPacketPump(packet_source.packets)
-    protocol.add_source(codec_capabilities(), packet_pump)
+    protocol.add_source(codec_capabilities, packet_pump)
 
 
 # -----------------------------------------------------------------------------
-async def stream_packets(read_function, protocol):
+async def stream_packets(
+    read_function, protocol, codec_capabilities: MediaCodecCapabilities
+):
     # Discover all endpoints on the remote device
     endpoints = await protocol.discover_remote_endpoints()
     for endpoint in endpoints:
         print('@@@', endpoint)
 
     # Select a sink
+    assert codec_capabilities.media_codec_type in [
+        A2DP_SBC_CODEC_TYPE,
+        A2DP_MPEG_2_4_AAC_CODEC_TYPE,
+    ]
     sink = protocol.find_remote_sink_by_codec(
-        AVDTP_AUDIO_MEDIA_TYPE, A2DP_SBC_CODEC_TYPE
+        AVDTP_AUDIO_MEDIA_TYPE, codec_capabilities.media_codec_type
     )
     if sink is None:
-        print(color('!!! no SBC sink found', 'red'))
+        print(color('!!! no Sink found', 'red'))
         return
     print(f'### Selected sink: {sink.seid}')
 
     # Stream the packets
-    packet_source = SbcPacketSource(read_function, protocol.l2cap_channel.peer_mtu)
-    packet_pump = MediaPacketPump(packet_source.packets)
-    source = protocol.add_source(codec_capabilities(), packet_pump)
+    packet_sources = {
+        A2DP_SBC_CODEC_TYPE: SbcPacketSource(
+            read_function, protocol.l2cap_channel.peer_mtu
+        ),
+        A2DP_MPEG_2_4_AAC_CODEC_TYPE: AacPacketSource(
+            read_function, protocol.l2cap_channel.peer_mtu
+        ),
+    }
+    packet_source = packet_sources[codec_capabilities.media_codec_type]
+    packet_pump = MediaPacketPump(packet_source.packets)  # type: ignore
+    source = protocol.add_source(codec_capabilities, packet_pump)
     stream = await protocol.create_stream(source, sink)
     await stream.start()
-    await asyncio.sleep(5)
-    await stream.stop()
-    await asyncio.sleep(5)
-    await stream.start()
-    await asyncio.sleep(5)
+    await asyncio.sleep(60)
     await stream.stop()
     await stream.close()
 
 
+# -----------------------------------------------------------------------------
+def fetch_codec_informations(filepath) -> MediaCodecCapabilities:
+    probe = ffmpeg.probe(filepath)
+    assert 'streams' in probe
+    streams = probe['streams']
+
+    if not streams or len(streams) > 1:
+        print(streams)
+        print(color('!!! file not supported', 'red'))
+        exit()
+    audio_stream = streams[0]
+
+    media_codec_type = None
+    media_codec_information: Union[
+        SbcMediaCodecInformation, AacMediaCodecInformation, None
+    ] = None
+
+    assert 'codec_name' in audio_stream
+    codec_name: str = audio_stream['codec_name']
+    if codec_name == "sbc":
+        media_codec_type = A2DP_SBC_CODEC_TYPE
+        sbc_sampling_frequency: Dict[
+            str, SbcMediaCodecInformation.SamplingFrequency
+        ] = {
+            '16000': SbcMediaCodecInformation.SamplingFrequency.SF_16000,
+            '32000': SbcMediaCodecInformation.SamplingFrequency.SF_32000,
+            '44100': SbcMediaCodecInformation.SamplingFrequency.SF_44100,
+            '48000': SbcMediaCodecInformation.SamplingFrequency.SF_48000,
+        }
+        sbc_channel_mode: Dict[int, SbcMediaCodecInformation.ChannelMode] = {
+            1: SbcMediaCodecInformation.ChannelMode.MONO,
+            2: SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
+        }
+
+        assert 'sample_rate' in audio_stream
+        assert 'channels' in audio_stream
+        media_codec_information = SbcMediaCodecInformation(
+            sampling_frequency=sbc_sampling_frequency[audio_stream['sample_rate']],
+            channel_mode=sbc_channel_mode[audio_stream['channels']],
+            block_length=SbcMediaCodecInformation.BlockLength.BL_16,
+            subbands=SbcMediaCodecInformation.Subbands.S_8,
+            allocation_method=SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
+            minimum_bitpool_value=2,
+            maximum_bitpool_value=53,
+        )
+    elif codec_name == "aac":
+        media_codec_type = A2DP_MPEG_2_4_AAC_CODEC_TYPE
+        object_type: Dict[str, AacMediaCodecInformation.ObjectType] = {
+            'LC': AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC,
+            'LTP': AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LTP,
+            'SSR': AacMediaCodecInformation.ObjectType.MPEG_4_AAC_SCALABLE,
+        }
+        aac_sampling_frequency: Dict[
+            str, AacMediaCodecInformation.SamplingFrequency
+        ] = {
+            '44100': AacMediaCodecInformation.SamplingFrequency.SF_44100,
+            '48000': AacMediaCodecInformation.SamplingFrequency.SF_48000,
+        }
+        aac_channel_mode: Dict[int, AacMediaCodecInformation.Channels] = {
+            1: AacMediaCodecInformation.Channels.MONO,
+            2: AacMediaCodecInformation.Channels.STEREO,
+        }
+
+        assert 'profile' in audio_stream
+        assert 'sample_rate' in audio_stream
+        assert 'channels' in audio_stream
+        media_codec_information = AacMediaCodecInformation(
+            object_type=object_type[audio_stream['profile']],
+            sampling_frequency=aac_sampling_frequency[audio_stream['sample_rate']],
+            channels=aac_channel_mode[audio_stream['channels']],
+            vbr=1,
+            bitrate=128000,
+        )
+    else:
+        print(color('!!! codec not supported, only aac & sbc are supported', 'red'))
+        exit()
+
+    assert media_codec_type is not None
+    assert media_codec_information is not None
+
+    return MediaCodecCapabilities(
+        media_type=AVDTP_AUDIO_MEDIA_TYPE,
+        media_codec_type=media_codec_type,
+        media_codec_information=media_codec_information,
+    )
+
+
 # -----------------------------------------------------------------------------
 async def main() -> None:
     if len(sys.argv) < 4:
         print(
-            'Usage: run_a2dp_source.py <device-config> <transport-spec> <sbc-file> '
+            'Usage: run_a2dp_source.py <device-config> <transport-spec> <audio-file> '
             '[<bluetooth-address>]'
         )
         print(
@@ -135,11 +230,13 @@ async def main() -> None:
         # Start
         await device.power_on()
 
-        with open(sys.argv[3], 'rb') as sbc_file:
+        with open(sys.argv[3], 'rb') as audio_file:
             # NOTE: this should be using asyncio file reading, but blocking reads are
             # good enough for testing
             async def read(byte_count):
-                return sbc_file.read(byte_count)
+                return audio_file.read(byte_count)
+
+            codec_capabilities = fetch_codec_informations(sys.argv[3])
 
             if len(sys.argv) > 4:
                 # Connect to a peer
@@ -170,12 +267,15 @@ async def main() -> None:
                 protocol = await Protocol.connect(connection, avdtp_version)
 
                 # Start streaming
-                await stream_packets(read, protocol)
+                await stream_packets(read, protocol, codec_capabilities)
             else:
                 # Create a listener to wait for AVDTP connections
                 listener = Listener.for_device(device=device, version=(1, 2))
                 listener.on(
-                    'connection', lambda protocol: on_avdtp_connection(read, protocol)
+                    'connection',
+                    lambda protocol: on_avdtp_connection(
+                        read, protocol, codec_capabilities
+                    ),
                 )
 
                 # Become connectable and wait for a connection

extras/android/BtBench/.gitignore

@@ -1,12 +1,7 @@
 *.iml
 .gradle
 /local.properties
-/.idea/caches
-/.idea/libraries
-/.idea/modules.xml
-/.idea/workspace.xml
-/.idea/navEditor.xml
-/.idea/assetWizardSettings.xml
+.idea/
 .DS_Store
 /build
 /captures

extras/android/BtBench/app/src/main/AndroidManifest.xml

@@ -5,6 +5,7 @@
     <!-- Request legacy Bluetooth permissions on older devices. -->
     <uses-permission android:name="android.permission.BLUETOOTH" android:maxSdkVersion="30" />
     <uses-permission android:name="android.permission.BLUETOOTH_ADMIN" android:maxSdkVersion="30" />
+    <uses-permission android:name="android.permission.POST_NOTIFICATIONS" />
 
     <uses-permission android:name="android.permission.BLUETOOTH_SCAN" android:usesPermissionFlags="neverForLocation"/>
     <uses-permission android:name="android.permission.BLUETOOTH_ADVERTISE" />

extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/Advertiser.kt

@@ -6,13 +6,14 @@ import android.bluetooth.le.AdvertiseCallback
 import android.bluetooth.le.AdvertiseData
 import android.bluetooth.le.AdvertiseSettings
 import android.bluetooth.le.AdvertiseSettings.ADVERTISE_MODE_LOW_LATENCY
-import android.os.Build
+import androidx.annotation.RequiresApi
 import java.util.logging.Logger
 
 private val Log = Logger.getLogger("btbench.advertiser")
 
 class Advertiser(private val bluetoothAdapter: BluetoothAdapter) : AdvertiseCallback() {
     @SuppressLint("MissingPermission")
+    @RequiresApi(34)
     fun start() {
         val advertiseSettingsBuilder = AdvertiseSettings.Builder()
             .setAdvertiseMode(ADVERTISE_MODE_LOW_LATENCY)

extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/GattServer.kt

@@ -26,6 +26,8 @@ import android.bluetooth.BluetoothGattService
 import android.bluetooth.BluetoothManager
 import android.bluetooth.BluetoothStatusCodes
 import android.content.Context
+import android.os.Build
+import androidx.annotation.RequiresApi
 import androidx.core.content.ContextCompat
 import java.io.IOException
 import java.util.concurrent.CountDownLatch
@@ -174,6 +176,7 @@ class GattServer(
         }
     }
 
+    @RequiresApi(Build.VERSION_CODES.UPSIDE_DOWN_CAKE)
     override fun run() {
         viewModel.running = true
 

extras/android/BtBench/app/src/main/java/com/github/google/bumble/btbench/L2capServer.kt

@@ -16,14 +16,9 @@ package com.github.google.bumble.btbench
 
 import android.annotation.SuppressLint
 import android.bluetooth.BluetoothAdapter
-import android.bluetooth.le.AdvertiseCallback
-import android.bluetooth.le.AdvertiseData
-import android.bluetooth.le.AdvertiseSettings
-import android.bluetooth.le.AdvertiseSettings.ADVERTISE_MODE_LOW_LATENCY
 import android.os.Build
-import java.io.IOException
+import androidx.annotation.RequiresApi
 import java.util.logging.Logger
-import kotlin.concurrent.thread
 
 private val Log = Logger.getLogger("btbench.l2cap-server")
 
@@ -34,6 +29,7 @@ class L2capServer(
 ) : Mode {
     private var socketServer: SocketServer? = null
 
+    @RequiresApi(Build.VERSION_CODES.UPSIDE_DOWN_CAKE)
     @SuppressLint("MissingPermission")
     override fun run() {
         // Advertise so that the peer can find us and connect.

pyproject.toml

@@ -13,11 +13,11 @@ dependencies = [
     "aiohttp ~= 3.8; platform_system!='Emscripten'",
     "appdirs >= 1.4; platform_system!='Emscripten'",
     "click >= 8.1.3; platform_system!='Emscripten'",
-    "cryptography >= 39; platform_system!='Emscripten'",
+    "cryptography >= 44.0.3; platform_system!='Emscripten'",
     # Pyodide bundles a version of cryptography that is built for wasm, which may not match the
     # versions available on PyPI. Relax the version requirement since it's better than being
     # completely unable to import the package in case of version mismatch.
-    "cryptography >= 39.0; platform_system=='Emscripten'",
+    "cryptography >= 44.0.3; platform_system=='Emscripten'",
     "grpcio >= 1.62.1; platform_system!='Emscripten'",
     "humanize >= 4.6.0; platform_system!='Emscripten'",
     "libusb1 >= 2.0.1; platform_system!='Emscripten'",
@@ -26,7 +26,7 @@ dependencies = [
     "prompt_toolkit >= 3.0.16; platform_system!='Emscripten'",
     "prettytable >= 3.6.0; platform_system!='Emscripten'",
     "protobuf >= 3.12.4; platform_system!='Emscripten'",
-    "pyee >= 8.2.2",
+    "pyee >= 13.0.0",
     "pyserial-asyncio >= 0.5; platform_system!='Emscripten'",
     "pyserial >= 3.5; platform_system!='Emscripten'",
     "pyusb >= 1.2; platform_system!='Emscripten'",
@@ -55,6 +55,9 @@ development = [
     "types-invoke >= 1.7.3",
     "types-protobuf >= 4.21.0",
 ]
+examples = [
+    "ffmpeg-python == 0.2.0",
+]
 avatar = [
     "pandora-avatar == 0.0.10",
     "rootcanal == 1.11.1 ; python_version>='3.10'",
@@ -184,6 +187,10 @@ ignore_missing_imports = true
 module = "construct.*"
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "ffmpeg.*"
+ignore_missing_imports = true
+
 [[tool.mypy.overrides]]
 module = "grpc.*"
 ignore_missing_imports = true