CS commands and events

bumble/host.py

@@ -389,6 +389,12 @@ async def reset(self, driver_factory=drivers.get_driver_for_host):
                     hci.HCI_LE_TRANSMIT_POWER_REPORTING_EVENT,
                     hci.HCI_LE_BIGINFO_ADVERTISING_REPORT_EVENT,
                     hci.HCI_LE_SUBRATE_CHANGE_EVENT,
+                    hci.HCI_LE_CS_READ_REMOTE_SUPPORTED_CAPABILITIES_COMPLETE_EVENT,
+                    hci.HCI_LE_CS_PROCEDURE_ENABLE_COMPLETE_EVENT,
+                    hci.HCI_LE_CS_SECURITY_ENABLE_COMPLETE_EVENT,
+                    hci.HCI_LE_CS_CONFIG_COMPLETE_EVENT,
+                    hci.HCI_LE_CS_SUBEVENT_RESULT_EVENT,
+                    hci.HCI_LE_CS_SUBEVENT_RESULT_CONTINUE_EVENT,
                 ]
             )
 
@@ -1296,5 +1302,23 @@ def on_hci_le_read_remote_features_complete_event(self, event):
                 int.from_bytes(event.le_features, 'little'),
             )
 
+    def on_hci_le_cs_read_remote_supported_capabilities_complete_event(self, event):
+        self.emit('cs_remote_supported_capabilities', event)
+
+    def on_hci_le_cs_security_enable_complete_event(self, event):
+        self.emit('cs_security', event)
+
+    def on_hci_le_cs_config_complete_event(self, event):
+        self.emit('cs_config', event)
+
+    def on_hci_le_cs_procedure_enable_complete_event(self, event):
+        self.emit('cs_procedure', event)
+
+    def on_hci_le_cs_subevent_result_event(self, event):
+        self.emit('cs_subevent_result', event)
+
+    def on_hci_le_cs_subevent_result_continue_event(self, event):
+        self.emit('cs_subevent_result_continue', event)
+
     def on_hci_vendor_event(self, event):
         self.emit('vendor_event', event)

bumble/smp.py

@@ -1326,7 +1326,7 @@ def on_peer_key_distribution_complete(self) -> None:
         self.connection.abort_on('disconnection', self.on_pairing())
 
     def on_connection_encryption_change(self) -> None:
-        if self.connection.is_encrypted:
+        if self.connection.is_encrypted and not self.completed:
             if self.is_responder:
                 # The responder distributes its keys first, the initiator later
                 self.distribute_keys()

bumble/utils.py

@@ -447,7 +447,7 @@ def deprecated(msg: str):
     def wrapper(function):
         @functools.wraps(function)
         def inner(*args, **kwargs):
-            warnings.warn(msg, DeprecationWarning)
+            warnings.warn(msg, DeprecationWarning, stacklevel=2)
             return function(*args, **kwargs)
 
         return inner
@@ -464,7 +464,7 @@ def experimental(msg: str):
     def wrapper(function):
         @functools.wraps(function)
         def inner(*args, **kwargs):
-            warnings.warn(msg, FutureWarning)
+            warnings.warn(msg, FutureWarning, stacklevel=2)
             return function(*args, **kwargs)
 
         return inner

examples/cs_initiator.json

@@ -0,0 +1,9 @@
+{
+    "name": "Bumble CS Initiator",
+    "address": "F0:F1:F2:F3:F4:F5",
+    "advertising_interval": 100,
+    "keystore": "JsonKeyStore",
+    "irk": "865F81FF5A8B486EAAE29A27AD9F77DC",
+    "identity_address_type": 1,
+    "channel_sounding_enabled": true
+}

examples/cs_reflector.json

@@ -0,0 +1,9 @@
+{
+    "name": "Bumble CS Reflector",
+    "address": "F0:F1:F2:F3:F4:F6",
+    "advertising_interval": 100,
+    "keystore": "JsonKeyStore",
+    "irk": "0c7d74db03a1c98e7be691f76141d53d",
+    "identity_address_type": 1,
+    "channel_sounding_enabled": true
+}

examples/run_channel_sounding.py

