Compare commits

..

1 Commits

Author SHA1 Message Date
khsiao-google
ca23d6b89a Revert "Improve connection related functions and names" 2025-09-10 15:00:41 +08:00
11 changed files with 295 additions and 497 deletions

View File

@@ -49,7 +49,7 @@ jobs:
strategy:
matrix:
python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
rust-version: [ "1.80.0", "stable" ]
rust-version: [ "1.76.0", "stable" ]
fail-fast: false
steps:
- name: Check out from Git
@@ -72,7 +72,7 @@ jobs:
- name: Check License Headers
run: cd rust && cargo run --features dev-tools --bin file-header check-all
- name: Rust Build
run: cd rust && cargo build --all-targets && cargo build-all-features
run: cd rust && cargo build --all-targets && cargo build-all-features --all-targets
# Lints after build so what clippy needs is already built
- name: Rust Lints
run: cd rust && cargo fmt --check && cargo clippy --all-targets -- --deny warnings && cargo clippy --all-features --all-targets -- --deny warnings

View File

@@ -33,6 +33,7 @@ from bumble.hci import (
HCI_COMMAND_DISALLOWED_ERROR,
HCI_COMMAND_PACKET,
HCI_COMMAND_STATUS_PENDING,
HCI_CONNECTION_TIMEOUT_ERROR,
HCI_CONTROLLER_BUSY_ERROR,
HCI_EVENT_PACKET,
HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR,
@@ -87,7 +88,6 @@ class CisLink:
cis_id: int
cig_id: int
acl_connection: Optional[Connection] = None
data_paths: set[int] = dataclasses.field(default_factory=set)
# -----------------------------------------------------------------------------
@@ -381,11 +381,6 @@ class Controller:
return connection
return None
def find_iso_link_by_handle(self, handle: int) -> Optional[CisLink]:
return self.central_cis_links.get(handle) or self.peripheral_cis_links.get(
handle
)
def on_link_central_connected(self, central_address):
'''
Called when an incoming connection occurs from a central on the link
@@ -1858,51 +1853,16 @@ class Controller:
)
)
def on_hci_le_setup_iso_data_path_command(
self, command: hci.HCI_LE_Setup_ISO_Data_Path_Command
) -> bytes:
def on_hci_le_setup_iso_data_path_command(self, command):
'''
See Bluetooth spec Vol 4, Part E - 7.8.109 LE Setup ISO Data Path Command
'''
if not (iso_link := self.find_iso_link_by_handle(command.connection_handle)):
return struct.pack(
'<BH',
HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR,
command.connection_handle,
)
if command.data_path_direction in iso_link.data_paths:
return struct.pack(
'<BH',
HCI_COMMAND_DISALLOWED_ERROR,
command.connection_handle,
)
iso_link.data_paths.add(command.data_path_direction)
return struct.pack('<BH', HCI_SUCCESS, command.connection_handle)
def on_hci_le_remove_iso_data_path_command(
self, command: hci.HCI_LE_Remove_ISO_Data_Path_Command
) -> bytes:
def on_hci_le_remove_iso_data_path_command(self, command):
'''
See Bluetooth spec Vol 4, Part E - 7.8.110 LE Remove ISO Data Path Command
'''
if not (iso_link := self.find_iso_link_by_handle(command.connection_handle)):
return struct.pack(
'<BH',
HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR,
command.connection_handle,
)
data_paths: set[int] = set(
direction
for direction in hci.HCI_LE_Setup_ISO_Data_Path_Command.Direction
if (1 << direction) & command.data_path_direction
)
if not data_paths.issubset(iso_link.data_paths):
return struct.pack(
'<BH',
HCI_COMMAND_DISALLOWED_ERROR,
command.connection_handle,
)
iso_link.data_paths.difference_update(data_paths)
return struct.pack('<BH', HCI_SUCCESS, command.connection_handle)
def on_hci_le_set_host_feature_command(

View File

@@ -2110,6 +2110,23 @@ class AdvertisingData:
return self.to_string()
# -----------------------------------------------------------------------------
# Connection Parameters
# -----------------------------------------------------------------------------
class ConnectionParameters:
def __init__(self, connection_interval, peripheral_latency, supervision_timeout):
self.connection_interval = connection_interval
self.peripheral_latency = peripheral_latency
self.supervision_timeout = supervision_timeout
def __str__(self):
return (
f'ConnectionParameters(connection_interval={self.connection_interval}, '
f'peripheral_latency={self.peripheral_latency}, '
f'supervision_timeout={self.supervision_timeout}'
)
# -----------------------------------------------------------------------------
# Connection PHY
# -----------------------------------------------------------------------------

View File

@@ -1453,8 +1453,6 @@ class _IsoLink:
handle: int
device: Device
sink: Callable[[hci.HCI_IsoDataPacket], Any] | None = None
data_paths: set[_IsoLink.Direction]
_data_path_lock: asyncio.Lock
class Direction(IntEnum):
HOST_TO_CONTROLLER = (
@@ -1464,10 +1462,6 @@ class _IsoLink:
hci.HCI_LE_Setup_ISO_Data_Path_Command.Direction.CONTROLLER_TO_HOST
)
def __init__(self) -> None:
self._data_path_lock = asyncio.Lock()
self.data_paths = set()
async def setup_data_path(
self,
direction: _IsoLink.Direction,
@@ -1488,45 +1482,37 @@ class _IsoLink:
Raises:
HCI_Error: When command complete status is not HCI_SUCCESS.
"""
async with self._data_path_lock:
if direction in self.data_paths:
return
await self.device.send_command(
hci.HCI_LE_Setup_ISO_Data_Path_Command(
connection_handle=self.handle,
data_path_direction=direction,
data_path_id=data_path_id,
codec_id=codec_id or hci.CodingFormat(hci.CodecID.TRANSPARENT),
controller_delay=controller_delay,
codec_configuration=codec_configuration,
),
check_result=True,
)
self.data_paths.add(direction)
await self.device.send_command(
hci.HCI_LE_Setup_ISO_Data_Path_Command(
connection_handle=self.handle,
data_path_direction=direction,
data_path_id=data_path_id,
codec_id=codec_id or hci.CodingFormat(hci.CodecID.TRANSPARENT),
controller_delay=controller_delay,
codec_configuration=codec_configuration,
),
check_result=True,
)
async def remove_data_path(self, directions: Iterable[_IsoLink.Direction]) -> None:
async def remove_data_path(self, directions: Iterable[_IsoLink.Direction]) -> int:
"""Remove a data path with controller on given direction.
Args:
direction: Direction of data path.
Raises:
HCI_Error: When command complete status is not HCI_SUCCESS.
Returns:
Command status.
"""
async with self._data_path_lock:
directions_to_remove = set(directions).intersection(self.data_paths)
if not directions_to_remove:
return
await self.device.send_command(
hci.HCI_LE_Remove_ISO_Data_Path_Command(
connection_handle=self.handle,
data_path_direction=sum(
1 << direction for direction in directions_to_remove
),
response = await self.device.send_command(
hci.HCI_LE_Remove_ISO_Data_Path_Command(
connection_handle=self.handle,
data_path_direction=sum(
1 << direction for direction in set(directions)
),
check_result=True,
)
self.data_paths.difference_update(directions_to_remove)
),
check_result=False,
)
return response.return_parameters.status
def write(self, sdu: bytes) -> None:
"""Write an ISO SDU."""
@@ -1636,8 +1622,7 @@ class CisLink(utils.EventEmitter, _IsoLink):
EVENT_ESTABLISHMENT_FAILURE: ClassVar[str] = "establishment_failure"
def __post_init__(self) -> None:
utils.EventEmitter.__init__(self)
_IsoLink.__init__(self)
super().__init__()
async def disconnect(
self, reason: int = hci.HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR
@@ -1653,7 +1638,6 @@ class BisLink(_IsoLink):
sink: Callable[[hci.HCI_IsoDataPacket], Any] | None = None
def __post_init__(self) -> None:
super().__init__()
self.device = self.big.device
@@ -1798,22 +1782,15 @@ class Connection(utils.CompositeEventEmitter):
@dataclass
class Parameters:
"""
LE connection parameters.
Attributes:
connection_interval: Connection interval, in milliseconds.
peripheral_latency: Peripheral latency, in number of intervals.
supervision_timeout: Supervision timeout, in milliseconds.
subrate_factor: See Bluetooth spec Vol 6, Part B - 4.5.1 Connection events
continuation_number: See Bluetooth spec Vol 6, Part B - 4.5.1 Connection events
"""
connection_interval: float
peripheral_latency: int
supervision_timeout: float
subrate_factor: int = 1
continuation_number: int = 0
connection_interval: float # Connection interval, in milliseconds. [LE only]
peripheral_latency: int # Peripheral latency, in number of intervals. [LE only]
supervision_timeout: float # Supervision timeout, in milliseconds.
subrate_factor: int = (
1 # See Bluetooth spec Vol 6, Part B - 4.5.1 Connection events
)
continuation_number: int = (
0 # See Bluetooth spec Vol 6, Part B - 4.5.1 Connection events
)
def __init__(
self,
@@ -1854,6 +1831,36 @@ class Connection(utils.CompositeEventEmitter):
self.cs_configs = {}
self.cs_procedures = {}
# [Classic only]
@classmethod
def incomplete(cls, device, peer_address, role):
"""
Instantiate an incomplete connection (ie. one waiting for a HCI Connection
Complete event).
Once received it shall be completed using the `.complete` method.
"""
return cls(
device,
None,
PhysicalTransport.BR_EDR,
device.public_address,
None,
peer_address,
None,
role,
None,
)
# [Classic only]
def complete(self, handle, parameters):
"""
Finish an incomplete connection upon completion.
"""
assert self.handle is None
assert self.transport == PhysicalTransport.BR_EDR
self.handle = handle
self.parameters = parameters
@property
def role_name(self):
if self.role is None:
@@ -1865,7 +1872,7 @@ class Connection(utils.CompositeEventEmitter):
return f'UNKNOWN[{self.role}]'
@property
def is_encrypted(self) -> bool:
def is_encrypted(self):
return self.encryption != 0
@property
@@ -2170,6 +2177,8 @@ def with_connection_from_handle(function):
def with_connection_from_address(function):
@functools.wraps(function)
def wrapper(self, address: hci.Address, *args, **kwargs):
if connection := self.pending_connections.get(address, False):
return function(self, connection, *args, **kwargs)
for connection in self.connections.values():
if connection.peer_address == address:
return function(self, connection, *args, **kwargs)
@@ -2183,6 +2192,8 @@ def with_connection_from_address(function):
def try_with_connection_from_address(function):
@functools.wraps(function)
def wrapper(self, address, *args, **kwargs):
if connection := self.pending_connections.get(address, False):
return function(self, connection, address, *args, **kwargs)
for connection in self.connections.values():
if connection.peer_address == address:
return function(self, connection, address, *args, **kwargs)
@@ -2234,7 +2245,7 @@ class Device(utils.CompositeEventEmitter):
scan_response_data: bytes
cs_capabilities: ChannelSoundingCapabilities | None = None
connections: dict[int, Connection]
connection_roles: dict[hci.Address, hci.Role]
pending_connections: dict[hci.Address, Connection]
classic_pending_accepts: dict[
hci.Address,
list[asyncio.Future[Union[Connection, tuple[hci.Address, int, int]]]],
@@ -2356,9 +2367,7 @@ class Device(utils.CompositeEventEmitter):
self.le_connecting = False
self.disconnecting = False
self.connections = {} # Connections, by connection handle
self.connection_roles = (
{}
) # Local connection roles, by BD address (BR/EDR only)
self.pending_connections = {} # Connections, by BD address (BR/EDR only)
self.sco_links = {} # ScoLinks, by connection handle (BR/EDR only)
self.cis_links = {} # CisLinks, by connection handle (LE only)
self._pending_cis = {} # (CIS_ID, CIG_ID), by CIS_handle
@@ -3827,7 +3836,9 @@ class Device(utils.CompositeEventEmitter):
)
else:
# Save pending connection
self.connection_roles[peer_address] = hci.Role.CENTRAL
self.pending_connections[peer_address] = Connection.incomplete(
self, peer_address, hci.Role.CENTRAL
)
# TODO: allow passing other settings
result = await self.send_command(
@@ -3880,7 +3891,7 @@ class Device(utils.CompositeEventEmitter):
self.le_connecting = False
self.connect_own_address_type = None
else:
self.connection_roles.pop(peer_address, None)
self.pending_connections.pop(peer_address, None)
async def accept(
self,
@@ -3974,11 +3985,13 @@ class Device(utils.CompositeEventEmitter):
self.on(self.EVENT_CONNECTION, on_connection)
self.on(self.EVENT_CONNECTION_FAILURE, on_connection_failure)
# Save Peripheral hci.role.
# Save pending connection, with the Peripheral hci.role.
# Even if we requested a role switch in the hci.HCI_Accept_Connection_Request
# command, this connection is still considered Peripheral until an eventual
# role change event.
self.connection_roles[peer_address] = hci.Role.PERIPHERAL
self.pending_connections[peer_address] = Connection.incomplete(
self, peer_address, hci.Role.PERIPHERAL
)
try:
# Accept connection request
@@ -3996,7 +4009,7 @@ class Device(utils.CompositeEventEmitter):
finally:
self.remove_listener(self.EVENT_CONNECTION, on_connection)
self.remove_listener(self.EVENT_CONNECTION_FAILURE, on_connection_failure)
self.connection_roles.pop(peer_address, None)
self.pending_connections.pop(peer_address, None)
@asynccontextmanager
async def connect_as_gatt(self, peer_address: Union[hci.Address, str]):
@@ -5436,49 +5449,15 @@ class Device(utils.CompositeEventEmitter):
self.emit(self.EVENT_CONNECTION, connection)
@host_event_handler
def on_classic_connection(
self,
connection_handle: int,
peer_address: hci.Address,
) -> None:
connection_role = self.connection_roles.pop(peer_address, hci.Role.PERIPHERAL)
logger.debug(
f'*** Connection: [0x{connection_handle:04X}] '
f'{peer_address} {hci.HCI_Constant.role_name(connection_role)}'
)
if connection_handle in self.connections:
logger.warning(
'new connection reuses the same handle as a previous connection'
)
# Create a new connection
connection = Connection(
device=self,
handle=connection_handle,
transport=PhysicalTransport.BR_EDR,
self_address=self.public_address,
self_resolvable_address=None,
peer_address=peer_address,
peer_resolvable_address=None,
role=connection_role,
parameters=Connection.Parameters(0.0, 0, 0.0),
)
self.connections[connection_handle] = connection
self.emit(self.EVENT_CONNECTION, connection)
@host_event_handler
def on_le_connection(
def on_connection(
self,
connection_handle: int,
transport: core.PhysicalTransport,
peer_address: hci.Address,
self_resolvable_address: Optional[hci.Address],
peer_resolvable_address: Optional[hci.Address],
role: hci.Role,
connection_interval: int,
peripheral_latency: int,
supervision_timeout: int,
connection_parameters: Optional[core.ConnectionParameters],
) -> None:
# Convert all-zeros addresses into None.
if self_resolvable_address == hci.Address.ANY_RANDOM:
@@ -5498,6 +5477,19 @@ class Device(utils.CompositeEventEmitter):
'new connection reuses the same handle as a previous connection'
)
if transport == PhysicalTransport.BR_EDR:
# Create a new connection
connection = self.pending_connections.pop(peer_address)
connection.complete(connection_handle, connection_parameters)
self.connections[connection_handle] = connection
# Emit an event to notify listeners of the new connection
self.emit(self.EVENT_CONNECTION, connection)
return
assert connection_parameters is not None
if peer_resolvable_address is None:
# Resolve the peer address if we can
if self.address_resolver:
@@ -5547,16 +5539,16 @@ class Device(utils.CompositeEventEmitter):
connection = Connection(
self,
connection_handle,
PhysicalTransport.LE,
transport,
self_address,
self_resolvable_address,
peer_address,
peer_resolvable_address,
role,
Connection.Parameters(
connection_interval * 1.25,
peripheral_latency,
supervision_timeout * 10.0,
connection_parameters.connection_interval * 1.25,
connection_parameters.peripheral_latency,
connection_parameters.supervision_timeout * 10.0,
),
)
self.connections[connection_handle] = connection
@@ -5647,7 +5639,9 @@ class Device(utils.CompositeEventEmitter):
# device configuration is set to accept any incoming connection
elif self.classic_accept_any:
# Save pending connection
self.connection_roles[bd_addr] = hci.Role.PERIPHERAL
self.pending_connections[bd_addr] = Connection.incomplete(
self, bd_addr, hci.Role.PERIPHERAL
)
self.host.send_command_sync(
hci.HCI_Accept_Connection_Request_Command(
@@ -6189,27 +6183,27 @@ class Device(utils.CompositeEventEmitter):
@host_event_handler
@with_connection_from_handle
def on_connection_parameters_update(
self,
connection: Connection,
connection_interval: int,
peripheral_latency: int,
supervision_timeout: int,
self, connection: Connection, connection_parameters: core.ConnectionParameters
):
logger.debug(
f'*** Connection Parameters Update: [0x{connection.handle:04X}] '
f'{connection.peer_address} as {connection.role_name}, '
f'{connection_parameters}'
)
if connection.parameters.connection_interval != connection_interval * 1.25:
if (
connection.parameters.connection_interval
!= connection_parameters.connection_interval * 1.25
):
connection.parameters = Connection.Parameters(
connection_interval * 1.25,
peripheral_latency,
supervision_timeout * 10.0,
connection_parameters.connection_interval * 1.25,
connection_parameters.peripheral_latency,
connection_parameters.supervision_timeout * 10.0,
)
else:
connection.parameters = Connection.Parameters(
connection_interval * 1.25,
peripheral_latency,
supervision_timeout * 10.0,
connection_parameters.connection_interval * 1.25,
connection_parameters.peripheral_latency,
connection_parameters.supervision_timeout * 10.0,
connection.parameters.subrate_factor,
connection.parameters.continuation_number,
)
@@ -6409,15 +6403,10 @@ class Device(utils.CompositeEventEmitter):
# [Classic only]
@host_event_handler
@try_with_connection_from_address
def on_role_change(
self, connection: Connection, peer_address: hci.Address, new_role: hci.Role
):
if connection:
connection.role = new_role
connection.emit(connection.EVENT_ROLE_CHANGE, new_role)
else:
self.connection_roles[peer_address] = new_role
@with_connection_from_address
def on_role_change(self, connection: Connection, new_role: hci.Role):
connection.role = new_role
connection.emit(connection.EVENT_ROLE_CHANGE, new_role)
# [Classic only]
@host_event_handler

View File

@@ -26,7 +26,12 @@ from typing import TYPE_CHECKING, Any, Awaitable, Callable, Optional, Union, cas
from bumble import drivers, hci, utils
from bumble.colors import color
from bumble.core import ConnectionPHY, InvalidStateError, PhysicalTransport
from bumble.core import (
ConnectionParameters,
ConnectionPHY,
InvalidStateError,
PhysicalTransport,
)
from bumble.l2cap import L2CAP_PDU
from bumble.snoop import Snooper
from bumble.transport.common import TransportLostError
@@ -991,16 +996,20 @@ class Host(utils.EventEmitter):
self.connections[event.connection_handle] = connection
# Notify the client
connection_parameters = ConnectionParameters(
event.connection_interval,
event.peripheral_latency,
event.supervision_timeout,
)
self.emit(
'le_connection',
'connection',
event.connection_handle,
PhysicalTransport.LE,
event.peer_address,
getattr(event, 'local_resolvable_private_address', None),
getattr(event, 'peer_resolvable_private_address', None),
hci.Role(event.role),
event.connection_interval,
event.peripheral_latency,
event.supervision_timeout,
connection_parameters,
)
else:
logger.debug(f'### CONNECTION FAILED: {event.status}')
@@ -1051,9 +1060,14 @@ class Host(utils.EventEmitter):
# Notify the client
self.emit(
'classic_connection',
'connection',
event.connection_handle,
PhysicalTransport.BR_EDR,
event.bd_addr,
None,
None,
None,
None,
)
else:
logger.debug(f'### BR/EDR CONNECTION FAILED: {event.status}')
@@ -1116,13 +1130,14 @@ class Host(utils.EventEmitter):
# Notify the client
if event.status == hci.HCI_SUCCESS:
self.emit(
'connection_parameters_update',
connection.handle,
connection_parameters = ConnectionParameters(
event.connection_interval,
event.peripheral_latency,
event.supervision_timeout,
)
self.emit(
'connection_parameters_update', connection.handle, connection_parameters
)
else:
self.emit(
'connection_parameters_update_failure', connection.handle, event.status

View File

@@ -12,12 +12,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import os
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import logging
import os
import re
from contextlib import asynccontextmanager
from typing import Optional
from bumble import utils
@@ -84,14 +85,12 @@ async def open_transport(name: str) -> Transport:
scheme, *tail = name.split(':', 1)
spec = tail[0] if tail else None
metadata = None
if spec and (m := re.search(r'\[(\w+=\w+(?:,\w+=\w+)*,?)\]', spec)):
metadata_str = m.group(1)
if m.start() == 0:
# <metadata><spec>
spec = spec[m.end() :]
else:
spec = spec[: m.start()]
metadata = dict([entry.split('=') for entry in metadata_str.split(',')])
if spec:
# Metadata may precede the spec
if spec.startswith('['):
metadata_str, *tail = spec[1:].split(']')
spec = tail[0] if tail else None
metadata = dict([entry.split('=') for entry in metadata_str.split(',')])
transport = await _open_transport(scheme, spec)
if metadata:

View File

@@ -10,7 +10,7 @@ documentation = "https://docs.rs/crate/bumble"
authors = ["Marshall Pierce <marshallpierce@google.com>"]
keywords = ["bluetooth", "ble"]
categories = ["api-bindings", "network-programming"]
rust-version = "1.80.0"
rust-version = "1.76.0"
# https://github.com/frewsxcv/cargo-all-features#options
[package.metadata.cargo-all-features]

View File

@@ -18,12 +18,14 @@
import asyncio
import logging
import os
import struct
from typing import Awaitable
import pytest
from bumble import a2dp
from bumble.a2dp import (
AacMediaCodecInformation,
OpusMediaCodecInformation,
SbcMediaCodecInformation,
)
from bumble.avdtp import (
A2DP_SBC_CODEC_TYPE,
AVDTP_AUDIO_MEDIA_TYPE,
@@ -80,24 +82,6 @@ class TwoDevices:
self.paired[which] = keys
# -----------------------------------------------------------------------------
class Data:
pointer: int = 0
data: bytes
def __init__(self, data: bytes):
self.data = data
async def read(self, length: int) -> Awaitable[bytes]:
def generate_read():
end = min(self.pointer + length, len(self.data))
chunk = self.data[self.pointer : end]
self.pointer = end
return chunk
return generate_read()
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_self_connection():
@@ -138,12 +122,12 @@ def source_codec_capabilities():
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=a2dp.SbcMediaCodecInformation(
sampling_frequency=a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_44100,
channel_mode=a2dp.SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
block_length=a2dp.SbcMediaCodecInformation.BlockLength.BL_16,
subbands=a2dp.SbcMediaCodecInformation.Subbands.S_8,
allocation_method=a2dp.SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
media_codec_information=SbcMediaCodecInformation(
sampling_frequency=SbcMediaCodecInformation.SamplingFrequency.SF_44100,
channel_mode=SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
block_length=SbcMediaCodecInformation.BlockLength.BL_16,
subbands=SbcMediaCodecInformation.Subbands.S_8,
allocation_method=SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
minimum_bitpool_value=2,
maximum_bitpool_value=53,
),
@@ -155,23 +139,23 @@ def sink_codec_capabilities():
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=a2dp.SbcMediaCodecInformation(
sampling_frequency=a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_48000
| a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_44100
| a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_32000
| a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_16000,
channel_mode=a2dp.SbcMediaCodecInformation.ChannelMode.MONO
| a2dp.SbcMediaCodecInformation.ChannelMode.DUAL_CHANNEL
| a2dp.SbcMediaCodecInformation.ChannelMode.STEREO
| a2dp.SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
block_length=a2dp.SbcMediaCodecInformation.BlockLength.BL_4
| a2dp.SbcMediaCodecInformation.BlockLength.BL_8
| a2dp.SbcMediaCodecInformation.BlockLength.BL_12
| a2dp.SbcMediaCodecInformation.BlockLength.BL_16,
subbands=a2dp.SbcMediaCodecInformation.Subbands.S_4
| a2dp.SbcMediaCodecInformation.Subbands.S_8,
allocation_method=a2dp.SbcMediaCodecInformation.AllocationMethod.LOUDNESS
| a2dp.SbcMediaCodecInformation.AllocationMethod.SNR,
media_codec_information=SbcMediaCodecInformation(
sampling_frequency=SbcMediaCodecInformation.SamplingFrequency.SF_48000
| SbcMediaCodecInformation.SamplingFrequency.SF_44100
| SbcMediaCodecInformation.SamplingFrequency.SF_32000
| SbcMediaCodecInformation.SamplingFrequency.SF_16000,
channel_mode=SbcMediaCodecInformation.ChannelMode.MONO
| SbcMediaCodecInformation.ChannelMode.DUAL_CHANNEL
| SbcMediaCodecInformation.ChannelMode.STEREO
| SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
block_length=SbcMediaCodecInformation.BlockLength.BL_4
| SbcMediaCodecInformation.BlockLength.BL_8
| SbcMediaCodecInformation.BlockLength.BL_12
| SbcMediaCodecInformation.BlockLength.BL_16,
subbands=SbcMediaCodecInformation.Subbands.S_4
| SbcMediaCodecInformation.Subbands.S_8,
allocation_method=SbcMediaCodecInformation.AllocationMethod.LOUDNESS
| SbcMediaCodecInformation.AllocationMethod.SNR,
minimum_bitpool_value=2,
maximum_bitpool_value=53,
),
@@ -290,54 +274,52 @@ async def test_source_sink_1():
# -----------------------------------------------------------------------------
def test_sbc_codec_specific_information():
sbc_info = a2dp.SbcMediaCodecInformation.from_bytes(bytes.fromhex("3fff0235"))
sbc_info = SbcMediaCodecInformation.from_bytes(bytes.fromhex("3fff0235"))
assert (
sbc_info.sampling_frequency
== a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_44100
| a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_48000
== SbcMediaCodecInformation.SamplingFrequency.SF_44100
| SbcMediaCodecInformation.SamplingFrequency.SF_48000
)
assert (
sbc_info.channel_mode
== a2dp.SbcMediaCodecInformation.ChannelMode.MONO
| a2dp.SbcMediaCodecInformation.ChannelMode.DUAL_CHANNEL
| a2dp.SbcMediaCodecInformation.ChannelMode.STEREO
| a2dp.SbcMediaCodecInformation.ChannelMode.JOINT_STEREO
== SbcMediaCodecInformation.ChannelMode.MONO
| SbcMediaCodecInformation.ChannelMode.DUAL_CHANNEL
| SbcMediaCodecInformation.ChannelMode.STEREO
| SbcMediaCodecInformation.ChannelMode.JOINT_STEREO
)
assert (
sbc_info.block_length
== a2dp.SbcMediaCodecInformation.BlockLength.BL_4
| a2dp.SbcMediaCodecInformation.BlockLength.BL_8
| a2dp.SbcMediaCodecInformation.BlockLength.BL_12
| a2dp.SbcMediaCodecInformation.BlockLength.BL_16
== SbcMediaCodecInformation.BlockLength.BL_4
| SbcMediaCodecInformation.BlockLength.BL_8
| SbcMediaCodecInformation.BlockLength.BL_12
| SbcMediaCodecInformation.BlockLength.BL_16
)
assert (
sbc_info.subbands
== a2dp.SbcMediaCodecInformation.Subbands.S_4
| a2dp.SbcMediaCodecInformation.Subbands.S_8
== SbcMediaCodecInformation.Subbands.S_4 | SbcMediaCodecInformation.Subbands.S_8
)
assert (
sbc_info.allocation_method
== a2dp.SbcMediaCodecInformation.AllocationMethod.SNR
| a2dp.SbcMediaCodecInformation.AllocationMethod.LOUDNESS
== SbcMediaCodecInformation.AllocationMethod.SNR
| SbcMediaCodecInformation.AllocationMethod.LOUDNESS
)
assert sbc_info.minimum_bitpool_value == 2
assert sbc_info.maximum_bitpool_value == 53
sbc_info2 = a2dp.SbcMediaCodecInformation(
a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_44100
| a2dp.SbcMediaCodecInformation.SamplingFrequency.SF_48000,
a2dp.SbcMediaCodecInformation.ChannelMode.MONO
| a2dp.SbcMediaCodecInformation.ChannelMode.DUAL_CHANNEL
| a2dp.SbcMediaCodecInformation.ChannelMode.STEREO
| a2dp.SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
a2dp.SbcMediaCodecInformation.BlockLength.BL_4
| a2dp.SbcMediaCodecInformation.BlockLength.BL_8
| a2dp.SbcMediaCodecInformation.BlockLength.BL_12
| a2dp.SbcMediaCodecInformation.BlockLength.BL_16,
a2dp.SbcMediaCodecInformation.Subbands.S_4
| a2dp.SbcMediaCodecInformation.Subbands.S_8,
a2dp.SbcMediaCodecInformation.AllocationMethod.SNR
| a2dp.SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
sbc_info2 = SbcMediaCodecInformation(
SbcMediaCodecInformation.SamplingFrequency.SF_44100
| SbcMediaCodecInformation.SamplingFrequency.SF_48000,
SbcMediaCodecInformation.ChannelMode.MONO
| SbcMediaCodecInformation.ChannelMode.DUAL_CHANNEL
| SbcMediaCodecInformation.ChannelMode.STEREO
| SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
SbcMediaCodecInformation.BlockLength.BL_4
| SbcMediaCodecInformation.BlockLength.BL_8
| SbcMediaCodecInformation.BlockLength.BL_12
| SbcMediaCodecInformation.BlockLength.BL_16,
SbcMediaCodecInformation.Subbands.S_4 | SbcMediaCodecInformation.Subbands.S_8,
SbcMediaCodecInformation.AllocationMethod.SNR
| SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
2,
53,
)
@@ -347,36 +329,36 @@ def test_sbc_codec_specific_information():
# -----------------------------------------------------------------------------
def test_aac_codec_specific_information():
aac_info = a2dp.AacMediaCodecInformation.from_bytes(bytes.fromhex("f0018c83e800"))
aac_info = AacMediaCodecInformation.from_bytes(bytes.fromhex("f0018c83e800"))
assert (
aac_info.object_type
== a2dp.AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC
| a2dp.AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LC
| a2dp.AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LTP
| a2dp.AacMediaCodecInformation.ObjectType.MPEG_4_AAC_SCALABLE
== AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC
| AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LC
| AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LTP
| AacMediaCodecInformation.ObjectType.MPEG_4_AAC_SCALABLE
)
assert (
aac_info.sampling_frequency
== a2dp.AacMediaCodecInformation.SamplingFrequency.SF_44100
| a2dp.AacMediaCodecInformation.SamplingFrequency.SF_48000
== AacMediaCodecInformation.SamplingFrequency.SF_44100
| AacMediaCodecInformation.SamplingFrequency.SF_48000
)
assert (
aac_info.channels
== a2dp.AacMediaCodecInformation.Channels.MONO
| a2dp.AacMediaCodecInformation.Channels.STEREO
== AacMediaCodecInformation.Channels.MONO
| AacMediaCodecInformation.Channels.STEREO
)
assert aac_info.vbr == 1
assert aac_info.bitrate == 256000
aac_info2 = a2dp.AacMediaCodecInformation(
a2dp.AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC
| a2dp.AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LC
| a2dp.AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LTP
| a2dp.AacMediaCodecInformation.ObjectType.MPEG_4_AAC_SCALABLE,
a2dp.AacMediaCodecInformation.SamplingFrequency.SF_44100
| a2dp.AacMediaCodecInformation.SamplingFrequency.SF_48000,
a2dp.AacMediaCodecInformation.Channels.MONO
| a2dp.AacMediaCodecInformation.Channels.STEREO,
aac_info2 = AacMediaCodecInformation(
AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC
| AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LC
| AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LTP
| AacMediaCodecInformation.ObjectType.MPEG_4_AAC_SCALABLE,
AacMediaCodecInformation.SamplingFrequency.SF_44100
| AacMediaCodecInformation.SamplingFrequency.SF_48000,
AacMediaCodecInformation.Channels.MONO
| AacMediaCodecInformation.Channels.STEREO,
1,
256000,
)
@@ -386,159 +368,25 @@ def test_aac_codec_specific_information():
# -----------------------------------------------------------------------------
def test_opus_codec_specific_information():
opus_info = a2dp.OpusMediaCodecInformation.from_bytes(bytes([0x92]))
assert opus_info.vendor_id == a2dp.OpusMediaCodecInformation.VENDOR_ID
assert opus_info.codec_id == a2dp.OpusMediaCodecInformation.CODEC_ID
assert opus_info.frame_size == a2dp.OpusMediaCodecInformation.FrameSize.FS_20MS
assert opus_info.channel_mode == a2dp.OpusMediaCodecInformation.ChannelMode.STEREO
opus_info = OpusMediaCodecInformation.from_bytes(bytes([0x92]))
assert opus_info.vendor_id == OpusMediaCodecInformation.VENDOR_ID
assert opus_info.codec_id == OpusMediaCodecInformation.CODEC_ID
assert opus_info.frame_size == OpusMediaCodecInformation.FrameSize.FS_20MS
assert opus_info.channel_mode == OpusMediaCodecInformation.ChannelMode.STEREO
assert (
opus_info.sampling_frequency
== a2dp.OpusMediaCodecInformation.SamplingFrequency.SF_48000
== OpusMediaCodecInformation.SamplingFrequency.SF_48000
)
opus_info2 = a2dp.OpusMediaCodecInformation(
a2dp.OpusMediaCodecInformation.ChannelMode.STEREO,
a2dp.OpusMediaCodecInformation.FrameSize.FS_20MS,
a2dp.OpusMediaCodecInformation.SamplingFrequency.SF_48000,
opus_info2 = OpusMediaCodecInformation(
OpusMediaCodecInformation.ChannelMode.STEREO,
OpusMediaCodecInformation.FrameSize.FS_20MS,
OpusMediaCodecInformation.SamplingFrequency.SF_48000,
)
assert opus_info2 == opus_info
assert opus_info2.value == bytes([0x92])
# -----------------------------------------------------------------------------
async def test_sbc_parser():
header = b'\x9c\x80\x08\x00'
payload = b'\x00\x00\x00\x00\x00\x00'
data = Data(header + payload)
parser = a2dp.SbcParser(data.read)
async for frame in parser.frames:
assert frame.sampling_frequency == 44100
assert frame.block_count == 4
assert frame.channel_mode == 0
assert frame.allocation_method == 0
assert frame.subband_count == 4
assert frame.bitpool == 8
assert frame.payload == header + payload
# -----------------------------------------------------------------------------
async def test_sbc_packet_source():
header = b'\x9c\x80\x08\x00'
payload = b'\x00\x00\x00\x00\x00\x00'
data = Data((header + payload) * 2)
packet_source = a2dp.SbcPacketSource(data.read, 23)
async for packet in packet_source.packets:
assert packet.sequence_number == 0
assert packet.timestamp == 0
assert packet.payload == b'\x01' + header + payload
# -----------------------------------------------------------------------------
async def test_aac_parser():
header = b'\xff\xf0\x10\x00\x01\xa0\x00'
payload = b'\x00\x00\x00\x00\x00\x00'
data = Data(header + payload)
parser = a2dp.AacParser(data.read)
async for frame in parser.frames:
assert frame.profile == a2dp.AacFrame.Profile.MAIN
assert frame.sampling_frequency == 44100
assert frame.channel_configuration == 0
assert frame.payload == payload
# -----------------------------------------------------------------------------
async def test_aac_packet_source():
header = b'\xff\xf0\x10\x00\x01\xa0\x00'
payload = b'\x00\x00\x00\x00\x00\x00'
data = Data(header + payload)
packet_source = a2dp.AacPacketSource(data.read, 0)
async for packet in packet_source.packets:
assert packet.sequence_number == 0
assert packet.timestamp == 0
assert packet.payload == b' \x00\x12\x00\x00\x000\x00\x00\x00\x00\x00\x00'
# -----------------------------------------------------------------------------
async def test_opus_parser():
packed_header_data_revised = struct.pack(
"<QIIIB",
0, # granule_position
2, # bitstream_serial_number
2, # page_sequence_number
0, # crc_checksum
3, # page_segments
)
first_page_header_revised = (
b'OggS' # Capture pattern
+ b'\x00' # Version
+ b'\x02' # Header type
+ packed_header_data_revised
)
segment_table_revised = b'\x0a\x08\x0a'
opus_head_packet_data = b'OpusHead' + b'\x00' + b'\x00'
opus_tags_packet_data = b'OpusTags'
audio_data_packet = b'0123456789'
data = Data(
first_page_header_revised
+ segment_table_revised
+ opus_head_packet_data
+ opus_tags_packet_data
+ audio_data_packet
)
parser = a2dp.OpusParser(data.read)
async for packet in parser.packets:
assert packet.channel_mode == a2dp.OpusPacket.ChannelMode.STEREO
assert packet.payload == audio_data_packet
# -----------------------------------------------------------------------------
async def test_opus_packet_source():
packed_header_data_revised = struct.pack(
"<QIIIB",
0, # granule_position
2, # bitstream_serial_number
2, # page_sequence_number
0, # crc_checksum
3, # page_segments
)
first_page_header_revised = (
b'OggS' # Capture pattern
+ b'\x00' # Version
+ b'\x02' # Header type
+ packed_header_data_revised
)
segment_table_revised = b'\x0a\x08\x0a'
opus_head_packet_data = b'OpusHead' + b'\x00' + b'\x00'
opus_tags_packet_data = b'OpusTags'
audio_data_packet = b'0123456789'
data = Data(
first_page_header_revised
+ segment_table_revised
+ opus_head_packet_data
+ opus_tags_packet_data
+ audio_data_packet
)
parser = a2dp.OpusPacketSource(data.read, 0)
async for packet in parser.packets:
assert packet.sequence_number == 0
assert packet.timestamp == 0
assert packet.payload == b'\x01' + audio_data_packet
# -----------------------------------------------------------------------------
async def async_main():
test_sbc_codec_specific_information()
@@ -546,12 +394,6 @@ async def async_main():
test_opus_codec_specific_information()
await test_self_connection()
await test_source_sink_1()
test_sbc_parser()
test_sbc_packet_source()
test_aac_parser()
test_aac_packet_source()
test_opus_parser()
test_opus_packet_source()
# -----------------------------------------------------------------------------

View File

@@ -310,12 +310,12 @@ async def test_pacs():
@pytest.mark.asyncio
async def test_ascs():
devices = TwoDevices()
devices[1].add_service(
AudioStreamControlService(device=devices[1], sink_ase_id=[1, 2])
devices[0].add_service(
AudioStreamControlService(device=devices[0], sink_ase_id=[1, 2])
)
await devices.setup_connection()
peer = device.Peer(devices.connections[0])
peer = device.Peer(devices.connections[1])
ascs_client = await peer.discover_service_and_create_proxy(
AudioStreamControlServiceProxy
)
@@ -369,7 +369,7 @@ async def test_ascs():
await ascs_client.ase_control_point.write_value(
ASE_Config_QOS(
ase_id=[1, 2],
cig_id=[1, 1],
cig_id=[1, 2],
cis_id=[3, 4],
sdu_interval=[5, 6],
framing=[0, 1],
@@ -402,19 +402,25 @@ async def test_ascs():
)
# CIS establishment
cis_handles = await devices[0].setup_cig(
device.CigParameters(
devices[0].emit(
'cis_establishment',
device.CisLink(
device=devices[0],
acl_connection=devices.connections[0],
handle=5,
cis_id=3,
cig_id=1,
cis_parameters=[
device.CigParameters.CisParameters(cis_id=3),
device.CigParameters.CisParameters(cis_id=4),
],
sdu_interval_c_to_p=0,
sdu_interval_p_to_c=0,
)
),
)
await devices[0].create_cis(
[(cis_handle, devices.connections[0]) for cis_handle in cis_handles]
devices[0].emit(
'cis_establishment',
device.CisLink(
device=devices[0],
acl_connection=devices.connections[0],
handle=6,
cis_id=4,
cig_id=2,
),
)
assert (await notifications[1].get())[:2] == bytes(
[1, AseStateMachine.State.STREAMING]

View File

@@ -24,7 +24,7 @@ from unittest import mock
import pytest
from bumble import device, gatt, hci, utils
from bumble.core import PhysicalTransport
from bumble.core import ConnectionParameters, PhysicalTransport
from bumble.device import (
AdvertisingEventProperties,
AdvertisingParameters,
@@ -289,15 +289,14 @@ async def test_legacy_advertising_disconnection(auto_restart):
await device.power_on()
peer_address = Address('F0:F1:F2:F3:F4:F5')
await device.start_advertising(auto_restart=auto_restart)
device.on_le_connection(
device.on_connection(
0x0001,
PhysicalTransport.LE,
peer_address,
None,
None,
Role.PERIPHERAL,
0,
0,
0,
ConnectionParameters(0, 0, 0),
)
device.on_advertising_set_termination(
@@ -348,15 +347,14 @@ async def test_extended_advertising_connection(own_address_type):
advertising_set = await device.create_advertising_set(
advertising_parameters=AdvertisingParameters(own_address_type=own_address_type)
)
device.on_le_connection(
device.on_connection(
0x0001,
PhysicalTransport.LE,
peer_address,
None,
None,
Role.PERIPHERAL,
0,
0,
0,
ConnectionParameters(0, 0, 0),
)
device.on_advertising_set_termination(
HCI_SUCCESS,
@@ -393,15 +391,14 @@ async def test_extended_advertising_connection_out_of_order(own_address_type):
0x0001,
0,
)
device.on_le_connection(
device.on_connection(
0x0001,
PhysicalTransport.LE,
Address('F0:F1:F2:F3:F4:F5'),
None,
None,
Role.PERIPHERAL,
0,
0,
0,
ConnectionParameters(0, 0, 0),
)
if own_address_type == OwnAddressType.PUBLIC:

View File

@@ -24,7 +24,7 @@ import sys
import pytest
from bumble import controller, device, hci, link, transport
from bumble.transport import common
from bumble.transport.common import PacketParser
# -----------------------------------------------------------------------------
@@ -61,9 +61,9 @@ class Sink:
# -----------------------------------------------------------------------------
def test_parser():
sink1 = Sink()
parser1 = common.PacketParser(sink1)
parser1 = PacketParser(sink1)
sink2 = Sink()
parser2 = common.PacketParser(sink2)
parser2 = PacketParser(sink2)
for parser in [parser1, parser2]:
with open(
@@ -82,7 +82,7 @@ def test_parser():
# -----------------------------------------------------------------------------
def test_parser_extensions():
sink = Sink()
parser = common.PacketParser(sink)
parser = PacketParser(sink)
# Check that an exception is thrown for an unknown type
try:
@@ -206,7 +206,7 @@ async def test_unix_connection_abstract():
# -----------------------------------------------------------------------------
@pytest.mark.parametrize(
"address,",
("127.0.0.1", "[::1]"),
("127.0.0.1",),
)
async def test_android_netsim_connection(address):
controller_transport = await transport.open_transport(
@@ -222,33 +222,6 @@ async def test_android_netsim_connection(address):
await client_device.power_on()
await client_transport.close()
await controller_transport.source.grpc_server.stop(None)
await controller_transport.close()
# -----------------------------------------------------------------------------
@pytest.mark.parametrize(
"spec,",
(
"android-netsim:[::1]:{port},mode=host[a=b,c=d]",
"android-netsim:localhost:{port},mode=host[a=b,c=d]",
"android-netsim:[a=b,c=d][::1]:{port},mode=host",
"android-netsim:[a=b,c=d]localhost:{port},mode=host",
),
)
async def test_open_transport_with_metadata(spec):
controller_transport = await transport.open_transport(
"android-netsim:_:0,mode=controller"
)
port = controller_transport.source.port
_make_controller_from_transport(controller_transport)
client_transport = await transport.open_transport(spec.format(port=port))
assert client_transport.source.metadata['a'] == 'b'
assert client_transport.source.metadata['c'] == 'd'
await client_transport.close()
await controller_transport.source.grpc_server.stop(None)
await controller_transport.close()