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

Controller: SCO implementation

Browse Source
This commit is contained in:
Josh Wu
2023-12-28 14:18:56 +08:00
parent 8ccfc90fe6
commit 8d46bc04d2
8 changed files with 274 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.vscode/settings.json vendored
View File

@@ -32,6 +32,7 @@
"dhkey",
"diversifier",
"endianness",
"ESCO",
"Fitbit",
"GATTLINK",
"HANDSFREE",

154
bumble/controller.py
View File

@@ -19,6 +19,7 @@ from __future__ import annotations
import logging
import asyncio
import dataclasses
import itertools
import random
import struct
@@ -42,6 +43,7 @@ from bumble.hci import (
HCI_LE_1M_PHY,
HCI_SUCCESS,
HCI_UNKNOWN_HCI_COMMAND_ERROR,
HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR,
HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR,
HCI_VERSION_BLUETOOTH_CORE_5_0,
Address,
@@ -53,6 +55,7 @@ from bumble.hci import (
HCI_Connection_Request_Event,
HCI_Disconnection_Complete_Event,
HCI_Encryption_Change_Event,
HCI_Synchronous_Connection_Complete_Event,
HCI_LE_Advertising_Report_Event,
HCI_LE_Connection_Complete_Event,
HCI_LE_Read_Remote_Features_Complete_Event,
@@ -60,10 +63,11 @@ from bumble.hci import (
HCI_Packet,
HCI_Role_Change_Event,
)
from typing import Optional, Union, Dict, TYPE_CHECKING
from typing import Optional, Union, Dict, Any, TYPE_CHECKING
if TYPE_CHECKING:
from bumble.transport.common import TransportSink, TransportSource
from bumble.link import LocalLink
from bumble.transport.common import TransportSink
# -----------------------------------------------------------------------------
# Logging
@@ -79,15 +83,18 @@ class DataObject:
# -----------------------------------------------------------------------------
@dataclasses.dataclass
class Connection:
def __init__(self, controller, handle, role, peer_address, link, transport):
self.controller = controller
self.handle = handle
self.role = role
self.peer_address = peer_address
self.link = link
controller: Controller
handle: int
role: int
peer_address: Address
link: Any
transport: int
link_type: int
def __post_init__(self):
self.assembler = HCI_AclDataPacketAssembler(self.on_acl_pdu)
self.transport = transport
def on_hci_acl_data_packet(self, packet):
self.assembler.feed_packet(packet)
@@ -106,10 +113,10 @@ class Connection:
class Controller:
def __init__(
self,
name,
name: str,
host_source=None,
host_sink: Optional[TransportSink] = None,
link=None,
link: Optional[LocalLink] = None,
public_address: Optional[Union[bytes, str, Address]] = None,
):
self.name = name
@@ -357,12 +364,13 @@ class Controller:
if connection is None:
connection_handle = self.allocate_connection_handle()
connection = Connection(
self,
connection_handle,
BT_PERIPHERAL_ROLE,
peer_address,
self.link,
BT_LE_TRANSPORT,
controller=self,
handle=connection_handle,
role=BT_PERIPHERAL_ROLE,
peer_address=peer_address,
link=self.link,
transport=BT_LE_TRANSPORT,
link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
)
self.peripheral_connections[peer_address] = connection
logger.debug(f'New PERIPHERAL connection handle: 0x{connection_handle:04X}')
@@ -416,12 +424,13 @@ class Controller:
if connection is None:
connection_handle = self.allocate_connection_handle()
connection = Connection(
self,
connection_handle,
BT_CENTRAL_ROLE,
peer_address,
self.link,
BT_LE_TRANSPORT,
controller=self,
handle=connection_handle,
role=BT_CENTRAL_ROLE,
peer_address=peer_address,
link=self.link,
transport=BT_LE_TRANSPORT,
link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
)
self.central_connections[peer_address] = connection
logger.debug(
@@ -566,6 +575,7 @@ class Controller:
peer_address=peer_address,
link=self.link,
transport=BT_BR_EDR_TRANSPORT,
link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
)
self.classic_connections[peer_address] = connection
logger.debug(
@@ -619,6 +629,42 @@ class Controller:
)
)
def on_classic_sco_connection_complete(
self, peer_address: Address, status: int, link_type: int
):
if status == HCI_SUCCESS:
# Allocate (or reuse) a connection handle
connection_handle = self.allocate_connection_handle()
connection = Connection(
controller=self,
handle=connection_handle,
# Role doesn't matter in SCO.
role=BT_CENTRAL_ROLE,
peer_address=peer_address,
link=self.link,
transport=BT_BR_EDR_TRANSPORT,
link_type=link_type,
)
self.classic_connections[peer_address] = connection
logger.debug(f'New SCO connection handle: 0x{connection_handle:04X}')
else:
connection_handle = 0
self.send_hci_packet(
HCI_Synchronous_Connection_Complete_Event(
status=status,
connection_handle=connection_handle,
bd_addr=peer_address,
link_type=link_type,
# TODO: Provide SCO connection parameters.
transmission_interval=0,
retransmission_window=0,
rx_packet_length=0,
tx_packet_length=0,
air_mode=0,
)
)
############################################################
# Advertising support
############################################################
@@ -738,6 +784,68 @@ class Controller:
)
self.link.classic_accept_connection(self, command.bd_addr, command.role)
def on_hci_enhanced_setup_synchronous_connection_command(self, command):
'''
See Bluetooth spec Vol 4, Part E - 7.1.45 Enhanced Setup Synchronous Connection command
'''
if self.link is None:
return
if not (
connection := self.find_classic_connection_by_handle(
command.connection_handle
)
):
self.send_hci_packet(
HCI_Command_Status_Event(
status=HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR,
num_hci_command_packets=1,
command_opcode=command.op_code,
)
)
return
self.send_hci_packet(
HCI_Command_Status_Event(
status=HCI_SUCCESS,
num_hci_command_packets=1,
command_opcode=command.op_code,
)
)
self.link.classic_sco_connect(
self, connection.peer_address, HCI_Connection_Complete_Event.ESCO_LINK_TYPE
)
def on_hci_enhanced_accept_synchronous_connection_request_command(self, command):
'''
See Bluetooth spec Vol 4, Part E - 7.1.46 Enhanced Accept Synchronous Connection Request command
'''
if self.link is None:
return
if not (connection := self.find_classic_connection_by_address(command.bd_addr)):
self.send_hci_packet(
HCI_Command_Status_Event(
status=HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR,
num_hci_command_packets=1,
command_opcode=command.op_code,
)
)
return
self.send_hci_packet(
HCI_Command_Status_Event(
status=HCI_SUCCESS,
num_hci_command_packets=1,
command_opcode=command.op_code,
)
)
self.link.classic_accept_sco_connection(
self, connection.peer_address, HCI_Connection_Complete_Event.ESCO_LINK_TYPE
)
def on_hci_switch_role_command(self, command):
'''
See Bluetooth spec Vol 4, Part E - 7.2.8 Switch Role command

18
bumble/device.py
View File

@@ -61,7 +61,6 @@ from .hci import (
HCI_LE_1M_PHY_BIT,
HCI_LE_2M_PHY,
HCI_LE_2M_PHY_LE_SUPPORTED_FEATURE,
HCI_LE_CLEAR_RESOLVING_LIST_COMMAND,
HCI_LE_CODED_PHY,
HCI_LE_CODED_PHY_BIT,
HCI_LE_CODED_PHY_LE_SUPPORTED_FEATURE,
@@ -86,7 +85,7 @@ from .hci import (
HCI_Constant,
HCI_Create_Connection_Cancel_Command,
HCI_Create_Connection_Command,
HCI_Create_Connection_Command,
HCI_Connection_Complete_Event,
HCI_Disconnect_Command,
HCI_Encryption_Change_Event,
HCI_Error,
@@ -3319,8 +3318,21 @@ class Device(CompositeEventEmitter):
def on_connection_request(self, bd_addr, class_of_device, link_type):
logger.debug(f'*** Connection request: {bd_addr}')
# Handle SCO request.
if link_type in (
HCI_Connection_Complete_Event.SCO_LINK_TYPE,
HCI_Connection_Complete_Event.ESCO_LINK_TYPE,
):
if connection := self.find_connection_by_bd_addr(
bd_addr, transport=BT_BR_EDR_TRANSPORT
):
self.emit('sco_request', connection, link_type)
else:
logger.error(f'SCO request from a non-connected device {bd_addr}')
return
# match a pending future using `bd_addr`
if bd_addr in self.classic_pending_accepts:
elif bd_addr in self.classic_pending_accepts:
future, *_ = self.classic_pending_accepts.pop(bd_addr)
future.set_result((bd_addr, class_of_device, link_type))

6
bumble/hci.py
View File

@@ -4733,7 +4733,11 @@ class HCI_Event(HCI_Packet):
HCI_Object.init_from_bytes(self, parameters, 0, fields)
return self
def __init__(self, event_code, parameters=None, **kwargs):
def __init__(self, event_code=-1, parameters=None, **kwargs):
# Since the legacy implementation relies on an __init__ injector, typing always
# complains that positional argument event_code is not passed, so here sets a
# default value to allow building derived HCI_Event without event_code.
assert event_code != -1
super().__init__(HCI_Event.event_name(event_code))
if (fields := getattr(self, 'fields', None)) and kwargs:
HCI_Object.init_from_fields(self, fields, kwargs)

56
bumble/link.py
View File

@@ -26,9 +26,13 @@ from bumble.hci import (
HCI_SUCCESS,
HCI_CONNECTION_ACCEPT_TIMEOUT_ERROR,
HCI_CONNECTION_TIMEOUT_ERROR,
HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR,
HCI_PAGE_TIMEOUT_ERROR,
HCI_Connection_Complete_Event,
)
from bumble import controller
from typing import Optional, Set
# -----------------------------------------------------------------------------
# Logging
@@ -57,6 +61,8 @@ class LocalLink:
Link bus for controllers to communicate with each other
'''
controllers: Set[controller.Controller]
def __init__(self):
self.controllers = set()
self.pending_connection = None
@@ -79,7 +85,9 @@ class LocalLink:
return controller
return None
def find_classic_controller(self, address):
def find_classic_controller(
self, address: Address
) -> Optional[controller.Controller]:
for controller in self.controllers:
if controller.public_address == address:
return controller
@@ -271,6 +279,52 @@ class LocalLink:
initiator_controller.public_address, int(not (initiator_new_role))
)
def classic_sco_connect(
self,
initiator_controller: controller.Controller,
responder_address: Address,
link_type: int,
):
logger.debug(
f'[Classic] {initiator_controller.public_address} connects SCO to {responder_address}'
)
responder_controller = self.find_classic_controller(responder_address)
# Initiator controller should handle it.
assert responder_controller
responder_controller.on_classic_connection_request(
initiator_controller.public_address,
link_type,
)
def classic_accept_sco_connection(
self,
responder_controller: controller.Controller,
initiator_address: Address,
link_type: int,
):
logger.debug(
f'[Classic] {responder_controller.public_address} accepts to connect SCO {initiator_address}'
)
initiator_controller = self.find_classic_controller(initiator_address)
if initiator_controller is None:
responder_controller.on_classic_sco_connection_complete(
responder_controller.public_address,
HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR,
link_type,
)
return
async def task():
initiator_controller.on_classic_sco_connection_complete(
responder_controller.public_address, HCI_SUCCESS, link_type
)
asyncio.create_task(task())
responder_controller.on_classic_sco_connection_complete(
initiator_controller.public_address, HCI_SUCCESS, link_type
)
# -----------------------------------------------------------------------------
class RemoteLink:

