From 7569da37e401edba966beb7594f629986f51a046 Mon Sep 17 00:00:00 2001 From: Josh Wu Date: Wed, 9 Apr 2025 17:59:01 +0800 Subject: [PATCH] Replace legacy transport and role constants --- apps/bench.py | 16 +++--- apps/console.py | 2 +- apps/pair.py | 7 +-- apps/player/player.py | 6 +-- apps/rfcomm_bridge.py | 2 +- apps/speaker/speaker.py | 6 ++- bumble/controller.py | 13 +++-- bumble/device.py | 85 ++++++++++++++++---------------- bumble/hci.py | 4 +- bumble/host.py | 23 +++++---- bumble/link.py | 7 ++- bumble/pandora/host.py | 13 +++-- bumble/pandora/security.py | 21 ++++---- bumble/profiles/csip.py | 4 +- bumble/rfcomm.py | 4 +- bumble/smp.py | 19 +++---- examples/run_a2dp_info.py | 6 ++- examples/run_a2dp_sink.py | 4 +- examples/run_a2dp_source.py | 4 +- examples/run_avrcp.py | 4 +- examples/run_channel_sounding.py | 2 +- examples/run_classic_connect.py | 4 +- examples/run_esco_connection.py | 7 +-- examples/run_hfp_gateway.py | 6 +-- examples/run_hid_device.py | 4 +- examples/run_hid_host.py | 12 +++-- examples/run_rfcomm_client.py | 6 ++- tests/a2dp_test.py | 6 +-- tests/avrcp_test.py | 2 +- tests/device_test.py | 13 +++-- tests/hfp_test.py | 2 +- tests/self_test.py | 6 +-- web/speaker/speaker.py | 6 ++- 33 files changed, 169 insertions(+), 157 deletions(-) diff --git a/apps/bench.py b/apps/bench.py index 50c395f..860f58c 100644 --- a/apps/bench.py +++ b/apps/bench.py @@ -28,8 +28,7 @@ import click from bumble import l2cap from bumble.core import ( - BT_BR_EDR_TRANSPORT, - BT_LE_TRANSPORT, + PhysicalTransport, BT_L2CAP_PROTOCOL_ID, BT_RFCOMM_PROTOCOL_ID, UUID, @@ -42,8 +41,7 @@ from bumble.hci import ( HCI_LE_1M_PHY, HCI_LE_2M_PHY, HCI_LE_CODED_PHY, - HCI_CENTRAL_ROLE, - HCI_PERIPHERAL_ROLE, + Role, HCI_Constant, HCI_Error, HCI_StatusError, @@ -113,7 +111,7 @@ def print_connection_phy(phy): def print_connection(connection): params = [] - if connection.transport == BT_LE_TRANSPORT: + if connection.transport == PhysicalTransport.LE: params.append( 'DL=(' f'TX:{connection.data_length[0]}/{connection.data_length[1]},' @@ -189,7 +187,7 @@ def log_stats(title, stats, precision=2): async def switch_roles(connection, role): - target_role = HCI_CENTRAL_ROLE if role == "central" else HCI_PERIPHERAL_ROLE + target_role = Role.CENTRAL if role == "central" else Role.PERIPHERAL if connection.role != target_role: logging.info(f'{color("### Switching roles to:", "cyan")} {role}') try: @@ -1275,7 +1273,11 @@ class Central(Connection.Listener): self.connection = await self.device.connect( self.peripheral_address, connection_parameters_preferences=self.connection_parameter_preferences, - transport=BT_BR_EDR_TRANSPORT if self.classic else BT_LE_TRANSPORT, + transport=( + PhysicalTransport.BR_EDR + if self.classic + else PhysicalTransport.LE + ), ) except CommandTimeoutError: logging.info(color('!!! Connection timed out', 'red')) diff --git a/apps/console.py b/apps/console.py index 8b896a7..606aadb 100644 --- a/apps/console.py +++ b/apps/console.py @@ -55,7 +55,7 @@ from prompt_toolkit.layout import ( from bumble import __version__ import bumble.core from bumble import colors -from bumble.core import UUID, AdvertisingData, BT_LE_TRANSPORT +from bumble.core import UUID, AdvertisingData, PhysicalTransport from bumble.device import ( ConnectionParametersPreferences, ConnectionPHY, diff --git a/apps/pair.py b/apps/pair.py index daf010a..5ed7679 100644 --- a/apps/pair.py +++ b/apps/pair.py @@ -31,8 +31,7 @@ from bumble.keys import JsonKeyStore from bumble.core import ( AdvertisingData, ProtocolError, - BT_LE_TRANSPORT, - BT_BR_EDR_TRANSPORT, + PhysicalTransport, ) from bumble.gatt import ( GATT_DEVICE_NAME_CHARACTERISTIC, @@ -422,7 +421,9 @@ async def pair( print(color(f'=== Connecting to {address_or_name}...', 'green')) connection = await device.connect( address_or_name, - transport=BT_LE_TRANSPORT if mode == 'le' else BT_BR_EDR_TRANSPORT, + transport=( + PhysicalTransport.LE if mode == 'le' else PhysicalTransport.BR_EDR + ), ) if not request: diff --git a/apps/player/player.py b/apps/player/player.py index 828fa43..448b8cc 100644 --- a/apps/player/player.py +++ b/apps/player/player.py @@ -56,7 +56,7 @@ from bumble.core import ( AdvertisingData, ConnectionError as BumbleConnectionError, DeviceClass, - BT_BR_EDR_TRANSPORT, + PhysicalTransport, ) from bumble.device import Connection, Device, DeviceConfiguration from bumble.hci import Address, HCI_CONNECTION_ALREADY_EXISTS_ERROR, HCI_Constant @@ -286,7 +286,7 @@ class Player: async def connect(self, device: Device, address: str) -> Connection: print(color(f"Connecting to {address}...", "green")) - connection = await device.connect(address, transport=BT_BR_EDR_TRANSPORT) + connection = await device.connect(address, transport=PhysicalTransport.BR_EDR) # Request authentication if self.authenticate: @@ -402,7 +402,7 @@ class Player: async def pair(self, device: Device, address: str) -> None: print(color(f"Connecting to {address}...", "green")) - connection = await device.connect(address, transport=BT_BR_EDR_TRANSPORT) + connection = await device.connect(address, transport=PhysicalTransport.BR_EDR) print(color("Pairing...", "magenta")) await connection.authenticate() diff --git a/apps/rfcomm_bridge.py b/apps/rfcomm_bridge.py index 345e46d..9fb55c9 100644 --- a/apps/rfcomm_bridge.py +++ b/apps/rfcomm_bridge.py @@ -271,7 +271,7 @@ class ClientBridge: print(color(f"@@@ Connecting to Bluetooth {self.address}", "blue")) assert self.device self.connection = await self.device.connect( - self.address, transport=core.BT_BR_EDR_TRANSPORT + self.address, transport=core.PhysicalTransport.BR_EDR ) print(color(f"@@@ Bluetooth connection: {self.connection}", "blue")) self.connection.on("disconnection", self.on_disconnection) diff --git a/apps/speaker/speaker.py b/apps/speaker/speaker.py index 38cd201..92f55cb 100644 --- a/apps/speaker/speaker.py +++ b/apps/speaker/speaker.py @@ -34,7 +34,7 @@ from aiohttp import web import bumble from bumble.colors import color -from bumble.core import BT_BR_EDR_TRANSPORT, CommandTimeoutError +from bumble.core import PhysicalTransport, CommandTimeoutError from bumble.device import Connection, Device, DeviceConfiguration from bumble.hci import HCI_StatusError from bumble.pairing import PairingConfig @@ -568,7 +568,9 @@ class Speaker: async def connect(self, address): # Connect to the source print(f'=== Connecting to {address}...') - connection = await self.device.connect(address, transport=BT_BR_EDR_TRANSPORT) + connection = await self.device.connect( + address, transport=PhysicalTransport.BR_EDR + ) print(f'=== Connected to {connection.peer_address}') # Request authentication diff --git a/bumble/controller.py b/bumble/controller.py index 5fb1ff7..20dfcc5 100644 --- a/bumble/controller.py +++ b/bumble/controller.py @@ -25,8 +25,7 @@ import random import struct from bumble.colors import color from bumble.core import ( - BT_LE_TRANSPORT, - BT_BR_EDR_TRANSPORT, + PhysicalTransport, ) from bumble.hci import ( @@ -392,7 +391,7 @@ class Controller: role=Role.PERIPHERAL, peer_address=peer_address, link=self.link, - transport=BT_LE_TRANSPORT, + transport=PhysicalTransport.LE, link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE, ) self.peripheral_connections[peer_address] = connection @@ -452,7 +451,7 @@ class Controller: role=Role.CENTRAL, peer_address=peer_address, link=self.link, - transport=BT_LE_TRANSPORT, + transport=PhysicalTransport.LE, link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE, ) self.central_connections[peer_address] = connection @@ -530,7 +529,7 @@ class Controller: def on_link_acl_data(self, sender_address, transport, data): # Look for the connection to which this data belongs - if transport == BT_LE_TRANSPORT: + if transport == PhysicalTransport.LE: connection = self.find_le_connection_by_address(sender_address) else: connection = self.find_classic_connection_by_address(sender_address) @@ -695,7 +694,7 @@ class Controller: role=Role.CENTRAL, peer_address=peer_address, link=self.link, - transport=BT_BR_EDR_TRANSPORT, + transport=PhysicalTransport.BR_EDR, link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE, ) self.classic_connections[peer_address] = connection @@ -763,7 +762,7 @@ class Controller: role=Role.CENTRAL, peer_address=peer_address, link=self.link, - transport=BT_BR_EDR_TRANSPORT, + transport=PhysicalTransport.BR_EDR, link_type=link_type, ) self.classic_connections[peer_address] = connection diff --git a/bumble/device.py b/bumble/device.py index 4defac8..08c1263 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -57,8 +57,7 @@ from .gatt import Attribute, Characteristic, Descriptor, Service from .host import DataPacketQueue, Host from .profiles.gap import GenericAccessService from .core import ( - BT_BR_EDR_TRANSPORT, - BT_LE_TRANSPORT, + PhysicalTransport, AdvertisingData, BaseBumbleError, ConnectionParameterUpdateError, @@ -1660,7 +1659,7 @@ class Connection(CompositeEventEmitter): return cls( device, None, - BT_BR_EDR_TRANSPORT, + PhysicalTransport.BR_EDR, device.public_address, None, peer_address, @@ -1675,7 +1674,7 @@ class Connection(CompositeEventEmitter): Finish an incomplete connection upon completion. """ assert self.handle is None - assert self.transport == BT_BR_EDR_TRANSPORT + assert self.transport == PhysicalTransport.BR_EDR self.handle = handle self.parameters = parameters @@ -1830,7 +1829,7 @@ class Connection(CompositeEventEmitter): raise def __str__(self): - if self.transport == BT_LE_TRANSPORT: + if self.transport == PhysicalTransport.LE: return ( f'Connection(transport=LE, handle=0x{self.handle:04X}, ' f'role={self.role_name}, ' @@ -3397,7 +3396,7 @@ class Device(CompositeEventEmitter): async def connect( self, peer_address: Union[hci.Address, str], - transport: core.PhysicalTransport = BT_LE_TRANSPORT, + transport: core.PhysicalTransport = PhysicalTransport.LE, connection_parameters_preferences: Optional[ dict[hci.Phy, ConnectionParametersPreferences] ] = None, @@ -3447,23 +3446,23 @@ class Device(CompositeEventEmitter): ''' # Check parameters - if transport not in (BT_LE_TRANSPORT, BT_BR_EDR_TRANSPORT): + if transport not in (PhysicalTransport.LE, PhysicalTransport.BR_EDR): raise InvalidArgumentError('invalid transport') transport = core.PhysicalTransport(transport) # Adjust the transport automatically if we need to - if transport == BT_LE_TRANSPORT and not self.le_enabled: - transport = BT_BR_EDR_TRANSPORT - elif transport == BT_BR_EDR_TRANSPORT and not self.classic_enabled: - transport = BT_LE_TRANSPORT + if transport == PhysicalTransport.LE and not self.le_enabled: + transport = PhysicalTransport.BR_EDR + elif transport == PhysicalTransport.BR_EDR and not self.classic_enabled: + transport = PhysicalTransport.LE # Check that there isn't already a pending connection - if transport == BT_LE_TRANSPORT and self.is_le_connecting: + if transport == PhysicalTransport.LE and self.is_le_connecting: raise InvalidStateError('connection already pending') if isinstance(peer_address, str): try: - if transport == BT_LE_TRANSPORT and peer_address.endswith('@'): + if transport == PhysicalTransport.LE and peer_address.endswith('@'): peer_address = hci.Address.from_string_for_transport( peer_address[:-1], transport ) @@ -3483,21 +3482,21 @@ class Device(CompositeEventEmitter): else: # All BR/EDR addresses should be public addresses if ( - transport == BT_BR_EDR_TRANSPORT + transport == PhysicalTransport.BR_EDR and peer_address.address_type != hci.Address.PUBLIC_DEVICE_ADDRESS ): raise InvalidArgumentError('BR/EDR addresses must be PUBLIC') assert isinstance(peer_address, hci.Address) - if transport == BT_LE_TRANSPORT and always_resolve: + if transport == PhysicalTransport.LE and always_resolve: logger.debug('resolving address') peer_address = await self.find_peer_by_identity_address( peer_address ) # TODO: timeout def on_connection(connection): - if transport == BT_LE_TRANSPORT or ( + if transport == PhysicalTransport.LE or ( # match BR/EDR connection event against peer address connection.transport == transport and connection.peer_address == peer_address @@ -3505,7 +3504,7 @@ class Device(CompositeEventEmitter): pending_connection.set_result(connection) def on_connection_failure(error): - if transport == BT_LE_TRANSPORT or ( + if transport == PhysicalTransport.LE or ( # match BR/EDR connection failure event against peer address error.transport == transport and error.peer_address == peer_address @@ -3519,7 +3518,7 @@ class Device(CompositeEventEmitter): try: # Tell the controller to connect - if transport == BT_LE_TRANSPORT: + if transport == PhysicalTransport.LE: if connection_parameters_preferences is None: if connection_parameters_preferences is None: connection_parameters_preferences = { @@ -3664,7 +3663,7 @@ class Device(CompositeEventEmitter): raise hci.HCI_StatusError(result) # Wait for the connection process to complete - if transport == BT_LE_TRANSPORT: + if transport == PhysicalTransport.LE: self.le_connecting = True if timeout is None: @@ -3675,7 +3674,7 @@ class Device(CompositeEventEmitter): asyncio.shield(pending_connection), timeout ) except asyncio.TimeoutError: - if transport == BT_LE_TRANSPORT: + if transport == PhysicalTransport.LE: await self.send_command( hci.HCI_LE_Create_Connection_Cancel_Command() ) @@ -3691,7 +3690,7 @@ class Device(CompositeEventEmitter): finally: self.remove_listener('connection', on_connection) self.remove_listener('connection_failure', on_connection_failure) - if transport == BT_LE_TRANSPORT: + if transport == PhysicalTransport.LE: self.le_connecting = False self.connect_own_address_type = None else: @@ -3721,7 +3720,7 @@ class Device(CompositeEventEmitter): # If the address is not parsable, assume it is a name instead logger.debug('looking for peer by name') peer_address = await self.find_peer_by_name( - peer_address, BT_BR_EDR_TRANSPORT + peer_address, PhysicalTransport.BR_EDR ) # TODO: timeout assert isinstance(peer_address, hci.Address) @@ -3771,14 +3770,14 @@ class Device(CompositeEventEmitter): def on_connection(connection): if ( - connection.transport == BT_BR_EDR_TRANSPORT + connection.transport == PhysicalTransport.BR_EDR and connection.peer_address == peer_address ): pending_connection.set_result(connection) def on_connection_failure(error): if ( - error.transport == BT_BR_EDR_TRANSPORT + error.transport == PhysicalTransport.BR_EDR and error.peer_address == peer_address ): pending_connection.set_exception(error) @@ -3848,7 +3847,7 @@ class Device(CompositeEventEmitter): # If the address is not parsable, assume it is a name instead logger.debug('looking for peer by name') peer_address = await self.find_peer_by_name( - peer_address, BT_BR_EDR_TRANSPORT + peer_address, PhysicalTransport.BR_EDR ) # TODO: timeout await self.send_command( @@ -4031,7 +4030,7 @@ class Device(CompositeEventEmitter): check_result=True, ) - async def find_peer_by_name(self, name, transport=BT_LE_TRANSPORT): + async def find_peer_by_name(self, name, transport=PhysicalTransport.LE): """ Scan for a peer with a given name and return its address. """ @@ -4050,7 +4049,7 @@ class Device(CompositeEventEmitter): was_scanning = self.scanning was_discovering = self.discovering try: - if transport == BT_LE_TRANSPORT: + if transport == PhysicalTransport.LE: event_name = 'advertisement' listener = self.on( event_name, @@ -4062,7 +4061,7 @@ class Device(CompositeEventEmitter): if not self.scanning: await self.start_scanning(filter_duplicates=True) - elif transport == BT_BR_EDR_TRANSPORT: + elif transport == PhysicalTransport.BR_EDR: event_name = 'inquiry_result' listener = self.on( event_name, @@ -4081,9 +4080,9 @@ class Device(CompositeEventEmitter): if listener is not None: self.remove_listener(event_name, listener) - if transport == BT_LE_TRANSPORT and not was_scanning: + if transport == PhysicalTransport.LE and not was_scanning: await self.stop_scanning() - elif transport == BT_BR_EDR_TRANSPORT and not was_discovering: + elif transport == PhysicalTransport.BR_EDR and not was_discovering: await self.stop_discovery() async def find_peer_by_identity_address( @@ -4238,7 +4237,7 @@ class Device(CompositeEventEmitter): ) async def encrypt(self, connection, enable=True): - if not enable and connection.transport == BT_LE_TRANSPORT: + if not enable and connection.transport == PhysicalTransport.LE: raise InvalidArgumentError('`enable` parameter is classic only.') # Set up event handlers @@ -4255,7 +4254,7 @@ class Device(CompositeEventEmitter): # Request the encryption try: - if connection.transport == BT_LE_TRANSPORT: + if connection.transport == PhysicalTransport.LE: # Look for a key in the key store if self.keystore is None: raise InvalidOperationError('no key store') @@ -4276,7 +4275,7 @@ class Device(CompositeEventEmitter): else: raise InvalidOperationError('no LTK found for peer') - if connection.role != hci.HCI_CENTRAL_ROLE: + if connection.role != hci.Role.CENTRAL: raise InvalidStateError('only centrals can start encryption') result = await self.send_command( @@ -4932,7 +4931,7 @@ class Device(CompositeEventEmitter): self.abort_on('flush', self.update_keys(str(bd_addr), pairing_keys)) if connection := self.find_connection_by_bd_addr( - bd_addr, transport=BT_BR_EDR_TRANSPORT + bd_addr, transport=PhysicalTransport.BR_EDR ): connection.link_key_type = key_type @@ -5232,7 +5231,7 @@ class Device(CompositeEventEmitter): 'new connection reuses the same handle as a previous connection' ) - if transport == BT_BR_EDR_TRANSPORT: + if transport == PhysicalTransport.BR_EDR: # Create a new connection connection = self.pending_connections.pop(peer_address) connection.complete(connection_handle, connection_parameters) @@ -5255,7 +5254,7 @@ class Device(CompositeEventEmitter): self_address = None own_address_type: Optional[hci.OwnAddressType] = None - if role == hci.HCI_CENTRAL_ROLE: + if role == hci.Role.CENTRAL: own_address_type = self.connect_own_address_type assert own_address_type is not None else: @@ -5302,7 +5301,7 @@ class Device(CompositeEventEmitter): ) self.connections[connection_handle] = connection - if role == hci.HCI_PERIPHERAL_ROLE and self.legacy_advertiser: + if role == hci.Role.PERIPHERAL and self.legacy_advertiser: if self.legacy_advertiser.auto_restart: advertiser = self.legacy_advertiser connection.once( @@ -5312,12 +5311,12 @@ class Device(CompositeEventEmitter): else: self.legacy_advertiser = None - if role == hci.HCI_CENTRAL_ROLE or not self.supports_le_extended_advertising: + if role == hci.Role.CENTRAL or not self.supports_le_extended_advertising: # We can emit now, we have all the info we need self.emit('connection', connection) return - if role == hci.HCI_PERIPHERAL_ROLE and self.supports_le_extended_advertising: + if role == hci.Role.PERIPHERAL and self.supports_le_extended_advertising: if advertising_set := self.connecting_extended_advertising_sets.pop( connection_handle, None ): @@ -5334,7 +5333,7 @@ class Device(CompositeEventEmitter): # For directed advertising, this means a timeout if ( - transport == BT_LE_TRANSPORT + transport == PhysicalTransport.LE and self.legacy_advertiser and self.legacy_advertiser.advertising_type.is_directed ): @@ -5361,7 +5360,7 @@ class Device(CompositeEventEmitter): hci.HCI_Connection_Complete_Event.ESCO_LINK_TYPE, ): if connection := self.find_connection_by_bd_addr( - bd_addr, transport=BT_BR_EDR_TRANSPORT + bd_addr, transport=PhysicalTransport.BR_EDR ): self.emit('sco_request', connection, link_type) else: @@ -5812,14 +5811,14 @@ class Device(CompositeEventEmitter): connection.encryption = encryption if ( not connection.authenticated - and connection.transport == BT_BR_EDR_TRANSPORT + and connection.transport == PhysicalTransport.BR_EDR and encryption == hci.HCI_Encryption_Change_Event.AES_CCM ): connection.authenticated = True connection.sc = True if ( not connection.authenticated - and connection.transport == BT_LE_TRANSPORT + and connection.transport == PhysicalTransport.LE and encryption == hci.HCI_Encryption_Change_Event.E0_OR_AES_CCM ): connection.authenticated = True diff --git a/bumble/hci.py b/bumble/hci.py index 5f8c63a..175bd4e 100644 --- a/bumble/hci.py +++ b/bumble/hci.py @@ -29,7 +29,7 @@ from typing_extensions import Self from bumble import crypto from bumble.colors import color from bumble.core import ( - BT_BR_EDR_TRANSPORT, + PhysicalTransport, AdvertisingData, DeviceClass, InvalidArgumentError, @@ -1976,7 +1976,7 @@ class Address: def from_string_for_transport( cls: type[Self], string: str, transport: PhysicalTransport ) -> Self: - if transport == BT_BR_EDR_TRANSPORT: + if transport == PhysicalTransport.BR_EDR: address_type = Address.PUBLIC_DEVICE_ADDRESS else: address_type = Address.RANDOM_DEVICE_ADDRESS diff --git a/bumble/host.py b/bumble/host.py index f86c5b2..4cb00a5 100644 --- a/bumble/host.py +++ b/bumble/host.py @@ -42,8 +42,7 @@ from bumble.snoop import Snooper from bumble import drivers from bumble import hci from bumble.core import ( - BT_BR_EDR_TRANSPORT, - BT_LE_TRANSPORT, + PhysicalTransport, PhysicalTransport, ConnectionPHY, ConnectionParameters, @@ -200,7 +199,7 @@ class Connection: self.transport = transport acl_packet_queue: Optional[DataPacketQueue] = ( host.le_acl_packet_queue - if transport == BT_LE_TRANSPORT + if transport == PhysicalTransport.LE else host.acl_packet_queue ) assert acl_packet_queue @@ -967,7 +966,7 @@ class Host(AbortableEventEmitter): self, event.connection_handle, event.peer_address, - BT_LE_TRANSPORT, + PhysicalTransport.LE, ) self.connections[event.connection_handle] = connection @@ -980,7 +979,7 @@ class Host(AbortableEventEmitter): self.emit( 'connection', event.connection_handle, - BT_LE_TRANSPORT, + PhysicalTransport.LE, event.peer_address, getattr(event, 'local_resolvable_private_address', None), getattr(event, 'peer_resolvable_private_address', None), @@ -992,7 +991,10 @@ class Host(AbortableEventEmitter): # Notify the listeners self.emit( - 'connection_failure', BT_LE_TRANSPORT, event.peer_address, event.status + 'connection_failure', + PhysicalTransport.LE, + event.peer_address, + event.status, ) def on_hci_le_enhanced_connection_complete_event(self, event): @@ -1017,7 +1019,7 @@ class Host(AbortableEventEmitter): self, event.connection_handle, event.bd_addr, - BT_BR_EDR_TRANSPORT, + PhysicalTransport.BR_EDR, ) self.connections[event.connection_handle] = connection @@ -1025,7 +1027,7 @@ class Host(AbortableEventEmitter): self.emit( 'connection', event.connection_handle, - BT_BR_EDR_TRANSPORT, + PhysicalTransport.BR_EDR, event.bd_addr, None, None, @@ -1037,7 +1039,10 @@ class Host(AbortableEventEmitter): # Notify the client self.emit( - 'connection_failure', BT_BR_EDR_TRANSPORT, event.bd_addr, event.status + 'connection_failure', + PhysicalTransport.BR_EDR, + event.bd_addr, + event.status, ) def on_hci_disconnection_complete_event(self, event): diff --git a/bumble/link.py b/bumble/link.py index a24247a..971b6fb 100644 --- a/bumble/link.py +++ b/bumble/link.py @@ -20,8 +20,7 @@ import asyncio from functools import partial from bumble.core import ( - BT_BR_EDR_TRANSPORT, - BT_LE_TRANSPORT, + PhysicalTransport, InvalidStateError, ) from bumble.colors import color @@ -116,10 +115,10 @@ class LocalLink: def send_acl_data(self, sender_controller, destination_address, transport, data): # Send the data to the first controller with a matching address - if transport == BT_LE_TRANSPORT: + if transport == PhysicalTransport.LE: destination_controller = self.find_controller(destination_address) source_address = sender_controller.random_address - elif transport == BT_BR_EDR_TRANSPORT: + elif transport == PhysicalTransport.BR_EDR: destination_controller = self.find_classic_controller(destination_address) source_address = sender_controller.public_address else: diff --git a/bumble/pandora/host.py b/bumble/pandora/host.py index 9c01952..2f7fe52 100644 --- a/bumble/pandora/host.py +++ b/bumble/pandora/host.py @@ -23,8 +23,7 @@ import struct from . import utils from .config import Config from bumble.core import ( - BT_BR_EDR_TRANSPORT, - BT_LE_TRANSPORT, + PhysicalTransport, UUID, AdvertisingData, Appearance, @@ -185,7 +184,7 @@ class HostService(HostServicer): try: connection = await self.device.connect( - address, transport=BT_BR_EDR_TRANSPORT + address, transport=PhysicalTransport.BR_EDR ) except ConnectionError as e: if e.error_code == HCI_PAGE_TIMEOUT_ERROR: @@ -218,7 +217,7 @@ class HostService(HostServicer): self.log.debug(f"WaitConnection from {address}...") connection = self.device.find_connection_by_bd_addr( - address, transport=BT_BR_EDR_TRANSPORT + address, transport=PhysicalTransport.BR_EDR ) if connection and id(connection) in self.waited_connections: # this connection was already returned: wait for a new one. @@ -250,7 +249,7 @@ class HostService(HostServicer): try: connection = await self.device.connect( address, - transport=BT_LE_TRANSPORT, + transport=PhysicalTransport.LE, own_address_type=OwnAddressType(request.own_address_type), ) except ConnectionError as e: @@ -378,7 +377,7 @@ class HostService(HostServicer): def on_connection(connection: bumble.device.Connection) -> None: if ( - connection.transport == BT_LE_TRANSPORT + connection.transport == PhysicalTransport.LE and connection.role == Role.PERIPHERAL ): connections.put_nowait(connection) @@ -496,7 +495,7 @@ class HostService(HostServicer): def on_connection(connection: bumble.device.Connection) -> None: if ( - connection.transport == BT_LE_TRANSPORT + connection.transport == PhysicalTransport.LE and connection.role == Role.PERIPHERAL ): connections.put_nowait(connection) diff --git a/bumble/pandora/security.py b/bumble/pandora/security.py index 4e29539..d60e95b 100644 --- a/bumble/pandora/security.py +++ b/bumble/pandora/security.py @@ -22,8 +22,7 @@ from . import utils from .config import Config from bumble import hci from bumble.core import ( - BT_BR_EDR_TRANSPORT, - BT_LE_TRANSPORT, + PhysicalTransport, ProtocolError, ) from bumble.device import Connection as BumbleConnection, Device @@ -94,7 +93,7 @@ class PairingDelegate(BasePairingDelegate): else: # In BR/EDR, connection may not be complete, # use address instead - assert self.connection.transport == BT_BR_EDR_TRANSPORT + assert self.connection.transport == PhysicalTransport.BR_EDR ev.address = bytes(reversed(bytes(self.connection.peer_address))) return ev @@ -173,7 +172,7 @@ class PairingDelegate(BasePairingDelegate): async def display_number(self, number: int, digits: int = 6) -> None: if ( - self.connection.transport == BT_BR_EDR_TRANSPORT + self.connection.transport == PhysicalTransport.BR_EDR and self.io_capability == BasePairingDelegate.DISPLAY_OUTPUT_ONLY ): return @@ -286,7 +285,7 @@ class SecurityService(SecurityServicer): oneof = request.WhichOneof('level') level = getattr(request, oneof) - assert {BT_BR_EDR_TRANSPORT: 'classic', BT_LE_TRANSPORT: 'le'}[ + assert {PhysicalTransport.BR_EDR: 'classic', PhysicalTransport.LE: 'le'}[ connection.transport ] == oneof @@ -316,7 +315,7 @@ class SecurityService(SecurityServicer): security_result.set_result('connection_died') if ( - connection.transport == BT_LE_TRANSPORT + connection.transport == PhysicalTransport.LE and connection.role == Role.PERIPHERAL ): connection.request_pairing() @@ -378,7 +377,7 @@ class SecurityService(SecurityServicer): assert request.level level = request.level - assert {BT_BR_EDR_TRANSPORT: 'classic', BT_LE_TRANSPORT: 'le'}[ + assert {PhysicalTransport.BR_EDR: 'classic', PhysicalTransport.LE: 'le'}[ connection.transport ] == request.level_variant() @@ -426,7 +425,7 @@ class SecurityService(SecurityServicer): self.log.debug('Wait for security: done') wait_for_security.set_result('success') elif ( - connection.transport == BT_BR_EDR_TRANSPORT + connection.transport == PhysicalTransport.BR_EDR and self.need_authentication(connection, level) ): nonlocal authenticate_task @@ -504,12 +503,12 @@ class SecurityService(SecurityServicer): return BR_LEVEL_REACHED[level](connection) def need_pairing(self, connection: BumbleConnection, level: int) -> bool: - if connection.transport == BT_LE_TRANSPORT: + if connection.transport == PhysicalTransport.LE: return level >= LE_LEVEL3 and not connection.authenticated return False def need_authentication(self, connection: BumbleConnection, level: int) -> bool: - if connection.transport == BT_LE_TRANSPORT: + if connection.transport == PhysicalTransport.LE: return False if level == LEVEL2 and connection.encryption != 0: return not connection.authenticated @@ -517,7 +516,7 @@ class SecurityService(SecurityServicer): def need_encryption(self, connection: BumbleConnection, level: int) -> bool: # TODO(abel): need to support MITM - if connection.transport == BT_LE_TRANSPORT: + if connection.transport == PhysicalTransport.LE: return level == LE_LEVEL2 and not connection.encryption return level >= LEVEL2 and not connection.encryption diff --git a/bumble/profiles/csip.py b/bumble/profiles/csip.py index 9ba3baf..4a7e941 100644 --- a/bumble/profiles/csip.py +++ b/bumble/profiles/csip.py @@ -170,7 +170,7 @@ class CoordinatedSetIdentificationService(gatt.TemplateService): else: assert connection - if connection.transport == core.BT_LE_TRANSPORT: + if connection.transport == core.PhysicalTransport.LE: key = await connection.device.get_long_term_key( connection_handle=connection.handle, rand=b'', ediv=0 ) @@ -242,7 +242,7 @@ class CoordinatedSetIdentificationProxy(gatt_client.ProfileServiceProxy): else: connection = self.service_proxy.client.connection device = connection.device - if connection.transport == core.BT_LE_TRANSPORT: + if connection.transport == core.PhysicalTransport.LE: key = await device.get_long_term_key( connection_handle=connection.handle, rand=b'', ediv=0 ) diff --git a/bumble/rfcomm.py b/bumble/rfcomm.py index 2de7374..64af375 100644 --- a/bumble/rfcomm.py +++ b/bumble/rfcomm.py @@ -34,7 +34,7 @@ from .colors import color from .core import ( UUID, BT_RFCOMM_PROTOCOL_ID, - BT_BR_EDR_TRANSPORT, + PhysicalTransport, BT_L2CAP_PROTOCOL_ID, InvalidArgumentError, InvalidStateError, @@ -845,7 +845,7 @@ class Multiplexer(EventEmitter): self.open_result.set_exception( core.ConnectionError( core.ConnectionError.CONNECTION_REFUSED, - BT_BR_EDR_TRANSPORT, + PhysicalTransport.BR_EDR, self.l2cap_channel.connection.peer_address, 'rfcomm', ) diff --git a/bumble/smp.py b/bumble/smp.py index 2c66393..2966aa1 100644 --- a/bumble/smp.py +++ b/bumble/smp.py @@ -52,8 +52,7 @@ from .hci import ( key_with_value, ) from .core import ( - BT_BR_EDR_TRANSPORT, - BT_LE_TRANSPORT, + PhysicalTransport, AdvertisingData, InvalidArgumentError, ProtocolError, @@ -857,7 +856,7 @@ class Session: initiator_io_capability: int, responder_io_capability: int, ) -> None: - if self.connection.transport == BT_BR_EDR_TRANSPORT: + if self.connection.transport == PhysicalTransport.BR_EDR: self.pairing_method = PairingMethod.CTKD_OVER_CLASSIC return if (not self.mitm) and (auth_req & SMP_MITM_AUTHREQ == 0): @@ -1170,7 +1169,7 @@ class Session: if self.is_initiator: # CTKD: Derive LTK from LinkKey if ( - self.connection.transport == BT_BR_EDR_TRANSPORT + self.connection.transport == PhysicalTransport.BR_EDR and self.initiator_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG ): self.ctkd_task = self.connection.abort_on( @@ -1209,7 +1208,7 @@ class Session: else: # CTKD: Derive LTK from LinkKey if ( - self.connection.transport == BT_BR_EDR_TRANSPORT + self.connection.transport == PhysicalTransport.BR_EDR and self.responder_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG ): self.ctkd_task = self.connection.abort_on( @@ -1248,7 +1247,7 @@ class Session: def compute_peer_expected_distributions(self, key_distribution_flags: int) -> None: # Set our expectations for what to wait for in the key distribution phase self.peer_expected_distributions = [] - if not self.sc and self.connection.transport == BT_LE_TRANSPORT: + if not self.sc and self.connection.transport == PhysicalTransport.LE: if key_distribution_flags & SMP_ENC_KEY_DISTRIBUTION_FLAG != 0: self.peer_expected_distributions.append( SMP_Encryption_Information_Command @@ -1365,7 +1364,7 @@ class Session: keys = PairingKeys() keys.address_type = peer_address.address_type authenticated = self.pairing_method != PairingMethod.JUST_WORKS - if self.sc or self.connection.transport == BT_BR_EDR_TRANSPORT: + if self.sc or self.connection.transport == PhysicalTransport.BR_EDR: keys.ltk = PairingKeys.Key(value=self.ltk, authenticated=authenticated) else: our_ltk_key = PairingKeys.Key( @@ -1506,7 +1505,7 @@ class Session: # CTKD over BR/EDR should happen after the connection has been encrypted, # so when receiving pairing requests, responder should start distributing keys if ( - self.connection.transport == BT_BR_EDR_TRANSPORT + self.connection.transport == PhysicalTransport.BR_EDR and self.connection.is_encrypted and self.is_responder and accepted @@ -1950,7 +1949,9 @@ class Manager(EventEmitter): f'>>> Sending SMP Command on connection [0x{connection.handle:04X}] ' f'{connection.peer_address}: {command}' ) - cid = SMP_BR_CID if connection.transport == BT_BR_EDR_TRANSPORT else SMP_CID + cid = ( + SMP_BR_CID if connection.transport == PhysicalTransport.BR_EDR else SMP_CID + ) connection.send_l2cap_pdu(cid, bytes(command)) def on_smp_security_request_command( diff --git a/examples/run_a2dp_info.py b/examples/run_a2dp_info.py index e05c87e..ff8ded8 100644 --- a/examples/run_a2dp_info.py +++ b/examples/run_a2dp_info.py @@ -24,7 +24,7 @@ from bumble.colors import color from bumble.device import Device from bumble.transport import open_transport_or_link from bumble.core import ( - BT_BR_EDR_TRANSPORT, + PhysicalTransport, BT_AVDTP_PROTOCOL_ID, BT_AUDIO_SINK_SERVICE, BT_L2CAP_PROTOCOL_ID, @@ -165,7 +165,9 @@ async def main() -> None: # Connect to a peer target_address = sys.argv[3] print(f'=== Connecting to {target_address}...') - connection = await device.connect(target_address, transport=BT_BR_EDR_TRANSPORT) + connection = await device.connect( + target_address, transport=PhysicalTransport.BR_EDR + ) print(f'=== Connected to {connection.peer_address}!') # Request authentication diff --git a/examples/run_a2dp_sink.py b/examples/run_a2dp_sink.py index aa0a152..a3bcb29 100644 --- a/examples/run_a2dp_sink.py +++ b/examples/run_a2dp_sink.py @@ -23,7 +23,7 @@ from typing import Any, Dict from bumble.device import Device from bumble.transport import open_transport_or_link -from bumble.core import BT_BR_EDR_TRANSPORT +from bumble.core import PhysicalTransport from bumble.avdtp import ( AVDTP_AUDIO_MEDIA_TYPE, Protocol, @@ -145,7 +145,7 @@ async def main() -> None: target_address = sys.argv[4] print(f'=== Connecting to {target_address}...') connection = await device.connect( - target_address, transport=BT_BR_EDR_TRANSPORT + target_address, transport=PhysicalTransport.BR_EDR ) print(f'=== Connected to {connection.peer_address}!') diff --git a/examples/run_a2dp_source.py b/examples/run_a2dp_source.py index 20182e1..ba4e24e 100644 --- a/examples/run_a2dp_source.py +++ b/examples/run_a2dp_source.py @@ -23,7 +23,7 @@ import logging from bumble.colors import color from bumble.device import Device from bumble.transport import open_transport_or_link -from bumble.core import BT_BR_EDR_TRANSPORT +from bumble.core import PhysicalTransport from bumble.avdtp import ( find_avdtp_service_with_connection, AVDTP_AUDIO_MEDIA_TYPE, @@ -146,7 +146,7 @@ async def main() -> None: target_address = sys.argv[4] print(f'=== Connecting to {target_address}...') connection = await device.connect( - target_address, transport=BT_BR_EDR_TRANSPORT + target_address, transport=PhysicalTransport.BR_EDR ) print(f'=== Connected to {connection.peer_address}!') diff --git a/examples/run_avrcp.py b/examples/run_avrcp.py index 8636103..eb31b0f 100644 --- a/examples/run_avrcp.py +++ b/examples/run_avrcp.py @@ -25,7 +25,7 @@ import websockets from bumble.device import Device from bumble.transport import open_transport_or_link -from bumble.core import BT_BR_EDR_TRANSPORT +from bumble.core import PhysicalTransport from bumble import avc from bumble import avrcp from bumble import avdtp @@ -379,7 +379,7 @@ async def main() -> None: target_address = sys.argv[4] print(f'=== Connecting to {target_address}...') connection = await device.connect( - target_address, transport=BT_BR_EDR_TRANSPORT + target_address, transport=PhysicalTransport.BR_EDR ) print(f'=== Connected to {connection.peer_address}!') diff --git a/examples/run_channel_sounding.py b/examples/run_channel_sounding.py index 02eadfd..346b775 100644 --- a/examples/run_channel_sounding.py +++ b/examples/run_channel_sounding.py @@ -112,7 +112,7 @@ async def main() -> None: print(f'<<< Connecting to {target_address}') connection = await device.connect( - target_address, transport=core.BT_LE_TRANSPORT + target_address, transport=core.PhysicalTransport.LE ) print('<<< ACL Connected') if not (await device.get_long_term_key(connection.handle, b'', 0)): diff --git a/examples/run_classic_connect.py b/examples/run_classic_connect.py index 362e6b8..652858b 100644 --- a/examples/run_classic_connect.py +++ b/examples/run_classic_connect.py @@ -23,7 +23,7 @@ from bumble.colors import color from bumble.device import Device from bumble.transport import open_transport_or_link -from bumble.core import BT_BR_EDR_TRANSPORT, BT_L2CAP_PROTOCOL_ID, CommandTimeoutError +from bumble.core import PhysicalTransport, BT_L2CAP_PROTOCOL_ID, CommandTimeoutError from bumble.sdp import ( Client as SDP_Client, SDP_PUBLIC_BROWSE_ROOT, @@ -57,7 +57,7 @@ async def main() -> None: print(f'=== Connecting to {target_address}...') try: connection = await device.connect( - target_address, transport=BT_BR_EDR_TRANSPORT + target_address, transport=PhysicalTransport.BR_EDR ) except CommandTimeoutError: print('!!! Connection timed out') diff --git a/examples/run_esco_connection.py b/examples/run_esco_connection.py index 6f3e800..3681df7 100644 --- a/examples/run_esco_connection.py +++ b/examples/run_esco_connection.py @@ -16,11 +16,10 @@ # Imports # ----------------------------------------------------------------------------- import asyncio -import dataclasses import logging import sys import os -from bumble.core import BT_BR_EDR_TRANSPORT +from bumble.core import PhysicalTransport from bumble.device import Device, ScoLink from bumble.hci import HCI_Enhanced_Setup_Synchronous_Connection_Command from bumble.hfp import DefaultCodecParameters, ESCO_PARAMETERS @@ -61,7 +60,9 @@ async def main() -> None: connections = await asyncio.gather( devices[0].accept(devices[1].public_address), - devices[1].connect(devices[0].public_address, transport=BT_BR_EDR_TRANSPORT), + devices[1].connect( + devices[0].public_address, transport=PhysicalTransport.BR_EDR + ), ) def on_sco(sco_link: ScoLink): diff --git a/examples/run_hfp_gateway.py b/examples/run_hfp_gateway.py index cf61190..7b2bb80 100644 --- a/examples/run_hfp_gateway.py +++ b/examples/run_hfp_gateway.py @@ -28,9 +28,7 @@ import websockets import bumble.core from bumble.device import Device, ScoLink from bumble.transport import open_transport_or_link -from bumble.core import ( - BT_BR_EDR_TRANSPORT, -) +from bumble.core import PhysicalTransport from bumble import hci, rfcomm, hfp @@ -234,7 +232,7 @@ async def main() -> None: target_address = sys.argv[3] print(f'=== Connecting to {target_address}...') connection = await device.connect( - target_address, transport=BT_BR_EDR_TRANSPORT + target_address, transport=PhysicalTransport.BR_EDR ) print(f'=== Connected to {connection.peer_address}!') diff --git a/examples/run_hid_device.py b/examples/run_hid_device.py index 160e395..d9e5b73 100644 --- a/examples/run_hid_device.py +++ b/examples/run_hid_device.py @@ -26,7 +26,7 @@ import struct from bumble.device import Device from bumble.transport import open_transport_or_link from bumble.core import ( - BT_BR_EDR_TRANSPORT, + PhysicalTransport, BT_L2CAP_PROTOCOL_ID, BT_HUMAN_INTERFACE_DEVICE_SERVICE, BT_HIDP_PROTOCOL_ID, @@ -721,7 +721,7 @@ async def main() -> None: elif choice == '9': hid_host_bd_addr = str(hid_device.remote_device_bd_address) connection = await device.connect( - hid_host_bd_addr, transport=BT_BR_EDR_TRANSPORT + hid_host_bd_addr, transport=PhysicalTransport.BR_EDR ) await connection.authenticate() await connection.encrypt() diff --git a/examples/run_hid_host.py b/examples/run_hid_host.py index cc17cc1..12d0108 100644 --- a/examples/run_hid_host.py +++ b/examples/run_hid_host.py @@ -26,7 +26,7 @@ from bumble.device import Device from bumble.transport import open_transport_or_link from bumble.core import ( BT_HUMAN_INTERFACE_DEVICE_SERVICE, - BT_BR_EDR_TRANSPORT, + PhysicalTransport, ) from bumble.hci import Address from bumble.hid import Host, Message @@ -349,7 +349,9 @@ async def main() -> None: # Connect to a peer target_address = sys.argv[3] print(f'=== Connecting to {target_address}...') - connection = await device.connect(target_address, transport=BT_BR_EDR_TRANSPORT) + connection = await device.connect( + target_address, transport=PhysicalTransport.BR_EDR + ) print(f'=== Connected to {connection.peer_address}!') # Request authentication @@ -519,10 +521,10 @@ async def main() -> None: elif choice == '13': peer_address = Address.from_string_for_transport( - target_address, transport=BT_BR_EDR_TRANSPORT + target_address, transport=PhysicalTransport.BR_EDR ) connection = device.find_connection_by_bd_addr( - peer_address, transport=BT_BR_EDR_TRANSPORT + peer_address, transport=PhysicalTransport.BR_EDR ) if connection is not None: await connection.disconnect() @@ -538,7 +540,7 @@ async def main() -> None: elif choice == '15': connection = await device.connect( - target_address, transport=BT_BR_EDR_TRANSPORT + target_address, transport=PhysicalTransport.BR_EDR ) await connection.authenticate() await connection.encrypt() diff --git a/examples/run_rfcomm_client.py b/examples/run_rfcomm_client.py index 9232dc9..7b97fc8 100644 --- a/examples/run_rfcomm_client.py +++ b/examples/run_rfcomm_client.py @@ -28,7 +28,7 @@ from bumble.transport import open_transport_or_link from bumble.core import ( BT_L2CAP_PROTOCOL_ID, BT_RFCOMM_PROTOCOL_ID, - BT_BR_EDR_TRANSPORT, + PhysicalTransport, ) from bumble.rfcomm import Client from bumble.sdp import ( @@ -191,7 +191,9 @@ async def main() -> None: # Connect to a peer target_address = sys.argv[3] print(f'=== Connecting to {target_address}...') - connection = await device.connect(target_address, transport=BT_BR_EDR_TRANSPORT) + connection = await device.connect( + target_address, transport=PhysicalTransport.BR_EDR + ) print(f'=== Connected to {connection.peer_address}!') channel_str = sys.argv[4] diff --git a/tests/a2dp_test.py b/tests/a2dp_test.py index d28b4a4..2a359b5 100644 --- a/tests/a2dp_test.py +++ b/tests/a2dp_test.py @@ -21,7 +21,7 @@ import os import pytest from bumble.controller import Controller -from bumble.core import BT_BR_EDR_TRANSPORT +from bumble.core import PhysicalTransport from bumble.link import LocalLink from bumble.device import Device from bumble.host import Host @@ -106,7 +106,7 @@ async def test_self_connection(): # Connect the two devices await asyncio.gather( two_devices.devices[0].connect( - two_devices.devices[1].public_address, transport=BT_BR_EDR_TRANSPORT + two_devices.devices[1].public_address, transport=PhysicalTransport.BR_EDR ), two_devices.devices[1].accept(two_devices.devices[0].public_address), ) @@ -190,7 +190,7 @@ async def test_source_sink_1(): async def make_connection(): connections = await asyncio.gather( two_devices.devices[0].connect( - two_devices.devices[1].public_address, BT_BR_EDR_TRANSPORT + two_devices.devices[1].public_address, PhysicalTransport.BR_EDR ), two_devices.devices[1].accept(two_devices.devices[0].public_address), ) diff --git a/tests/avrcp_test.py b/tests/avrcp_test.py index 103f360..71955ea 100644 --- a/tests/avrcp_test.py +++ b/tests/avrcp_test.py @@ -70,7 +70,7 @@ class TwoDevices: self.connections = await asyncio.gather( self.devices[0].connect( - self.devices[1].public_address, core.BT_BR_EDR_TRANSPORT + self.devices[1].public_address, core.PhysicalTransport.BR_EDR ), self.devices[1].accept(self.devices[0].public_address), ) diff --git a/tests/device_test.py b/tests/device_test.py index ffb407c..8cd78aa 100644 --- a/tests/device_test.py +++ b/tests/device_test.py @@ -22,8 +22,7 @@ import os import pytest from bumble.core import ( - BT_BR_EDR_TRANSPORT, - BT_LE_TRANSPORT, + PhysicalTransport, ConnectionParameters, ) from bumble.device import ( @@ -229,10 +228,10 @@ async def test_device_connect_parallel(): [c01, c02, a10, a20] = await asyncio.gather( *[ asyncio.create_task( - d0.connect(d1.public_address, transport=BT_BR_EDR_TRANSPORT) + d0.connect(d1.public_address, transport=PhysicalTransport.BR_EDR) ), asyncio.create_task( - d0.connect(d2.public_address, transport=BT_BR_EDR_TRANSPORT) + d0.connect(d2.public_address, transport=PhysicalTransport.BR_EDR) ), d1_accept_task, d2_accept_task, @@ -291,7 +290,7 @@ async def test_legacy_advertising_disconnection(auto_restart): await device.start_advertising(auto_restart=auto_restart) device.on_connection( 0x0001, - BT_LE_TRANSPORT, + PhysicalTransport.LE, peer_address, None, None, @@ -349,7 +348,7 @@ async def test_extended_advertising_connection(own_address_type): ) device.on_connection( 0x0001, - BT_LE_TRANSPORT, + PhysicalTransport.LE, peer_address, None, None, @@ -393,7 +392,7 @@ async def test_extended_advertising_connection_out_of_order(own_address_type): ) device.on_connection( 0x0001, - BT_LE_TRANSPORT, + PhysicalTransport.LE, Address('F0:F1:F2:F3:F4:F5'), None, None, diff --git a/tests/hfp_test.py b/tests/hfp_test.py index 47ffe6b..1926342 100644 --- a/tests/hfp_test.py +++ b/tests/hfp_test.py @@ -522,7 +522,7 @@ async def test_sco_setup(): connections = await asyncio.gather( devices[0].connect( - devices[1].public_address, transport=core.BT_BR_EDR_TRANSPORT + devices[1].public_address, transport=core.PhysicalTransport.BR_EDR ), devices[1].accept(devices[0].public_address), ) diff --git a/tests/self_test.py b/tests/self_test.py index 1853433..2456742 100644 --- a/tests/self_test.py +++ b/tests/self_test.py @@ -24,7 +24,7 @@ import pytest from unittest.mock import AsyncMock, MagicMock, patch from bumble.controller import Controller -from bumble.core import BT_BR_EDR_TRANSPORT, BT_LE_TRANSPORT +from bumble.core import PhysicalTransport from bumble.link import LocalLink from bumble.device import Device, Peer from bumble.host import Host @@ -137,7 +137,7 @@ async def test_self_classic_connection(responder_role): # Connect the two devices await asyncio.gather( two_devices.devices[0].connect( - two_devices.devices[1].public_address, transport=BT_BR_EDR_TRANSPORT + two_devices.devices[1].public_address, transport=PhysicalTransport.BR_EDR ), two_devices.devices[1].accept( two_devices.devices[0].public_address, responder_role @@ -507,7 +507,7 @@ async def test_self_smp_over_classic(): # Connect the two devices await asyncio.gather( two_devices.devices[0].connect( - two_devices.devices[1].public_address, transport=BT_BR_EDR_TRANSPORT + two_devices.devices[1].public_address, transport=PhysicalTransport.BR_EDR ), two_devices.devices[1].accept(two_devices.devices[0].public_address), ) diff --git a/web/speaker/speaker.py b/web/speaker/speaker.py index 5187581..e70d0a2 100644 --- a/web/speaker/speaker.py +++ b/web/speaker/speaker.py @@ -20,7 +20,7 @@ import enum import logging from typing import Dict, List -from bumble.core import BT_BR_EDR_TRANSPORT, CommandTimeoutError +from bumble.core import PhysicalTransport, CommandTimeoutError from bumble.device import Device, DeviceConfiguration from bumble.pairing import PairingConfig from bumble.sdp import ServiceAttribute @@ -229,7 +229,9 @@ class Speaker: async def connect(self, address): # Connect to the source print(f'=== Connecting to {address}...') - connection = await self.device.connect(address, transport=BT_BR_EDR_TRANSPORT) + connection = await self.device.connect( + address, transport=PhysicalTransport.BR_EDR + ) print(f'=== Connected to {connection.peer_address}') # Request authentication