Channel Sounding device handlers

bumble/hci.py

@@ -791,6 +791,27 @@ class CsSnr(OpenIntEnum):
     NOT_APPLIED = 0xFF
 
 
+class CsDoneStatus(OpenIntEnum):
+    ALL_RESULTS_COMPLETED = 0x00
+    PARTIAL = 0x01
+    ABORTED = 0x0F
+
+
+class CsProcedureAbortReason(OpenIntEnum):
+    NO_ABORT = 0x00
+    LOCAL_HOST_OR_REMOTE_REQUEST = 0x01
+    CHANNEL_MAP_UPDATE_INSTANT_PASSED = 0x02
+    UNSPECIFIED = 0x0F
+
+
+class CsSubeventAbortReason(OpenIntEnum):
+    NO_ABORT = 0x00
+    LOCAL_HOST_OR_REMOTE_REQUEST = 0x01
+    NO_CS_SYNC_RECEIVED = 0x02
+    SCHEDULING_CONFLICT_OR_LIMITED_RESOURCES = 0x03
+    UNSPECIFIED = 0x0F
+
+
 # Connection Parameters
 HCI_CONNECTION_INTERVAL_MS_PER_UNIT = 1.25
 HCI_CONNECTION_LATENCY_MS_PER_UNIT  = 1.25
@@ -6490,7 +6511,7 @@ class Action(OpenIntEnum):
         ('config_id', 1),
         ('state', 1),
         ('tone_antenna_config_selection', 1),
-        ('selected_tx_power', 1),
+        ('selected_tx_power', -1),
         ('subevent_len', 3),
         ('subevents_per_event', 1),
         ('subevent_interval', 2),
@@ -6532,7 +6553,7 @@ class State(OpenIntEnum):
         ('start_acl_conn_event_counter', 2),
         ('procedure_counter', 2),
         ('frequency_compensation', 2),
-        ('reference_power_level', 1),
+        ('reference_power_level', -1),
         ('procedure_done_status', 1),
         ('subevent_done_status', 1),
         ('abort_reason', 1),
@@ -6549,7 +6570,7 @@ class HCI_LE_CS_Subevent_Result_Event(HCI_LE_Meta_Event):
     See Bluetooth spec @ 7.7.65.44 LE CS Subevent Result event
     '''
 
-    status: int
+    connection_handle: int
     config_id: int
     start_acl_conn_event_counter: int
     procedure_counter: int
@@ -6585,7 +6606,7 @@ class HCI_LE_CS_Subevent_Result_Continue_Event(HCI_LE_Meta_Event):
     See Bluetooth spec @ 7.7.65.45 LE CS Subevent Result Continue event
     '''
 
-    status: int
+    connection_handle: int
     config_id: int
     procedure_done_status: int
     subevent_done_status: int

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,154 @@
+# 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:
+            print('<<< Start Advertising')
+            await device.start_advertising(
+                own_address_type=hci.OwnAddressType.RANDOM, auto_restart=True
+            )
+
+            def on_cs_capabilities(
+                connection: Connection, capabilities: ChannelSoundingCapabilities
+            ):
+                del capabilities
+                print('<<< Set CS Settings')
+                asyncio.create_task(device.set_default_cs_settings(connection))
+
+            device.on(
+                'connection',
+                lambda connection: connection.on(
+                    'channel_sounding_capabilities',
+                    functools.partial(on_cs_capabilities, connection),
+                ),
+            )
+        else:
+            target_address = hci.Address(sys.argv[3])
+
+            print(f'<<< Connecting to {target_address}')
+            connection = await device.connect(
+                target_address, transport=core.BT_LE_TRANSPORT
+            )
+            print('<<< ACL Connected')
+            if not (await device.get_long_term_key(connection.handle, b'', 0)):
+                print('<<< No bond, start pairing')
+                await connection.pair()
+                print('<<< Pairing complete')
+
+            print('<<< Encrypting Connection')
+            await connection.encrypt()
+
+            print('<<< Getting remote CS Capabilities...')
+            remote_capabilities = await device.get_remote_cs_capabilities(connection)
+            print('<<< Set CS Settings...')
+            await device.set_default_cs_settings(connection)
+            print('<<< Set CS Config...')
+            config = await device.create_cs_config(connection)
+            print('<<< Enable CS Security...')
+            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]
+            print('<<< Set CS Procedure Parameters...')
+            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
+                ],
+            )
+            print('<<< Enable CS Procedure...')
+            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())