3
bumble/transport/__init__.py
View File

@@ -198,12 +198,13 @@ async def open_transport_or_link(name: str) -> Transport:
"""
if name.startswith('link-relay:'):
logger.warning('Link Relay has been deprecated.')
from ..controller import Controller
from ..link import RemoteLink # lazy import
link = RemoteLink(name[11:])
await link.wait_until_connected()
controller = Controller('remote', link=link)
controller = Controller('remote', link=link) # type:ignore[arg-type]
class LinkTransport(Transport):
async def close(self):

61
tests/hfp_test.py
View File

@@ -23,9 +23,11 @@ import pytest
from typing import Tuple
from .test_utils import TwoDevices
from bumble import core
from bumble import device
from bumble import hfp
from bumble import rfcomm
from bumble import hci
# -----------------------------------------------------------------------------
# Logging
@@ -87,6 +89,63 @@ async def test_slc():
ag_task.cancel()
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_sco_setup():
devices = TwoDevices()
# Enable Classic connections
devices[0].classic_enabled = True
devices[1].classic_enabled = True
# Start
await devices[0].power_on()
await devices[1].power_on()
connections = await asyncio.gather(
devices[0].connect(
devices[1].public_address, transport=core.BT_BR_EDR_TRANSPORT
),
devices[1].accept(devices[0].public_address),
)
def on_sco_request(_connection: device.Connection, _link_type: int):
connections[1].abort_on(
'disconnection',
devices[1].send_command(
hci.HCI_Enhanced_Accept_Synchronous_Connection_Request_Command(
bd_addr=connections[1].peer_address,
**hfp.ESCO_PARAMETERS[
hfp.DefaultCodecParameters.ESCO_CVSD_S1
].asdict(),
)
),
)
devices[1].on('sco_request', on_sco_request)
sco_connections = [
asyncio.get_running_loop().create_future(),
asyncio.get_running_loop().create_future(),
]
devices[0].on(
'sco_connection', lambda sco_link: sco_connections[0].set_result(sco_link)
)
devices[1].on(
'sco_connection', lambda sco_link: sco_connections[1].set_result(sco_link)
)
await devices[0].send_command(
hci.HCI_Enhanced_Setup_Synchronous_Connection_Command(
connection_handle=connections[0].handle,
**hfp.ESCO_PARAMETERS[hfp.DefaultCodecParameters.ESCO_CVSD_S1].asdict(),
)
)
await asyncio.gather(*sco_connections)
# -----------------------------------------------------------------------------
async def run():
await test_slc()

9
tests/test_utils.py
View File

@@ -29,17 +29,18 @@ class TwoDevices:
self.connections = [None, None]
self.link = LocalLink()
addresses = ['F0:F1:F2:F3:F4:F5', 'F5:F4:F3:F2:F1:F0']
self.controllers = [
Controller('C1', link=self.link),
Controller('C2', link=self.link),
Controller('C1', link=self.link, public_address=addresses[0]),
Controller('C2', link=self.link, public_address=addresses[1]),
]
self.devices = [
Device(
address=Address('F0:F1:F2:F3:F4:F5'),
address=Address(addresses[0]),
host=Host(self.controllers[0], AsyncPipeSink(self.controllers[0])),
),
Device(
address=Address('F5:F4:F3:F2:F1:F0'),
address=Address(addresses[1]),
host=Host(self.controllers[1], AsyncPipeSink(self.controllers[1])),
),
]