@@ -0,0 +1,160 @@
+# Copyright 2024 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
+#
+#      https://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.
+
+# -----------------------------------------------------------------------------
+# Imports
+# -----------------------------------------------------------------------------
+from __future__ import annotations
+
+import asyncio
+import logging
+import sys
+import os
+import functools
+
+from bumble import core
+from bumble import hci
+from bumble.device import Connection, Device, ChannelSoundingCapabilities
+from bumble.transport import open_transport_or_link
+
+# From https://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Bluetooth/system/gd/hci/distance_measurement_manager.cc.
+CS_TONE_ANTENNA_CONFIG_MAPPING_TABLE = [
+    [0, 4, 5, 6],
+    [1, 7, 7, 7],
+    [2, 7, 7, 7],
+    [3, 7, 7, 7],
+]
+CS_PREFERRED_PEER_ANTENNA_MAPPING_TABLE = [1, 1, 1, 1, 3, 7, 15, 3]
+CS_ANTENNA_PERMUTATION_ARRAY = [
+    [1, 2, 3, 4],
+    [2, 1, 3, 4],
+    [1, 3, 2, 4],
+    [3, 1, 2, 4],
+    [3, 2, 1, 4],
+    [2, 3, 1, 4],
+    [1, 2, 4, 3],
+    [2, 1, 4, 3],
+    [1, 4, 2, 3],
+    [4, 1, 2, 3],
+    [4, 2, 1, 3],
+    [2, 4, 1, 3],
+    [1, 4, 3, 2],
+    [4, 1, 3, 2],
+    [1, 3, 4, 2],
+    [3, 1, 4, 2],
+    [3, 4, 1, 2],
+    [4, 3, 1, 2],
+    [4, 2, 3, 1],
+    [2, 4, 3, 1],
+    [4, 3, 2, 1],
+    [3, 4, 2, 1],
+    [3, 2, 4, 1],
+    [2, 3, 4, 1],
+]
+
+
+# -----------------------------------------------------------------------------
+async def main() -> None:
+    if len(sys.argv) < 3:
+        print(
+            'Usage: run_channel_sounding.py <config-file> <transport-spec-for-device>'
+            '[target_address](If missing, run as reflector)'
+        )
+        print('example: run_channel_sounding.py cs_reflector.json usb:0')
+        print(
+            'example: run_channel_sounding.py cs_initiator.json usb:0 F0:F1:F2:F3:F4:F5'
+        )
+        return
+
+    print('<<< connecting to HCI...')
+    async with await open_transport_or_link(sys.argv[2]) as hci_transport:
+        print('<<< connected')
+
+        device = Device.from_config_file_with_hci(
+            sys.argv[1], hci_transport.source, hci_transport.sink
+        )
+        await device.power_on()
+        assert (local_cs_capabilities := device.cs_capabilities)
+
+        if len(sys.argv) == 3:
+            await device.start_advertising(
+                own_address_type=hci.OwnAddressType.RANDOM, auto_restart=True
+            )
+
+            def on_cs_capabilities(
+                connection: Connection, capabilities: ChannelSoundingCapabilities
+            ):
+                del capabilities
+                asyncio.create_task(
+                    device.send_command(
+                        hci.HCI_LE_CS_Set_Default_Settings_Command(
+                            connection_handle=connection.handle,
+                            role_enable=(
+                                hci.CsRoleMask.INITIATOR | hci.CsRoleMask.REFLECTOR
+                            ),
+                            cs_sync_antenna_selection=0xFF,
+                            max_tx_power=0x04,
+                        ),
+                        check_result=True,
+                    )
+                )
+
+            device.on(
+                'connection',
+                lambda connection: connection.on(
+                    'channel_sounding_capabilities',
+                    functools.partial(on_cs_capabilities, connection),
+                ),
+            )
+        else:
+            target_address = hci.Address(sys.argv[3])
+
+            connection = await device.connect(
+                target_address, transport=core.BT_LE_TRANSPORT
+            )
+            if not (await device.get_long_term_key(connection.handle, b'', 0)):
+                await connection.pair()
+            await connection.encrypt()
+
+            remote_capabilities = await device.get_remote_cs_capabilities(connection)
+            await device.send_command(
+                hci.HCI_LE_CS_Set_Default_Settings_Command(
+                    connection_handle=connection.handle,
+                    role_enable=(hci.CsRoleMask.INITIATOR | hci.CsRoleMask.REFLECTOR),
+                    cs_sync_antenna_selection=0xFF,
+                    max_tx_power=0x04,
+                ),
+                check_result=True,
+            )
+            config = await device.create_cs_config(connection)
+            await device.enable_cs_security(connection)
+            tone_antenna_config_selection = CS_TONE_ANTENNA_CONFIG_MAPPING_TABLE[
+                local_cs_capabilities.num_antennas_supported - 1
+            ][remote_capabilities.num_antennas_supported - 1]
+            await device.set_cs_procedure_parameters(
+                connection=connection,
+                config=config,
+                tone_antenna_config_selection=tone_antenna_config_selection,
+                preferred_peer_antenna=CS_PREFERRED_PEER_ANTENNA_MAPPING_TABLE[
+                    tone_antenna_config_selection
+                ],
+            )
+            await device.enable_cs_procedure(connection=connection, config=config)
+
+        await hci_transport.source.terminated
+
+
+# -----------------------------------------------------------------------------
+logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
+asyncio.run(main())