auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-04-16 00:25:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

add basic support for SCI

Browse Source
This commit is contained in:
Gilles Boccon-Gibod
2026-01-10 11:11:53 -08:00
parent 4986f55043
commit c577f17c99
7 changed files with 883 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
apps/controller_info.py
View File

@@ -27,9 +27,8 @@ from bumble.core import name_or_number
from bumble.hci import (
HCI_LE_READ_BUFFER_SIZE_COMMAND,
HCI_LE_READ_BUFFER_SIZE_V2_COMMAND,
HCI_LE_READ_MAXIMUM_ADVERTISING_DATA_LENGTH_COMMAND,
HCI_LE_READ_MAXIMUM_DATA_LENGTH_COMMAND,
HCI_LE_READ_NUMBER_OF_SUPPORTED_ADVERTISING_SETS_COMMAND,
HCI_LE_READ_MINIMUM_SUPPORTED_CONNECTION_INTERVAL_COMMAND,
HCI_LE_READ_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND,
HCI_READ_BD_ADDR_COMMAND,
HCI_READ_BUFFER_SIZE_COMMAND,
@@ -37,9 +36,8 @@ from bumble.hci import (
HCI_Command,
HCI_LE_Read_Buffer_Size_Command,
HCI_LE_Read_Buffer_Size_V2_Command,
HCI_LE_Read_Maximum_Advertising_Data_Length_Command,
HCI_LE_Read_Maximum_Data_Length_Command,
HCI_LE_Read_Number_Of_Supported_Advertising_Sets_Command,
HCI_LE_Read_Minimum_Supported_Connection_Interval_Command,
HCI_LE_Read_Suggested_Default_Data_Length_Command,
HCI_Read_BD_ADDR_Command,
HCI_Read_Buffer_Size_Command,
@@ -78,52 +76,61 @@ async def get_classic_info(host: Host) -> None:
async def get_le_info(host: Host) -> None:
print()
if host.supports_command(HCI_LE_READ_NUMBER_OF_SUPPORTED_ADVERTISING_SETS_COMMAND):
response1 = await host.send_sync_command(
HCI_LE_Read_Number_Of_Supported_Advertising_Sets_Command()
)
print(
color('LE Number Of Supported Advertising Sets:', 'yellow'),
response1.num_supported_advertising_sets,
'\n',
)
print(
color('LE Number Of Supported Advertising Sets:', 'yellow'),
host.number_of_supported_advertising_sets,
'\n',
)
if host.supports_command(HCI_LE_READ_MAXIMUM_ADVERTISING_DATA_LENGTH_COMMAND):
response2 = await host.send_sync_command(
HCI_LE_Read_Maximum_Advertising_Data_Length_Command()
)
print(
color('LE Maximum Advertising Data Length:', 'yellow'),
response2.max_advertising_data_length,
'\n',
)
print(
color('LE Maximum Advertising Data Length:', 'yellow'),
host.maximum_advertising_data_length,
'\n',
)
if host.supports_command(HCI_LE_READ_MAXIMUM_DATA_LENGTH_COMMAND):
response3 = await host.send_sync_command(
response1 = await host.send_sync_command(
HCI_LE_Read_Maximum_Data_Length_Command()
)
print(
color('Maximum Data Length:', 'yellow'),
color('LE Maximum Data Length:', 'yellow'),
(
f'tx:{response3.supported_max_tx_octets}/'
f'{response3.supported_max_tx_time}, '
f'rx:{response3.supported_max_rx_octets}/'
f'{response3.supported_max_rx_time}'
f'tx:{response1.supported_max_tx_octets}/'
f'{response1.supported_max_tx_time}, '
f'rx:{response1.supported_max_rx_octets}/'
f'{response1.supported_max_rx_time}'
),
'\n',
)
if host.supports_command(HCI_LE_READ_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND):
response4 = await host.send_sync_command(
response2 = await host.send_sync_command(
HCI_LE_Read_Suggested_Default_Data_Length_Command()
)
print(
color('Suggested Default Data Length:', 'yellow'),
f'{response4.suggested_max_tx_octets}/'
f'{response4.suggested_max_tx_time}',
color('LE Suggested Default Data Length:', 'yellow'),
f'{response2.suggested_max_tx_octets}/'
f'{response2.suggested_max_tx_time}',
'\n',
)
if host.supports_command(HCI_LE_READ_MINIMUM_SUPPORTED_CONNECTION_INTERVAL_COMMAND):
response3 = await host.send_sync_command(
HCI_LE_Read_Minimum_Supported_Connection_Interval_Command()
)
print(
color('LE Minimum Supported Connection Interval:', 'yellow'),
f'{response3.minimum_supported_connection_interval * 125} µs',
)
for group in range(len(response3.group_min)):
print(
f' Group {group}: '
f'{response3.group_min[group] * 125} µs to '
f'{response3.group_max[group] * 125} µs '
'by increments of '
f'{response3.group_stride[group] * 125} µs',
'\n',
)
print(color('LE Features:', 'yellow'))
for feature in host.supported_le_features:
print(f' {LeFeature(feature).name}')

13
bumble/controller.py
View File

@@ -1898,6 +1898,19 @@ class Controller:
'''
return bytes([hci.HCI_SUCCESS]) + self.le_features.value.to_bytes(8, 'little')
def on_hci_le_read_all_local_supported_features_command(
self, _command: hci.HCI_LE_Read_All_Local_Supported_Features_Command
) -> bytes | None:
'''
See Bluetooth spec Vol 4, Part E - 7.8.128 LE Read All Local Supported Features
Command
'''
return (
bytes([hci.HCI_SUCCESS])
+ bytes([0])
+ self.le_features.value.to_bytes(248, 'little')
)
def on_hci_le_set_random_address_command(
self, command: hci.HCI_LE_Set_Random_Address_Command
) -> bytes | None:

437
bumble/device.py
View File

@@ -1409,8 +1409,8 @@ class ConnectionParametersPreferences:
connection_interval_max: float = DEVICE_DEFAULT_CONNECTION_INTERVAL_MAX
max_latency: int = DEVICE_DEFAULT_CONNECTION_MAX_LATENCY
supervision_timeout: int = DEVICE_DEFAULT_CONNECTION_SUPERVISION_TIMEOUT
min_ce_length: int = DEVICE_DEFAULT_CONNECTION_MIN_CE_LENGTH
max_ce_length: int = DEVICE_DEFAULT_CONNECTION_MAX_CE_LENGTH
min_ce_length: float = DEVICE_DEFAULT_CONNECTION_MIN_CE_LENGTH
max_ce_length: float = DEVICE_DEFAULT_CONNECTION_MAX_CE_LENGTH
ConnectionParametersPreferences.default = ConnectionParametersPreferences()
@@ -1520,7 +1520,7 @@ class _IsoLink:
self.device.host.send_iso_sdu(connection_handle=self.handle, sdu=sdu)
async def get_tx_time_stamp(self) -> tuple[int, int, int]:
response = await self.device.host.send_sync_command(
response = await self.device.send_sync_command(
hci.HCI_LE_Read_ISO_TX_Sync_Command(connection_handle=self.handle)
)
return (
@@ -1697,7 +1697,7 @@ class Connection(utils.CompositeEventEmitter):
peer_address: hci.Address
peer_name: str | None
peer_resolvable_address: hci.Address | None
peer_le_features: hci.LeFeatureMask | None
peer_le_features: hci.LeFeatureMask
role: hci.Role
parameters: Parameters
encryption: int
@@ -1750,8 +1750,8 @@ class Connection(utils.CompositeEventEmitter):
EVENT_CIS_REQUEST = "cis_request"
EVENT_CIS_ESTABLISHMENT = "cis_establishment"
EVENT_CIS_ESTABLISHMENT_FAILURE = "cis_establishment_failure"
EVENT_LE_SUBRATE_CHANGE = "le_subrate_change"
EVENT_LE_SUBRATE_CHANGE_FAILURE = "le_subrate_change_failure"
EVENT_LE_REMOTE_FEATURES_CHANGE = "le_remote_features_change"
EVENT_LE_REMOTE_FEATURES_CHANGE_FAILURE = "le_remote_features_change_failure"
@utils.composite_listener
class Listener:
@@ -1829,14 +1829,14 @@ class Connection(utils.CompositeEventEmitter):
self.authenticated = False
self.sc = False
self.att_mtu = att.ATT_DEFAULT_MTU
self.data_length = DEVICE_DEFAULT_DATA_LENGTH
self.data_length: tuple[int, int, int, int] = DEVICE_DEFAULT_DATA_LENGTH
self.gatt_client = gatt_client.Client(self) # Per-connection client
self.gatt_server = (
device.gatt_server
) # By default, use the device's shared server
self.pairing_peer_io_capability = None
self.pairing_peer_authentication_requirements = None
self.peer_le_features = None
self.peer_le_features = hci.LeFeatureMask(0)
self.cs_configs = {}
self.cs_procedures = {}
@@ -1918,16 +1918,21 @@ class Connection(utils.CompositeEventEmitter):
connection_interval_max: float,
max_latency: int,
supervision_timeout: float,
min_ce_length: float = 0.0,
max_ce_length: float = 0.0,
use_l2cap=False,
) -> None:
"""
Request an update of the connection parameters.
Request a change of the connection parameters.
For short connection intervals (below 7.5ms, introduced in Bluetooth 6.2),
use the `update_parameters_with_subrate` method instead.
Args:
connection_interval_min: Minimum interval, in milliseconds.
connection_interval_max: Maximum interval, in milliseconds.
max_latency: Latency, in number of intervals.
supervision_timeout: Timeout, in milliseconds.
max_latency: Max latency, in number of intervals.
supervision_timeout: Supervision Timeout, in milliseconds.
use_l2cap: Request the update via L2CAP.
"""
return await self.device.update_connection_parameters(
@@ -1937,6 +1942,77 @@ class Connection(utils.CompositeEventEmitter):
max_latency,
supervision_timeout,
use_l2cap=use_l2cap,
min_ce_length=min_ce_length,
max_ce_length=max_ce_length,
)
async def update_parameters_with_subrate(
self,
connection_interval_min: float,
connection_interval_max: float,
subrate_min: int,
subrate_max: int,
max_latency: int,
continuation_number: int,
supervision_timeout: float,
min_ce_length: float,
max_ce_length: float,
) -> None:
"""
Request a change of the connection parameters.
This is similar to `update_parameters` but also allows specifying
the subrate parameters and supports shorter connection intervals (below
7.5ms, as introduced in Bluetooth 6.2).
Args:
connection_interval_min: Minimum interval, in milliseconds.
connection_interval_max: Maximum interval, in milliseconds.
subrate_min: Minimum subrate factor.
subrate_max: Maximum subrate factor.
max_latency: Max latency, in number of intervals.
continuation_number: Continuation number.
supervision_timeout: Supervision Timeout, in milliseconds.
min_ce_length: Minimum connection event length, in milliseconds.
max_ce_length: Maximumsub connection event length, in milliseconds.
"""
return await self.device.update_connection_parameters_with_subrate(
self,
connection_interval_min,
connection_interval_max,
subrate_min,
subrate_max,
max_latency,
continuation_number,
supervision_timeout,
min_ce_length,
max_ce_length,
)
async def update_subrate(
self,
subrate_min: int,
subrate_max: int,
max_latency: int,
continuation_number: int,
supervision_timeout: float,
) -> None:
"""
Request request a change to the subrating factor and/or other parameters.
Args:
subrate_min: Minimum subrate factor.
subrate_max: Maximum subrate factor.
max_latency: Max latency, in number of intervals.
continuation_number: Continuation number.
supervision_timeout: Supervision Timeout, in milliseconds.
"""
return await self.device.update_connection_subrate(
self,
subrate_min,
subrate_max,
max_latency,
continuation_number,
supervision_timeout,
)
async def set_phy(
@@ -2043,6 +2119,7 @@ class DeviceConfiguration:
le_privacy_enabled: bool = False
le_rpa_timeout: int = DEVICE_DEFAULT_LE_RPA_TIMEOUT
le_subrate_enabled: bool = False
le_shorter_connection_intervals_enabled: bool = False
classic_enabled: bool = False
classic_sc_enabled: bool = True
classic_ssp_enabled: bool = True
@@ -2397,6 +2474,9 @@ class Device(utils.CompositeEventEmitter):
self.le_rpa_timeout = config.le_rpa_timeout
self.le_rpa_periodic_update_task: asyncio.Task | None = None
self.le_subrate_enabled = config.le_subrate_enabled
self.le_shorter_connection_intervals_enabled = (
config.le_shorter_connection_intervals_enabled
)
self.classic_enabled = config.classic_enabled
self.cis_enabled = config.cis_enabled
self.classic_sc_enabled = config.classic_sc_enabled
@@ -2800,7 +2880,9 @@ class Device(utils.CompositeEventEmitter):
)
)
if self.cis_enabled:
if self.cis_enabled and self.host.supports_command(
hci.HCI_LE_SET_HOST_FEATURE_COMMAND
):
await self.send_sync_command(
hci.HCI_LE_Set_Host_Feature_Command(
bit_number=hci.LeFeature.CONNECTED_ISOCHRONOUS_STREAM,
@@ -2808,7 +2890,13 @@ class Device(utils.CompositeEventEmitter):
)
)
if self.le_subrate_enabled:
if (
self.le_subrate_enabled
and self.host.supports_command(hci.HCI_LE_SET_HOST_FEATURE_COMMAND)
and self.host.supports_le_features(
hci.LeFeatureMask.CONNECTION_SUBRATING
)
):
await self.send_sync_command(
hci.HCI_LE_Set_Host_Feature_Command(
bit_number=hci.LeFeature.CONNECTION_SUBRATING_HOST_SUPPORT,
@@ -2816,7 +2904,9 @@ class Device(utils.CompositeEventEmitter):
)
)
if self.config.channel_sounding_enabled:
if self.config.channel_sounding_enabled and self.host.supports_command(
hci.HCI_LE_SET_HOST_FEATURE_COMMAND
):
await self.send_sync_command(
hci.HCI_LE_Set_Host_Feature_Command(
bit_number=hci.LeFeature.CHANNEL_SOUNDING_HOST_SUPPORT,
@@ -2849,6 +2939,20 @@ class Device(utils.CompositeEventEmitter):
tx_snr_capability=result.tx_snr_capability,
)
if (
self.le_shorter_connection_intervals_enabled
and self.host.supports_command(hci.HCI_LE_SET_HOST_FEATURE_COMMAND)
and self.host.supports_le_features(
hci.LeFeatureMask.SHORTER_CONNECTION_INTERVALS
)
):
await self.send_sync_command(
hci.HCI_LE_Set_Host_Feature_Command(
bit_number=hci.LeFeature.SHORTER_CONNECTION_INTERVALS_HOST_SUPPORT,
bit_value=1,
)
)
if self.classic_enabled:
await self.send_sync_command(
hci.HCI_Write_Local_Name_Command(local_name=self.name.encode('utf8')),
@@ -4180,6 +4284,9 @@ class Device(utils.CompositeEventEmitter):
'''
Request an update of the connection parameters.
For short connection intervals (below 7.5 ms, introduced in Bluetooth 6.2),
use `update_connection_parameters_with_subrate` instead.
Args:
connection: The connection to update
connection_interval_min: Minimum interval, in milliseconds.
@@ -4220,17 +4327,148 @@ class Device(utils.CompositeEventEmitter):
return
await self.send_async_command(
hci.HCI_LE_Connection_Update_Command(
connection_handle=connection.handle,
connection_interval_min=connection_interval_min,
connection_interval_max=connection_interval_max,
max_latency=max_latency,
supervision_timeout=supervision_timeout,
min_ce_length=min_ce_length,
max_ce_length=max_ce_length,
pending_result = asyncio.get_running_loop().create_future()
with closing(utils.EventWatcher()) as watcher:
@watcher.on(connection, connection.EVENT_CONNECTION_PARAMETERS_UPDATE)
def _():
pending_result.set_result(None)
@watcher.on(
connection, connection.EVENT_CONNECTION_PARAMETERS_UPDATE_FAILURE
)
)
def _(error_code: int):
pending_result.set_exception(hci.HCI_Error(error_code))
await self.send_async_command(
hci.HCI_LE_Connection_Update_Command(
connection_handle=connection.handle,
connection_interval_min=connection_interval_min,
connection_interval_max=connection_interval_max,
max_latency=max_latency,
supervision_timeout=supervision_timeout,
min_ce_length=min_ce_length,
max_ce_length=max_ce_length,
)
)
await connection.cancel_on_disconnection(pending_result)
async def update_connection_parameters_with_subrate(
self,
connection: Connection,
connection_interval_min: float,
connection_interval_max: float,
subrate_min: int,
subrate_max: int,
max_latency: int,
continuation_number: int,
supervision_timeout: float,
min_ce_length: float = 0.0,
max_ce_length: float = 0.0,
) -> None:
'''
Request a change of the connection parameters.
This is similar to `update_connection_parameters` but also allows specifying
the subrate parameters and supports shorter connection intervals (below
7.5ms, as introduced in Bluetooth 6.2).
Args:
connection: The connection to update
connection_interval_min: Minimum interval, in milliseconds.
connection_interval_max: Maximum interval, in milliseconds.
subrate_min: Minimum subrate factor.
subrate_max: Maximum subrate factor.
max_latency: Max latency, in number of intervals.
continuation_number: Continuation number.
supervision_timeout: Supervision Timeout, in milliseconds.
min_ce_length: Minimum connection event length, in milliseconds.
max_ce_length: Maximum connection event length, in milliseconds.
'''
# Convert the input parameters
connection_interval_min = int(connection_interval_min / 0.125)
connection_interval_max = int(connection_interval_max / 0.125)
supervision_timeout = int(supervision_timeout / 10)
min_ce_length = int(min_ce_length / 0.125)
max_ce_length = int(max_ce_length / 0.125)
pending_result = asyncio.get_running_loop().create_future()
with closing(utils.EventWatcher()) as watcher:
@watcher.on(connection, connection.EVENT_CONNECTION_PARAMETERS_UPDATE)
def _():
pending_result.set_result(None)
@watcher.on(
connection, connection.EVENT_CONNECTION_PARAMETERS_UPDATE_FAILURE
)
def _(error_code: int):
pending_result.set_exception(hci.HCI_Error(error_code))
await self.send_async_command(
hci.HCI_LE_Connection_Rate_Request_Command(
connection_handle=connection.handle,
connection_interval_min=connection_interval_min,
connection_interval_max=connection_interval_max,
subrate_min=subrate_min,
subrate_max=subrate_max,
max_latency=max_latency,
continuation_number=continuation_number,
supervision_timeout=supervision_timeout,
min_ce_length=min_ce_length,
max_ce_length=max_ce_length,
)
)
await connection.cancel_on_disconnection(pending_result)
async def update_connection_subrate(
self,
connection: Connection,
subrate_min: int,
subrate_max: int,
max_latency: int,
continuation_number: int,
supervision_timeout: float,
) -> None:
'''
Request a change to the subrating factor and/or other parameters.
Args:
connection: The connection to update
subrate_min: Minimum subrate factor.
subrate_max: Maximum subrate factor.
max_latency: Max latency, in number of intervals.
continuation_number: Continuation number.
supervision_timeout: Supervision Timeout, in milliseconds.
'''
pending_result = asyncio.get_running_loop().create_future()
with closing(utils.EventWatcher()) as watcher:
@watcher.on(connection, connection.EVENT_CONNECTION_PARAMETERS_UPDATE)
def _():
pending_result.set_result(None)
@watcher.on(
connection, connection.EVENT_CONNECTION_PARAMETERS_UPDATE_FAILURE
)
def _(error_code: int):
pending_result.set_exception(hci.HCI_Error(error_code))
await self.send_async_command(
hci.HCI_LE_Subrate_Request_Command(
connection_handle=connection.handle,
subrate_min=subrate_min,
subrate_max=subrate_max,
max_latency=max_latency,
continuation_number=continuation_number,
supervision_timeout=int(supervision_timeout / 10),
)
)
await connection.cancel_on_disconnection(pending_result)
async def get_connection_rssi(self, connection):
result = await self.send_sync_command(
@@ -4289,6 +4527,87 @@ class Device(utils.CompositeEventEmitter):
)
)
async def set_default_connection_subrate(
self,
subrate_min: int,
subrate_max: int,
max_latency: int,
continuation_number: int,
supervision_timeout: float,
) -> None:
'''
Set the default subrate parameters for new connections.
Args:
subrate_min: Minimum subrate factor.
subrate_max: Maximum subrate factor.
max_latency: Max latency, in number of intervals.
continuation_number: Continuation number.
supervision_timeout: Supervision Timeout, in milliseconds.
'''
# Convert the input parameters
supervision_timeout = int(supervision_timeout / 10)
await self.send_command(
hci.HCI_LE_Set_Default_Subrate_Command(
subrate_min=subrate_min,
subrate_max=subrate_max,
max_latency=max_latency,
continuation_number=continuation_number,
supervision_timeout=supervision_timeout,
),
check_result=True,
)
async def set_default_connection_parameters(
self,
connection_interval_min: float,
connection_interval_max: float,
subrate_min: int,
subrate_max: int,
max_latency: int,
continuation_number: int,
supervision_timeout: float,
min_ce_length: float = 0.0,
max_ce_length: float = 0.0,
) -> None:
'''
Set the default connection parameters for new connections.
Args:
connection_interval_min: Minimum interval, in milliseconds.
connection_interval_max: Maximum interval, in milliseconds.
subrate_min: Minimum subrate factor.
subrate_max: Maximum subrate factor.
max_latency: Max latency, in number of intervals.
continuation_number: Continuation number.
supervision_timeout: Supervision Timeout, in milliseconds.
min_ce_length: Minimum connection event length, in milliseconds.
max_ce_length: Maximum connection event length, in milliseconds.
'''
# Convert the input parameters
connection_interval_min = int(connection_interval_min / 0.125)
connection_interval_max = int(connection_interval_max / 0.125)
supervision_timeout = int(supervision_timeout / 10)
min_ce_length = int(min_ce_length / 0.125)
max_ce_length = int(max_ce_length / 0.125)
await self.send_sync_command(
hci.HCI_LE_Set_Default_Rate_Parameters_Command(
connection_interval_min=connection_interval_min,
connection_interval_max=connection_interval_max,
subrate_min=subrate_min,
subrate_max=subrate_max,
max_latency=max_latency,
continuation_number=continuation_number,
supervision_timeout=supervision_timeout,
min_ce_length=min_ce_length,
max_ce_length=max_ce_length,
)
)
async def transfer_periodic_sync(
self, connection: Connection, sync_handle: int, service_data: int = 0
) -> None:
@@ -4311,7 +4630,9 @@ class Device(utils.CompositeEventEmitter):
)
)
async def find_peer_by_name(self, name: str, transport=PhysicalTransport.LE):
async def find_peer_by_name(
self, name: str, transport=PhysicalTransport.LE
) -> hci.Address:
"""
Scan for a peer with a given name and return its address.
"""
@@ -4329,6 +4650,7 @@ class Device(utils.CompositeEventEmitter):
listener: Callable[..., None] | None = None
was_scanning = self.scanning
was_discovering = self.discovering
event_name: str | None = None
try:
if transport == PhysicalTransport.LE:
event_name = 'advertisement'
@@ -4354,11 +4676,11 @@ class Device(utils.CompositeEventEmitter):
if not self.discovering:
await self.start_discovery()
else:
return None
raise ValueError('invalid transport')
return await utils.cancel_on_event(self, Device.EVENT_FLUSH, peer_address)
finally:
if listener is not None:
if listener is not None and event_name is not None:
self.remove_listener(event_name, listener)
if transport == PhysicalTransport.LE and not was_scanning:
@@ -4384,7 +4706,7 @@ class Device(utils.CompositeEventEmitter):
peer_address.set_result(address)
return
if address.is_resolvable:
if address.is_resolvable and self.address_resolver is not None:
resolved_address = self.address_resolver.resolve(address)
if resolved_address == identity_address:
if not peer_address.done():
@@ -4599,7 +4921,6 @@ class Device(utils.CompositeEventEmitter):
# [Classic only]
async def request_remote_name(self, remote: hci.Address | Connection) -> str:
# Set up event handlers
pending_name: asyncio.Future[str] = asyncio.get_running_loop().create_future()
peer_address = (
@@ -6186,7 +6507,7 @@ class Device(utils.CompositeEventEmitter):
):
logger.debug(
f'*** Connection Parameters Update: [0x{connection.handle:04X}] '
f'{connection.peer_address} as {connection.role_name}, '
f'{connection.peer_address} as {connection.role_name}'
)
if connection.parameters.connection_interval != connection_interval * 1.25:
connection.parameters = Connection.Parameters(
@@ -6210,7 +6531,41 @@ class Device(utils.CompositeEventEmitter):
self, connection: Connection, error: int
):
logger.debug(
f'*** Connection Parameters Update Failed: [0x{connection.handle:04X}] '
f'*** Connection Parameters Update failed: [0x{connection.handle:04X}] '
f'{connection.peer_address} as {connection.role_name}, '
f'error={error}'
)
connection.emit(connection.EVENT_CONNECTION_PARAMETERS_UPDATE_FAILURE, error)
@host_event_handler
@with_connection_from_handle
def on_le_connection_rate_change(
self,
connection: Connection,
connection_interval: int,
subrate_factor: int,
peripheral_latency: int,
continuation_number: int,
supervision_timeout: int,
):
logger.debug(
f'*** Connection Rate Change: [0x{connection.handle:04X}] '
f'{connection.peer_address} as {connection.role_name}'
)
connection.parameters = Connection.Parameters(
connection_interval=connection_interval * 0.125,
subrate_factor=subrate_factor,
peripheral_latency=peripheral_latency,
continuation_number=continuation_number,
supervision_timeout=supervision_timeout * 10.0,
)
connection.emit(connection.EVENT_CONNECTION_PARAMETERS_UPDATE)
@host_event_handler
@with_connection_from_handle
def on_le_connection_rate_change_failure(self, connection: Connection, error: int):
logger.debug(
f'*** Connection Rate Change failed: [0x{connection.handle:04X}] '
f'{connection.peer_address} as {connection.role_name}, '
f'error={error}'
)
@@ -6253,7 +6608,25 @@ class Device(utils.CompositeEventEmitter):
subrate_factor,
continuation_number,
)
connection.emit(connection.EVENT_LE_SUBRATE_CHANGE)
connection.emit(connection.EVENT_CONNECTION_PARAMETERS_UPDATE)
@host_event_handler
@with_connection_from_handle
def on_le_subrate_change_failure(self, connection: Connection, status: int):
connection.emit(connection.EVENT_CONNECTION_PARAMETERS_UPDATE_FAILURE, status)
@host_event_handler
@with_connection_from_handle
def on_le_remote_features(
self, connection: Connection, le_features: hci.LeFeatureMask
):
connection.peer_le_features = le_features
connection.emit(connection.EVENT_LE_REMOTE_FEATURES_CHANGE)
@host_event_handler
@with_connection_from_handle
def on_le_remote_features_failure(self, connection: Connection, status: int):
connection.emit(connection.EVENT_LE_REMOTE_FEATURES_CHANGE_FAILURE, status)
@host_event_handler
@with_connection_from_handle

161
bumble/hci.py
View File

@@ -398,8 +398,9 @@ HCI_LE_CS_SUBEVENT_RESULT_EVENT = 0x31
HCI_LE_CS_SUBEVENT_RESULT_CONTINUE_EVENT = 0x32
HCI_LE_CS_TEST_END_COMPLETE_EVENT = 0x33
HCI_LE_MONITORED_ADVERTISERS_REPORT_EVENT = 0x34
HCI_LE_FRAME_SPACE_UPDATE_EVENT = 0x35
HCI_LE_FRAME_SPACE_UPDATE_COMPLETE_EVENT = 0x35
HCI_LE_UTP_RECEIVE_EVENT = 0x36
HCI_LE_CONNECTION_RATE_CHANGE_EVENT = 0x37
# HCI Command
@@ -736,6 +737,12 @@ HCI_LE_CLEAR_MONITORED_ADVERTISERS_LIST_COMMAND = hci_c
HCI_LE_READ_MONITORED_ADVERTISERS_LIST_SIZE_COMMAND = hci_command_op_code(0x08, 0x009B)
HCI_LE_ENABLE_MONITORING_ADVERTISERS_COMMAND = hci_command_op_code(0x08, 0x009C)
HCI_LE_FRAME_SPACE_UPDATE_COMMAND = hci_command_op_code(0x08, 0x009D)
HCI_LE_SET_RESOLVABLE_PRIVATE_ADDRESS_TIMEOUT_V2_COMMAND = hci_command_op_code(0x08, 0x009E)
HCI_LE_ENABLE_UTP_OTA_MODE_COMMAND = hci_command_op_code(0x08, 0x009F)
HCI_LE_UTP_SEND_COMMAND = hci_command_op_code(0x08, 0x00A0)
HCI_LE_CONNECTION_RATE_REQUEST_COMMAND = hci_command_op_code(0x08, 0x00A1)
HCI_LE_SET_DEFAULT_RATE_PARAMETERS_COMMAND = hci_command_op_code(0x08, 0x00A2)
HCI_LE_READ_MINIMUM_SUPPORTED_CONNECTION_INTERVAL_COMMAND = hci_command_op_code(0x08, 0x00A3)
# HCI Error Codes
@@ -1398,6 +1405,12 @@ HCI_SUPPORTED_COMMANDS_MASKS = {
HCI_LE_CLEAR_MONITORED_ADVERTISERS_LIST_COMMAND : 1 << (47*8+7),
HCI_LE_READ_MONITORED_ADVERTISERS_LIST_SIZE_COMMAND : 1 << (48*8+0),
HCI_LE_FRAME_SPACE_UPDATE_COMMAND : 1 << (48*8+1),
HCI_LE_SET_RESOLVABLE_PRIVATE_ADDRESS_TIMEOUT_V2_COMMAND : 1 << (48*8+2),
HCI_LE_ENABLE_UTP_OTA_MODE_COMMAND : 1 << (48*8+3),
HCI_LE_UTP_SEND_COMMAND : 1 << (48*8+4),
HCI_LE_CONNECTION_RATE_REQUEST_COMMAND : 1 << (48*8+5),
HCI_LE_SET_DEFAULT_RATE_PARAMETERS_COMMAND : 1 << (48*8+6),
HCI_LE_READ_MINIMUM_SUPPORTED_CONNECTION_INTERVAL_COMMAND : 1 << (48*8+7)
}
# LE Supported Features
@@ -1455,6 +1468,14 @@ class LeFeature(SpecableEnum):
LL_EXTENDED_FEATURE_SET = 63
MONITORING_ADVERTISERS = 64
FRAME_SPACE_UPDATE = 65
UTP_OTA_MODE = 66
UTP_HCI_MODE = 67
LL_OTA_UTP_IND_MAXIMUM_LENGTH_0 = 68
LL_OTA_UTP_IND_MAXIMUM_LENGTH_1 = 69
SHORTER_CONNECTION_INTERVALS = 72
SHORTER_CONNECTION_INTERVALS_HOST_SUPPORT = 73
LE_FLUSHABLE_ACL_DATA = 74
class LeFeatureMask(utils.CompatibleIntFlag):
LE_ENCRYPTION = 1 << LeFeature.LE_ENCRYPTION
@@ -1509,6 +1530,13 @@ class LeFeatureMask(utils.CompatibleIntFlag):
LL_EXTENDED_FEATURE_SET = 1 << LeFeature.LL_EXTENDED_FEATURE_SET
MONITORING_ADVERTISERS = 1 << LeFeature.MONITORING_ADVERTISERS
FRAME_SPACE_UPDATE = 1 << LeFeature.FRAME_SPACE_UPDATE
UTP_OTA_MODE = 1 << LeFeature.UTP_OTA_MODE
UTP_HCI_MODE = 1 << LeFeature.UTP_HCI_MODE
LL_OTA_UTP_IND_MAXIMUM_LENGTH_0 = 1 << LeFeature.LL_OTA_UTP_IND_MAXIMUM_LENGTH_0
LL_OTA_UTP_IND_MAXIMUM_LENGTH_1 = 1 << LeFeature.LL_OTA_UTP_IND_MAXIMUM_LENGTH_1
SHORTER_CONNECTION_INTERVALS = 1 << LeFeature.SHORTER_CONNECTION_INTERVALS
SHORTER_CONNECTION_INTERVALS_HOST_SUPPORT = 1 << LeFeature.SHORTER_CONNECTION_INTERVALS_HOST_SUPPORT
LE_FLUSHABLE_ACL_DATA = 1 << LeFeature.LE_FLUSHABLE_ACL_DATA
class LmpFeature(SpecableEnum):
# Page 0 (Legacy LMP features)
@@ -3746,7 +3774,7 @@ class HCI_Write_Extended_Inquiry_Response_Command(
'''
fec_required: int = field(metadata=metadata(1))
extended_inquiry_response: int = field(
extended_inquiry_response: bytes = field(
metadata=metadata({'size': 240, 'serializer': lambda x: padded_bytes(x, 240)})
)
@@ -5796,7 +5824,25 @@ class HCI_LE_Subrate_Request_Command(HCI_AsyncCommand):
# -----------------------------------------------------------------------------
@dataclasses.dataclass
class HHCI_LE_CS_Read_Local_Supported_Capabilities_ReturnParameters(
class HCI_LE_Read_All_Local_Supported_Features_ReturnParameters(
HCI_StatusReturnParameters
):
max_page: int = field(metadata=metadata(1))
le_features: bytes = field(metadata=metadata(248))
@HCI_SyncCommand.sync_command(HCI_LE_Read_All_Local_Supported_Features_ReturnParameters)
class HCI_LE_Read_All_Local_Supported_Features_Command(
HCI_SyncCommand[HCI_LE_Read_All_Local_Supported_Features_ReturnParameters]
):
'''
See Bluetooth spec @ 7.8.128 LE Read All Local Supported Features Command
'''
# -----------------------------------------------------------------------------
@dataclasses.dataclass
class HCI_LE_CS_Read_Local_Supported_Capabilities_ReturnParameters(
HCI_StatusReturnParameters
):
num_config_supported: int = field(metadata=metadata(1))
@@ -5822,11 +5868,11 @@ class HHCI_LE_CS_Read_Local_Supported_Capabilities_ReturnParameters(
@HCI_SyncCommand.sync_command(
HHCI_LE_CS_Read_Local_Supported_Capabilities_ReturnParameters
HCI_LE_CS_Read_Local_Supported_Capabilities_ReturnParameters
)
@dataclasses.dataclass
class HCI_LE_CS_Read_Local_Supported_Capabilities_Command(
HCI_SyncCommand[HHCI_LE_CS_Read_Local_Supported_Capabilities_ReturnParameters]
HCI_SyncCommand[HCI_LE_CS_Read_Local_Supported_Capabilities_ReturnParameters]
):
'''
See Bluetooth spec @ 7.8.130 LE CS Read Local Supported Capabilities command
@@ -6073,6 +6119,92 @@ class HCI_LE_CS_Test_End_Command(HCI_AsyncCommand):
'''
# -----------------------------------------------------------------------------
@HCI_Command.command
@dataclasses.dataclass
class HCI_LE_Frame_Space_Update_Command(HCI_AsyncCommand):
'''
See Bluetooth spec @ 7.8.151 LE Frame Space Update command
'''
class SpacingType(SpecableFlag):
T_IFS_ACL_CP = 1 << 0
T_IFS_ACL_PC = 1 << 1
T_MCES = 1 << 2
T_IFS_CIS = 1 << 3
T_MSS_CIS = 1 << 4
connection_handle: int = field(metadata=metadata(2))
frame_space_min: int = field(metadata=metadata(2))
frame_space_max: int = field(metadata=metadata(2))
phys: int = field(metadata=PhyBit.type_metadata(1))
spacing_types: int = field(metadata=SpacingType.type_metadata(1))
# -----------------------------------------------------------------------------
@HCI_Command.command
@dataclasses.dataclass
class HCI_LE_Connection_Rate_Request_Command(HCI_AsyncCommand):
'''
See Bluetooth spec @ 7.8.154 LE Connection Rate Request command
'''
connection_handle: int = field(metadata=metadata(2))
connection_interval_min: int = field(metadata=metadata(2))
connection_interval_max: int = field(metadata=metadata(2))
subrate_min: int = field(metadata=metadata(2))
subrate_max: int = field(metadata=metadata(2))
max_latency: int = field(metadata=metadata(2))
continuation_number: int = field(metadata=metadata(2))
supervision_timeout: int = field(metadata=metadata(2))
min_ce_length: int = field(metadata=metadata(2))
max_ce_length: int = field(metadata=metadata(2))
# -----------------------------------------------------------------------------
@HCI_SyncCommand.sync_command(HCI_StatusReturnParameters)
@dataclasses.dataclass
class HCI_LE_Set_Default_Rate_Parameters_Command(
HCI_SyncCommand[HCI_StatusReturnParameters]
):
'''
See Bluetooth spec @ 7.8.155 LE Set Default Rate Parameters command
'''
connection_interval_min: int = field(metadata=metadata(2))
connection_interval_max: int = field(metadata=metadata(2))
subrate_min: int = field(metadata=metadata(2))
subrate_max: int = field(metadata=metadata(2))
max_latency: int = field(metadata=metadata(2))
continuation_number: int = field(metadata=metadata(2))
supervision_timeout: int = field(metadata=metadata(2))
min_ce_length: int = field(metadata=metadata(2))
max_ce_length: int = field(metadata=metadata(2))
# -----------------------------------------------------------------------------
@dataclasses.dataclass
class HCI_LE_Read_Minimum_Supported_Connection_Interval_ReturnParameters(
HCI_StatusReturnParameters
):
minimum_supported_connection_interval: int = field(metadata=metadata(1))
group_min: Sequence[int] = field(metadata=metadata(2, list_begin=True))
group_max: Sequence[int] = field(metadata=metadata(2))
group_stride: Sequence[int] = field(metadata=metadata(2, list_end=True))
@HCI_SyncCommand.sync_command(
HCI_LE_Read_Minimum_Supported_Connection_Interval_ReturnParameters
)
@dataclasses.dataclass
class HCI_LE_Read_Minimum_Supported_Connection_Interval_Command(
HCI_SyncCommand[HCI_LE_Read_Minimum_Supported_Connection_Interval_ReturnParameters]
):
'''
See Bluetooth spec @ 7.8.156 LE Read Minimum Supported Connection Interval command
'''
# -----------------------------------------------------------------------------
# HCI Events
# -----------------------------------------------------------------------------
@@ -7142,6 +7274,23 @@ class HCI_LE_CS_Test_End_Complete_Event(HCI_LE_Meta_Event):
status: int = field(metadata=metadata(STATUS_SPEC))
# -----------------------------------------------------------------------------
@HCI_LE_Meta_Event.event
@dataclasses.dataclass
class HCI_LE_Connection_Rate_Change_Event(HCI_LE_Meta_Event):
'''
See Bluetooth spec @ 7.7.65.50 LE Connection Rate Change event
'''
status: int = field(metadata=metadata(STATUS_SPEC))
connection_handle: int = field(metadata=metadata(2))
connection_interval: int = field(metadata=metadata(2))
subrate_factor: int = field(metadata=metadata(2))
peripheral_latency: int = field(metadata=metadata(2))
continuation_number: int = field(metadata=metadata(2))
supervision_timeout: int = field(metadata=metadata(2))
# -----------------------------------------------------------------------------
@HCI_Event.event
@dataclasses.dataclass

88
bumble/host.py
View File

@@ -21,7 +21,6 @@ import asyncio
import collections
import dataclasses
import logging
import struct
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING, Any, TypeVar, cast, overload
@@ -278,7 +277,7 @@ class Host(utils.EventEmitter):
hci.HCI_Read_Local_Version_Information_ReturnParameters | None
) = None
self.local_supported_commands = 0
self.local_le_features = 0
self.local_le_features = hci.LeFeatureMask(0) # LE features
self.local_lmp_features = hci.LmpFeatureMask(0) # Classic LMP features
self.suggested_max_tx_octets = 251 # Max allowed
self.suggested_max_tx_time = 2120 # Max allowed
@@ -348,17 +347,26 @@ class Host(utils.EventEmitter):
response1.supported_commands, 'little'
)
if self.supports_command(hci.HCI_LE_READ_LOCAL_SUPPORTED_FEATURES_COMMAND):
response2 = await self.send_sync_command(
hci.HCI_LE_Read_Local_Supported_Features_Command()
)
self.local_le_features = struct.unpack('<Q', response2.le_features)[0]
if self.supports_command(hci.HCI_READ_LOCAL_VERSION_INFORMATION_COMMAND):
self.local_version = await self.send_sync_command(
hci.HCI_Read_Local_Version_Information_Command()
)
if self.supports_command(hci.HCI_LE_READ_ALL_LOCAL_SUPPORTED_FEATURES_COMMAND):
response2 = await self.send_sync_command(
hci.HCI_LE_Read_All_Local_Supported_Features_Command()
)
self.local_le_features = hci.LeFeatureMask(
int.from_bytes(response2.le_features, 'little')
)
elif self.supports_command(hci.HCI_LE_READ_LOCAL_SUPPORTED_FEATURES_COMMAND):
response3 = await self.send_sync_command(
hci.HCI_LE_Read_Local_Supported_Features_Command()
)
self.local_le_features = hci.LeFeatureMask(
int.from_bytes(response3.le_features, 'little')
)
if self.supports_command(hci.HCI_READ_LOCAL_EXTENDED_FEATURES_COMMAND):
max_page_number = 0
page_number = 0
@@ -375,7 +383,6 @@ class Host(utils.EventEmitter):
max_page_number = response4.maximum_page_number
page_number += 1
self.local_lmp_features = hci.LmpFeatureMask(lmp_features)
elif self.supports_command(hci.HCI_READ_LOCAL_SUPPORTED_FEATURES_COMMAND):
response5 = await self.send_sync_command(
hci.HCI_Read_Local_Supported_Features_Command()
@@ -494,12 +501,17 @@ class Host(utils.EventEmitter):
hci.HCI_LE_TRANSMIT_POWER_REPORTING_EVENT,
hci.HCI_LE_BIGINFO_ADVERTISING_REPORT_EVENT,
hci.HCI_LE_SUBRATE_CHANGE_EVENT,
hci.HCI_LE_READ_ALL_REMOTE_FEATURES_COMPLETE_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,
hci.HCI_LE_MONITORED_ADVERTISERS_REPORT_EVENT,
hci.HCI_LE_FRAME_SPACE_UPDATE_COMPLETE_EVENT,
hci.HCI_LE_UTP_RECEIVE_EVENT,
hci.HCI_LE_CONNECTION_RATE_CHANGE_EVENT,
]
)
@@ -889,16 +901,18 @@ class Host(utils.EventEmitter):
if self.local_supported_commands & mask
)
def supports_le_features(self, feature: hci.LeFeatureMask) -> bool:
return (self.local_le_features & feature) == feature
def supports_le_features(self, features: hci.LeFeatureMask) -> bool:
return (self.local_le_features & features) == features
def supports_lmp_features(self, feature: hci.LmpFeatureMask) -> bool:
return self.local_lmp_features & (feature) == feature
def supports_lmp_features(self, features: hci.LmpFeatureMask) -> bool:
return self.local_lmp_features & (features) == features
@property
def supported_le_features(self):
def supported_le_features(self) -> list[hci.LeFeature]:
return [
feature for feature in range(64) if self.local_le_features & (1 << feature)
feature
for feature in hci.LeFeature
if self.local_le_features & (1 << feature)
]
# Packet Sink protocol (packets coming from the controller via HCI)
@@ -1179,7 +1193,7 @@ class Host(utils.EventEmitter):
self, event: hci.HCI_LE_Connection_Update_Complete_Event
):
if (connection := self.connections.get(event.connection_handle)) is None:
logger.warning('!!! CONNECTION PARAMETERS UPDATE COMPLETE: unknown handle')
logger.warning('!!! CONNECTION UPDATE COMPLETE: unknown handle')
return
# Notify the client
@@ -1196,6 +1210,29 @@ class Host(utils.EventEmitter):
'connection_parameters_update_failure', connection.handle, event.status
)
def on_hci_le_connection_rate_change_event(
self, event: hci.HCI_LE_Connection_Rate_Change_Event
):
if (connection := self.connections.get(event.connection_handle)) is None:
logger.warning('!!! CONNECTION RATE CHANGE: unknown handle')
return
# Notify the client
if event.status == hci.HCI_SUCCESS:
self.emit(
'le_connection_rate_change',
connection.handle,
event.connection_interval,
event.subrate_factor,
event.peripheral_latency,
event.continuation_number,
event.supervision_timeout,
)
else:
self.emit(
'le_connection_rate_change_failure', connection.handle, event.status
)
def on_hci_le_phy_update_complete_event(
self, event: hci.HCI_LE_PHY_Update_Complete_Event
):
@@ -1755,12 +1792,13 @@ class Host(utils.EventEmitter):
self.emit(
'le_remote_features_failure', event.connection_handle, event.status
)
else:
self.emit(
'le_remote_features',
event.connection_handle,
int.from_bytes(event.le_features, 'little'),
)
return
self.emit(
'le_remote_features',
event.connection_handle,
hci.LeFeatureMask(int.from_bytes(event.le_features, 'little')),
)
def on_hci_le_cs_read_remote_supported_capabilities_complete_event(
self, event: hci.HCI_LE_CS_Read_Remote_Supported_Capabilities_Complete_Event
@@ -1793,6 +1831,12 @@ class Host(utils.EventEmitter):
self.emit('cs_subevent_result_continue', event)
def on_hci_le_subrate_change_event(self, event: hci.HCI_LE_Subrate_Change_Event):
if event.status != hci.HCI_SUCCESS:
self.emit(
'le_subrate_change_failure', event.connection_handle, event.status
)
return
self.emit(
'le_subrate_change',
event.connection_handle,

201
examples/run_connection_updates.py Normal file
View File

@@ -0,0 +1,201 @@
# Copyright 2026 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
# -----------------------------------------------------------------------------
import asyncio
import sys
from collections.abc import Callable
import bumble.logging
from bumble.core import BaseError
from bumble.device import Connection, Device
from bumble.hci import Address, LeFeatureMask
from bumble.transport import open_transport
# -----------------------------------------------------------------------------
DEFAULT_CENTRAL_ADDRESS = Address("F0:F0:F0:F0:F0:F0")
DEFAULT_PERIPHERAL_ADDRESS = Address("F1:F1:F1:F1:F1:F1")
# -----------------------------------------------------------------------------
async def run_as_central(
device: Device,
scenario: Callable | None,
) -> None:
# Connect to the peripheral
print(f'=== Connecting to {DEFAULT_PERIPHERAL_ADDRESS}...')
connection = await device.connect(DEFAULT_PERIPHERAL_ADDRESS)
print("=== Connected")
if scenario is not None:
await asyncio.sleep(1)
await scenario(connection)
await asyncio.get_running_loop().create_future()
async def run_as_peripheral(device: Device, scenario: Callable | None) -> None:
# Wait for a connection from the central
print(f'=== Advertising as {DEFAULT_PERIPHERAL_ADDRESS}...')
await device.start_advertising(auto_restart=True)
async def on_connection(connection: Connection) -> None:
assert scenario is not None
await asyncio.sleep(1)
await scenario(connection)
if scenario is not None:
device.on(Device.EVENT_CONNECTION, on_connection)
await asyncio.get_running_loop().create_future()
async def change_parameters(
connection: Connection,
parameter_request_procedure_supported: bool,
subrating_supported: bool,
shorter_connection_intervals_supported: bool,
) -> None:
if parameter_request_procedure_supported:
try:
print(">>> update_parameters(7.5, 200, 0, 4000)")
await connection.update_parameters(7.5, 200, 0, 4000)
await asyncio.sleep(3)
except BaseError as error:
print(f"Error: {error}")
if subrating_supported:
try:
print(">>> update_subrate(1, 2, 2, 1, 4000)")
await connection.update_subrate(1, 2, 2, 1, 4000)
await asyncio.sleep(3)
except BaseError as error:
print(f"Error: {error}")
if shorter_connection_intervals_supported:
try:
print(
">>> update_parameters_with_subrate(7.5, 200, 1, 1, 0, 0, 4000, 5, 1000)"
)
await connection.update_parameters_with_subrate(
7.5, 200, 1, 1, 0, 0, 4000, 5, 1000
)
await asyncio.sleep(3)
except BaseError as error:
print(f"Error: {error}")
try:
print(
">>> update_parameters_with_subrate(0.750, 5, 1, 1, 0, 0, 4000, 0.125, 1000)"
)
await connection.update_parameters_with_subrate(
0.750, 5, 1, 1, 0, 0, 4000, 0.125, 1000
)
await asyncio.sleep(3)
except BaseError as error:
print(f"Error: {error}")
print(">>> done")
def on_connection(connection: Connection) -> None:
print(f"+++ Connection established: {connection}")
def on_le_remote_features_change() -> None:
print(f'... LE Remote Features change: {connection.peer_le_features.name}')
connection.on(
connection.EVENT_LE_REMOTE_FEATURES_CHANGE, on_le_remote_features_change
)
def on_connection_parameters_change() -> None:
print(f'... LE Connection Parameters change: {connection.parameters}')
connection.on(
connection.EVENT_CONNECTION_PARAMETERS_UPDATE, on_connection_parameters_change
)
async def main() -> None:
if len(sys.argv) < 3:
print(
'Usage: run_connection_updates.py <transport-spec> '
'central|peripheral initiator|responder'
)
return
print('<<< connecting to HCI...')
async with await open_transport(sys.argv[1]) as hci_transport:
print('<<< connected')
role = sys.argv[2]
direction = sys.argv[3]
device = Device.with_hci(
role,
(
DEFAULT_CENTRAL_ADDRESS
if role == "central"
else DEFAULT_PERIPHERAL_ADDRESS
),
hci_transport.source,
hci_transport.sink,
)
device.le_subrate_enabled = True
device.le_shorter_connection_intervals_enabled = True
await device.power_on()
parameter_request_procedure_supported = device.supports_le_features(
LeFeatureMask.CONNECTION_PARAMETERS_REQUEST_PROCEDURE
)
print(
"Parameters Request Procedure supported: "
f"{parameter_request_procedure_supported}"
)
subrating_supported = device.supports_le_features(
LeFeatureMask.CONNECTION_SUBRATING
)
print(f"Subrating supported: {subrating_supported}")
shorter_connection_intervals_supported = device.supports_le_features(
LeFeatureMask.SHORTER_CONNECTION_INTERVALS
)
print(
"Shorter Connection Intervals supported: "
f"{shorter_connection_intervals_supported}"
)
device.on(Device.EVENT_CONNECTION, on_connection)
async def run(connection: Connection) -> None:
await change_parameters(
connection,
parameter_request_procedure_supported,
subrating_supported,
shorter_connection_intervals_supported,
)
scenario = run if direction == "initiator" else None
if role == "central":
await run_as_central(device, scenario)
else:
await run_as_peripheral(device, scenario)
# -----------------------------------------------------------------------------
bumble.logging.setup_basic_logging('DEBUG')
asyncio.run(main())

4
tests/device_test.py
View File

@@ -619,7 +619,9 @@ async def test_le_request_subrate():
def on_le_subrate_change():
q.put_nowait(lambda: None)
devices.connections[0].on(Connection.EVENT_LE_SUBRATE_CHANGE, on_le_subrate_change)
devices.connections[0].on(
Connection.EVENT_CONNECTION_PARAMETERS_UPDATE, on_le_subrate_change
)
await devices[0].send_command(
hci.HCI_LE_Subrate_Request_Command(