pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

address PR comments

Browse Source
This commit is contained in:
Gilles Boccon-Gibod
2024-10-15 15:26:19 -07:00
parent 009649abd1
commit 0a20f14ea9
13 changed files with 626 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
apps/player/player.py
View File

@@ -25,15 +25,10 @@ from typing import Optional, Union
import click
from bumble.a2dp import (
SBC_JOINT_STEREO_CHANNEL_MODE,
SBC_LOUDNESS_ALLOCATION_METHOD,
make_audio_source_service_sdp_records,
A2DP_SBC_CODEC_TYPE,
A2DP_MPEG_2_4_AAC_CODEC_TYPE,
MPEG_2_AAC_LC_OBJECT_TYPE,
A2DP_NON_A2DP_CODEC_TYPE,
OPUS_VENDOR_ID,
OPUS_CODEC_ID,
AacFrame,
AacParser,
AacPacketSource,
@@ -95,15 +90,37 @@ async def sbc_codec_capabilities(read_function) -> MediaCodecCapabilities:
print(color(f"SBC format: {sbc_frame}", "cyan"))
break
channel_mode = [
SbcMediaCodecInformation.ChannelMode.MONO,
SbcMediaCodecInformation.ChannelMode.DUAL_CHANNEL,
SbcMediaCodecInformation.ChannelMode.STEREO,
SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
][sbc_frame.channel_mode]
block_length = {
4: SbcMediaCodecInformation.BlockLength.BL_4,
8: SbcMediaCodecInformation.BlockLength.BL_8,
12: SbcMediaCodecInformation.BlockLength.BL_12,
16: SbcMediaCodecInformation.BlockLength.BL_16,
}[sbc_frame.block_count]
subbands = {
4: SbcMediaCodecInformation.Subbands.S_4,
8: SbcMediaCodecInformation.Subbands.S_8,
}[sbc_frame.subband_count]
allocation_method = [
SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
SbcMediaCodecInformation.AllocationMethod.SNR,
][sbc_frame.allocation_method]
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_discrete_values(
sampling_frequency=sbc_frame.sampling_frequency,
channel_mode=SBC_JOINT_STEREO_CHANNEL_MODE,
block_length=16,
subbands=8,
allocation_method=SBC_LOUDNESS_ALLOCATION_METHOD,
media_codec_information=SbcMediaCodecInformation(
sampling_frequency=SbcMediaCodecInformation.SamplingFrequency.from_int(
sbc_frame.sampling_frequency
),
channel_mode=channel_mode,
block_length=block_length,
subbands=subbands,
allocation_method=allocation_method,
minimum_bitpool_value=2,
maximum_bitpool_value=40,
),
@@ -119,13 +136,22 @@ async def aac_codec_capabilities(read_function) -> MediaCodecCapabilities:
print(color(f"AAC format: {aac_frame}", "cyan"))
break
sampling_frequency = AacMediaCodecInformation.SamplingFrequency.from_int(
aac_frame.sampling_frequency
)
channels = (
AacMediaCodecInformation.Channels.MONO
if aac_frame.channel_configuration == 1
else AacMediaCodecInformation.Channels.STEREO
)
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_MPEG_2_4_AAC_CODEC_TYPE,
media_codec_information=AacMediaCodecInformation.from_discrete_values(
object_type=MPEG_2_AAC_LC_OBJECT_TYPE,
sampling_frequency=aac_frame.sampling_frequency,
channels=aac_frame.channel_configuration,
media_codec_information=AacMediaCodecInformation(
object_type=AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC,
sampling_frequency=sampling_frequency,
channels=channels,
vbr=1,
bitrate=128000,
),
@@ -149,15 +175,17 @@ async def opus_codec_capabilities(read_function) -> MediaCodecCapabilities:
channel_mode = OpusMediaCodecInformation.ChannelMode.DUAL_MONO
if opus_packet.duration == 10:
frame_size = OpusMediaCodecInformation.FrameSize.F_10MS
frame_size = OpusMediaCodecInformation.FrameSize.FS_10MS
else:
frame_size = OpusMediaCodecInformation.FrameSize.F_20MS
frame_size = OpusMediaCodecInformation.FrameSize.FS_20MS
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_NON_A2DP_CODEC_TYPE,
media_codec_information=OpusMediaCodecInformation.from_discrete_values(
channel_mode=channel_mode, sampling_frequency=48000, frame_size=frame_size
media_codec_information=OpusMediaCodecInformation(
channel_mode=channel_mode,
sampling_frequency=OpusMediaCodecInformation.SamplingFrequency.SF_48000,
frame_size=frame_size,
),
)
@@ -292,6 +320,7 @@ class Player:
vendor_id: int,
codec_id: int,
packet_source: Union[SbcPacketSource, AacPacketSource, OpusPacketSource],
codec_capabilities: MediaCodecCapabilities,
):
# Discover all endpoints on the remote device
endpoints = await protocol.discover_remote_endpoints()
@@ -317,11 +346,25 @@ class Player:
def on_delay_report(delay: int):
print(color(f"*** DELAY REPORT: {delay}", "blue"))
# Adjust the codec capabilities for certain codecs
for capability in sink.capabilities:
if isinstance(capability, MediaCodecCapabilities):
if isinstance(
codec_capabilities.media_codec_information, SbcMediaCodecInformation
) and isinstance(
capability.media_codec_information, SbcMediaCodecInformation
):
codec_capabilities.media_codec_information.minimum_bitpool_value = (
capability.media_codec_information.minimum_bitpool_value
)
codec_capabilities.media_codec_information.maximum_bitpool_value = (
capability.media_codec_information.maximum_bitpool_value
)
print(color("Source media codec:", "green"), codec_capabilities)
# Stream the packets
packet_pump = MediaPacketPump(packet_source.packets)
source = protocol.add_source(
packet_source.codec_capabilities, packet_pump, delay_reporting
)
source = protocol.add_source(codec_capabilities, packet_pump, delay_reporting)
source.on("delay_report", on_delay_report)
stream = await protocol.create_stream(source, sink)
await stream.start()
@@ -358,7 +401,8 @@ class Player:
await device.start_discovery()
async def pair(self, device: Device, address: str) -> None:
connection = await self.connect(device, address)
print(color(f"Connecting to {address}...", "green"))
connection = await device.connect(address, transport=BT_BR_EDR_TRANSPORT)
print(color("Pairing...", "magenta"))
await connection.authenticate()
@@ -418,7 +462,6 @@ class Player:
packet_source = SbcPacketSource(
read_audio_data,
avdtp_protocol.l2cap_channel.peer_mtu,
codec_capabilities,
)
elif audio_format == "aac":
codec_type = A2DP_MPEG_2_4_AAC_CODEC_TYPE
@@ -426,17 +469,15 @@ class Player:
packet_source = AacPacketSource(
read_audio_data,
avdtp_protocol.l2cap_channel.peer_mtu,
codec_capabilities,
)
else:
codec_type = A2DP_NON_A2DP_CODEC_TYPE
vendor_id = OPUS_VENDOR_ID
codec_id = OPUS_CODEC_ID
vendor_id = OpusMediaCodecInformation.VENDOR_ID
codec_id = OpusMediaCodecInformation.CODEC_ID
codec_capabilities = await opus_codec_capabilities(read_audio_data)
packet_source = OpusPacketSource(
read_audio_data,
avdtp_protocol.l2cap_channel.peer_mtu,
codec_capabilities,
)
# Rewind to the start
@@ -444,7 +485,12 @@ class Player:
try:
await self.stream_packets(
avdtp_protocol, codec_type, vendor_id, codec_id, packet_source
avdtp_protocol,
codec_type,
vendor_id,
codec_id,
packet_source,
codec_capabilities,
)
except Exception as error:
print(color(f"!!! Error while streaming: {error}", "red"))
@@ -483,7 +529,7 @@ def create_player(context) -> Player:
default=False,
)
@click.option(
"--encrypt", is_flag=True, help="Request encryption when connecting", default=False
"--encrypt", is_flag=True, help="Request encryption when connecting", default=True
)
def player_cli(ctx, hci_transport, device_config, authenticate, encrypt):
ctx.ensure_object(dict)

50
apps/speaker/speaker.py
View File

@@ -44,25 +44,18 @@ from bumble.avdtp import (
AVDTP_AUDIO_MEDIA_TYPE,
Listener,
MediaCodecCapabilities,
MediaPacket,
Protocol,
)
from bumble.a2dp import (
MPEG_2_AAC_LC_OBJECT_TYPE,
make_audio_sink_service_sdp_records,
A2DP_SBC_CODEC_TYPE,
A2DP_MPEG_2_4_AAC_CODEC_TYPE,
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_SNR_ALLOCATION_METHOD,
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
SbcMediaCodecInformation,
AacMediaCodecInformation,
)
from bumble.utils import AsyncRunner
from bumble.codecs import AacAudioRtpPacket
from bumble.rtp import MediaPacket
# -----------------------------------------------------------------------------
@@ -451,10 +444,12 @@ class Speaker:
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_MPEG_2_4_AAC_CODEC_TYPE,
media_codec_information=AacMediaCodecInformation.from_lists(
object_types=[MPEG_2_AAC_LC_OBJECT_TYPE],
sampling_frequencies=[48000, 44100],
channels=[1, 2],
media_codec_information=AacMediaCodecInformation(
object_type=AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC,
sampling_frequency=AacMediaCodecInformation.SamplingFrequency.SF_48000
| AacMediaCodecInformation.SamplingFrequency.SF_44100,
channels=AacMediaCodecInformation.Channels.MONO
| AacMediaCodecInformation.Channels.STEREO,
vbr=1,
bitrate=256000,
),
@@ -464,20 +459,23 @@ class Speaker:
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_lists(
sampling_frequencies=[48000, 44100, 32000, 16000],
channel_modes=[
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
],
block_lengths=[4, 8, 12, 16],
subbands=[4, 8],
allocation_methods=[
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_SNR_ALLOCATION_METHOD,
],
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,
),

447
bumble/a2dp.py
View File

@@ -22,8 +22,8 @@ import dataclasses
import enum
import logging
import struct
from typing import Awaitable, Callable, Iterable, List
from typing_extensions import Self
from typing import Awaitable, Callable
from typing_extensions import ClassVar, Self
from .codecs import AacAudioRtpPacket
@@ -46,6 +46,7 @@ from .core import (
BT_ADVANCED_AUDIO_DISTRIBUTION_SERVICE,
name_or_number,
)
from .rtp import MediaPacket
# -----------------------------------------------------------------------------
@@ -137,8 +138,6 @@ MPEG_2_4_OBJECT_TYPE_NAMES = {
}
OPUS_VENDOR_ID = 0x000000E0
OPUS_CODEC_ID = 0x0001
OPUS_MAX_FRAMES_IN_RTP_PAYLOAD = 15
# fmt: on
@@ -268,38 +267,61 @@ class SbcMediaCodecInformation:
A2DP spec - 4.3.2 Codec Specific Information Elements
'''
sampling_frequency: int
channel_mode: int
block_length: int
subbands: int
allocation_method: int
sampling_frequency: SamplingFrequency
channel_mode: ChannelMode
block_length: BlockLength
subbands: Subbands
allocation_method: AllocationMethod
minimum_bitpool_value: int
maximum_bitpool_value: int
SAMPLING_FREQUENCY_BITS = {16000: 1 << 3, 32000: 1 << 2, 44100: 1 << 1, 48000: 1}
CHANNEL_MODE_BITS = {
SBC_MONO_CHANNEL_MODE: 1 << 3,
SBC_DUAL_CHANNEL_MODE: 1 << 2,
SBC_STEREO_CHANNEL_MODE: 1 << 1,
SBC_JOINT_STEREO_CHANNEL_MODE: 1,
}
BLOCK_LENGTH_BITS = {4: 1 << 3, 8: 1 << 2, 12: 1 << 1, 16: 1}
SUBBANDS_BITS = {4: 1 << 1, 8: 1}
ALLOCATION_METHOD_BITS = {
SBC_SNR_ALLOCATION_METHOD: 1 << 1,
SBC_LOUDNESS_ALLOCATION_METHOD: 1,
}
class SamplingFrequency(enum.IntFlag):
SF_16000 = 1 << 3
SF_32000 = 1 << 2
SF_44100 = 1 << 1
SF_48000 = 1 << 0
@staticmethod
def from_bytes(data: bytes) -> SbcMediaCodecInformation:
sampling_frequency = (data[0] >> 4) & 0x0F
channel_mode = (data[0] >> 0) & 0x0F
block_length = (data[1] >> 4) & 0x0F
subbands = (data[1] >> 2) & 0x03
allocation_method = (data[1] >> 0) & 0x03
@classmethod
def from_int(cls, sampling_frequency: int) -> Self:
sampling_frequencies = [
16000,
32000,
44100,
48000,
]
index = sampling_frequencies.index(sampling_frequency)
return cls(1 << (len(sampling_frequencies) - index - 1))
class ChannelMode(enum.IntFlag):
MONO = 1 << 3
DUAL_CHANNEL = 1 << 2
STEREO = 1 << 1
JOINT_STEREO = 1 << 0
class BlockLength(enum.IntFlag):
BL_4 = 1 << 3
BL_8 = 1 << 2
BL_12 = 1 << 1
BL_16 = 1 << 0
class Subbands(enum.IntFlag):
S_4 = 1 << 1
S_8 = 1 << 0
class AllocationMethod(enum.IntFlag):
SNR = 1 << 1
LOUDNESS = 1 << 0
@classmethod
def from_bytes(cls, data: bytes) -> Self:
sampling_frequency = cls.SamplingFrequency((data[0] >> 4) & 0x0F)
channel_mode = cls.ChannelMode((data[0] >> 0) & 0x0F)
block_length = cls.BlockLength((data[1] >> 4) & 0x0F)
subbands = cls.Subbands((data[1] >> 2) & 0x03)
allocation_method = cls.AllocationMethod((data[1] >> 0) & 0x03)
minimum_bitpool_value = (data[2] >> 0) & 0xFF
maximum_bitpool_value = (data[3] >> 0) & 0xFF
return SbcMediaCodecInformation(
return cls(
sampling_frequency,
channel_mode,
block_length,
@@ -309,52 +331,6 @@ class SbcMediaCodecInformation:
maximum_bitpool_value,
)
@classmethod
def from_discrete_values(
cls,
sampling_frequency: int,
channel_mode: int,
block_length: int,
subbands: int,
allocation_method: int,
minimum_bitpool_value: int,
maximum_bitpool_value: int,
) -> SbcMediaCodecInformation:
return SbcMediaCodecInformation(
sampling_frequency=cls.SAMPLING_FREQUENCY_BITS[sampling_frequency],
channel_mode=cls.CHANNEL_MODE_BITS[channel_mode],
block_length=cls.BLOCK_LENGTH_BITS[block_length],
subbands=cls.SUBBANDS_BITS[subbands],
allocation_method=cls.ALLOCATION_METHOD_BITS[allocation_method],
minimum_bitpool_value=minimum_bitpool_value,
maximum_bitpool_value=maximum_bitpool_value,
)
@classmethod
def from_lists(
cls,
sampling_frequencies: List[int],
channel_modes: List[int],
block_lengths: List[int],
subbands: List[int],
allocation_methods: List[int],
minimum_bitpool_value: int,
maximum_bitpool_value: int,
) -> SbcMediaCodecInformation:
return SbcMediaCodecInformation(
sampling_frequency=sum(
cls.SAMPLING_FREQUENCY_BITS[x] for x in sampling_frequencies
),
channel_mode=sum(cls.CHANNEL_MODE_BITS[x] for x in channel_modes),
block_length=sum(cls.BLOCK_LENGTH_BITS[x] for x in block_lengths),
subbands=sum(cls.SUBBANDS_BITS[x] for x in subbands),
allocation_method=sum(
cls.ALLOCATION_METHOD_BITS[x] for x in allocation_methods
),
minimum_bitpool_value=minimum_bitpool_value,
maximum_bitpool_value=maximum_bitpool_value,
)
def __bytes__(self) -> bytes:
return bytes(
[
@@ -367,23 +343,6 @@ class SbcMediaCodecInformation:
]
)
def __str__(self) -> str:
channel_modes = ['MONO', 'DUAL_CHANNEL', 'STEREO', 'JOINT_STEREO']
allocation_methods = ['SNR', 'Loudness']
return '\n'.join(
# pylint: disable=line-too-long
[
'SbcMediaCodecInformation(',
f' sampling_frequency: {",".join([str(x) for x in flags_to_list(self.sampling_frequency, SBC_SAMPLING_FREQUENCIES)])}',
f' channel_mode: {",".join([str(x) for x in flags_to_list(self.channel_mode, channel_modes)])}',
f' block_length: {",".join([str(x) for x in flags_to_list(self.block_length, SBC_BLOCK_LENGTHS)])}',
f' subbands: {",".join([str(x) for x in flags_to_list(self.subbands, SBC_SUBBANDS)])}',
f' allocation_method: {",".join([str(x) for x in flags_to_list(self.allocation_method, allocation_methods)])}',
f' minimum_bitpool_value: {self.minimum_bitpool_value}',
f' maximum_bitpool_value: {self.maximum_bitpool_value}' ')',
]
)
# -----------------------------------------------------------------------------
@dataclasses.dataclass
@@ -392,83 +351,66 @@ class AacMediaCodecInformation:
A2DP spec - 4.5.2 Codec Specific Information Elements
'''
object_type: int
sampling_frequency: int
channels: int
rfa: int
object_type: ObjectType
sampling_frequency: SamplingFrequency
channels: Channels
vbr: int
bitrate: int
OBJECT_TYPE_BITS = {
MPEG_2_AAC_LC_OBJECT_TYPE: 1 << 7,
MPEG_4_AAC_LC_OBJECT_TYPE: 1 << 6,
MPEG_4_AAC_LTP_OBJECT_TYPE: 1 << 5,
MPEG_4_AAC_SCALABLE_OBJECT_TYPE: 1 << 4,
}
SAMPLING_FREQUENCY_BITS = {
8000: 1 << 11,
11025: 1 << 10,
12000: 1 << 9,
16000: 1 << 8,
22050: 1 << 7,
24000: 1 << 6,
32000: 1 << 5,
44100: 1 << 4,
48000: 1 << 3,
64000: 1 << 2,
88200: 1 << 1,
96000: 1,
}
CHANNELS_BITS = {1: 1 << 1, 2: 1}
class ObjectType(enum.IntFlag):
MPEG_2_AAC_LC = 1 << 7
MPEG_4_AAC_LC = 1 << 6
MPEG_4_AAC_LTP = 1 << 5
MPEG_4_AAC_SCALABLE = 1 << 4
@staticmethod
def from_bytes(data: bytes) -> AacMediaCodecInformation:
object_type = data[0]
sampling_frequency = (data[1] << 4) | ((data[2] >> 4) & 0x0F)
channels = (data[2] >> 2) & 0x03
rfa = 0
class SamplingFrequency(enum.IntFlag):
SF_8000 = 1 << 11
SF_11025 = 1 << 10
SF_12000 = 1 << 9
SF_16000 = 1 << 8
SF_22050 = 1 << 7
SF_24000 = 1 << 6
SF_32000 = 1 << 5
SF_44100 = 1 << 4
SF_48000 = 1 << 3
SF_64000 = 1 << 2
SF_88200 = 1 << 1
SF_96000 = 1 << 0
@classmethod
def from_int(cls, sampling_frequency: int) -> Self:
sampling_frequencies = [
8000,
11025,
12000,
16000,
22050,
24000,
32000,
44100,
48000,
64000,
88200,
96000,
]
index = sampling_frequencies.index(sampling_frequency)
return cls(1 << (len(sampling_frequencies) - index - 1))
class Channels(enum.IntFlag):
MONO = 1 << 1
STEREO = 1 << 0
@classmethod
def from_bytes(cls, data: bytes) -> AacMediaCodecInformation:
object_type = cls.ObjectType(data[0])
sampling_frequency = cls.SamplingFrequency(
(data[1] << 4) | ((data[2] >> 4) & 0x0F)
)
channels = cls.Channels((data[2] >> 2) & 0x03)
vbr = (data[3] >> 7) & 0x01
bitrate = ((data[3] & 0x7F) << 16) | (data[4] << 8) | data[5]
return AacMediaCodecInformation(
object_type, sampling_frequency, channels, rfa, vbr, bitrate
)
@classmethod
def from_discrete_values(
cls,
object_type: int,
sampling_frequency: int,
channels: int,
vbr: int,
bitrate: int,
) -> AacMediaCodecInformation:
return AacMediaCodecInformation(
object_type=cls.OBJECT_TYPE_BITS[object_type],
sampling_frequency=cls.SAMPLING_FREQUENCY_BITS[sampling_frequency],
channels=cls.CHANNELS_BITS[channels],
rfa=0,
vbr=vbr,
bitrate=bitrate,
)
@classmethod
def from_lists(
cls,
object_types: List[int],
sampling_frequencies: List[int],
channels: List[int],
vbr: int,
bitrate: int,
) -> AacMediaCodecInformation:
return AacMediaCodecInformation(
object_type=sum(cls.OBJECT_TYPE_BITS[x] for x in object_types),
sampling_frequency=sum(
cls.SAMPLING_FREQUENCY_BITS[x] for x in sampling_frequencies
),
channels=sum(cls.CHANNELS_BITS[x] for x in channels),
rfa=0,
vbr=vbr,
bitrate=bitrate,
object_type, sampling_frequency, channels, vbr, bitrate
)
def __bytes__(self) -> bytes:
@@ -483,30 +425,6 @@ class AacMediaCodecInformation:
]
)
def __str__(self) -> str:
object_types = [
'MPEG_2_AAC_LC',
'MPEG_4_AAC_LC',
'MPEG_4_AAC_LTP',
'MPEG_4_AAC_SCALABLE',
'[4]',
'[5]',
'[6]',
'[7]',
]
channels = [1, 2]
# pylint: disable=line-too-long
return '\n'.join(
[
'AacMediaCodecInformation(',
f' object_type: {",".join([str(x) for x in flags_to_list(self.object_type, object_types)])}',
f' sampling_frequency: {",".join([str(x) for x in flags_to_list(self.sampling_frequency, MPEG_2_4_AAC_SAMPLING_FREQUENCIES)])}',
f' channels: {",".join([str(x) for x in flags_to_list(self.channels, channels)])}',
f' vbr: {self.vbr}',
f' bitrate: {self.bitrate}' ')',
]
)
@dataclasses.dataclass
# -----------------------------------------------------------------------------
@@ -542,100 +460,55 @@ class VendorSpecificMediaCodecInformation:
# -----------------------------------------------------------------------------
@dataclasses.dataclass
class OpusMediaCodecInformation(VendorSpecificMediaCodecInformation):
channel_mode: int
frame_size: int
sampling_frequency: int
vendor_id: int = dataclasses.field(init=False, repr=False)
codec_id: int = dataclasses.field(init=False, repr=False)
value: bytes = dataclasses.field(init=False, repr=False)
channel_mode: ChannelMode
frame_size: FrameSize
sampling_frequency: SamplingFrequency
class ChannelMode(enum.IntEnum):
MONO = 0
STEREO = 1
DUAL_MONO = 2
CHANNEL_MODE_BITS = {
ChannelMode.MONO: 1 << 0,
ChannelMode.STEREO: 1 << 1,
ChannelMode.DUAL_MONO: 1 << 2,
}
class ChannelMode(enum.IntFlag):
MONO = 1 << 0
STEREO = 1 << 1
DUAL_MONO = 1 << 2
class FrameSize(enum.IntFlag):
F_10MS = 0
F_20MS = 1
FS_10MS = 1 << 0
FS_20MS = 1 << 1
FRAME_SIZE_BITS = {FrameSize.F_10MS: 1 << 0, FrameSize.F_20MS: 1 << 1}
class SamplingFrequency(enum.IntFlag):
SF_48000 = 1 << 0
SAMPLING_FREQUENCIES = [48000]
SAMPLING_FREQUENCY_BITS = {
48000: 1 << 0,
}
VENDOR_ID: ClassVar[int] = 0x000000E0
CODEC_ID: ClassVar[int] = 0x0001
def __post_init__(self) -> None:
self.vendor_id = self.VENDOR_ID
self.codec_id = self.CODEC_ID
self.value = bytes(
[
self.channel_mode
| (self.frame_size << 3)
| (self.sampling_frequency << 7)
]
)
@classmethod
def from_bytes(cls, data: bytes) -> Self:
"""Create a new instance from the `value` part of the data, not including
the vendor id and codec id"""
channel_mode = data[0] & 0x07
frame_size = (data[0] >> 3) & 0x03
sampling_frequency = (data[0] >> 7) & 0x01
channel_mode = cls.ChannelMode(data[0] & 0x07)
frame_size = cls.FrameSize((data[0] >> 3) & 0x03)
sampling_frequency = cls.SamplingFrequency((data[0] >> 7) & 0x01)
return cls(
OPUS_VENDOR_ID,
OPUS_CODEC_ID,
data,
channel_mode,
frame_size,
sampling_frequency,
)
@classmethod
def from_discrete_values(
cls, channel_mode: ChannelMode, frame_size: FrameSize, sampling_frequency: int
) -> Self:
channel_mode_int = cls.CHANNEL_MODE_BITS[channel_mode]
frame_size_int = cls.FRAME_SIZE_BITS[frame_size]
sampling_frequency_int = cls.SAMPLING_FREQUENCY_BITS[sampling_frequency]
value = bytes(
[channel_mode_int | (frame_size_int << 3) | (sampling_frequency_int << 7)]
)
return cls(
vendor_id=OPUS_VENDOR_ID,
codec_id=OPUS_CODEC_ID,
value=value,
channel_mode=channel_mode_int,
frame_size=frame_size_int,
sampling_frequency=sampling_frequency_int,
)
@classmethod
def from_lists(
cls,
channel_modes: Iterable[ChannelMode],
frame_sizes: Iterable[FrameSize],
sampling_frequencies: Iterable[int],
) -> Self:
channel_mode = sum(channel_modes)
frame_size = sum(frame_sizes)
sampling_frequency = sum(
cls.SAMPLING_FREQUENCY_BITS[x] for x in sampling_frequencies
)
value = bytes([channel_mode | (frame_size << 3) | (sampling_frequency << 7)])
return cls(
vendor_id=OPUS_VENDOR_ID,
codec_id=OPUS_CODEC_ID,
value=value,
channel_mode=channel_mode,
frame_size=frame_size,
sampling_frequency=sampling_frequency,
)
def __str__(self) -> str:
# pylint: disable=line-too-long
return '\n'.join(
[
'OpusMediaCodecInformation(',
f' channel_mode: {",".join([x.name for x in flags_to_list(self.channel_mode, list(self.ChannelMode))])}',
f' frame_size: {",".join([x.name for x in flags_to_list(self.frame_size, list(self.FrameSize))])}',
f' sampling_frequency: {",".join([str(x) for x in flags_to_list(self.sampling_frequency, self.SAMPLING_FREQUENCIES)])}',
]
)
return repr(self)
# -----------------------------------------------------------------------------
@@ -644,6 +517,7 @@ class SbcFrame:
sampling_frequency: int
block_count: int
channel_mode: int
allocation_method: int
subband_count: int
bitpool: int
payload: bytes
@@ -664,6 +538,7 @@ class SbcFrame:
return (
f'SBC(sf={self.sampling_frequency},'
f'cm={self.channel_mode},'
f'am={self.allocation_method},'
f'br={self.bitrate},'
f'sc={self.sample_count},'
f'bp={self.bitpool},'
@@ -695,6 +570,7 @@ class SbcParser:
blocks = 4 * (1 + ((header[1] >> 4) & 3))
channel_mode = (header[1] >> 2) & 3
channels = 1 if channel_mode == SBC_MONO_CHANNEL_MODE else 2
allocation_method = (header[1] >> 1) & 1
subbands = 8 if ((header[1]) & 1) else 4
bitpool = header[2]
@@ -714,7 +590,13 @@ class SbcParser:
# Emit the next frame
yield SbcFrame(
sampling_frequency, blocks, channel_mode, subbands, bitpool, payload
sampling_frequency,
blocks,
channel_mode,
allocation_method,
subbands,
bitpool,
payload,
)
return generate_frames()
@@ -722,19 +604,13 @@ class SbcParser:
# -----------------------------------------------------------------------------
class SbcPacketSource:
def __init__(
self, read: Callable[[int], Awaitable[bytes]], mtu: int, codec_capabilities
) -> None:
def __init__(self, read: Callable[[int], Awaitable[bytes]], mtu: int) -> None:
self.read = read
self.mtu = mtu
self.codec_capabilities = codec_capabilities
@property
def packets(self):
async def generate_packets():
# pylint: disable=import-outside-toplevel
from .avdtp import MediaPacket # Import here to avoid a circular reference
sequence_number = 0
sample_count = 0
frames = []
@@ -870,19 +746,13 @@ class AacParser:
# -----------------------------------------------------------------------------
class AacPacketSource:
def __init__(
self, read: Callable[[int], Awaitable[bytes]], mtu: int, codec_capabilities
) -> None:
def __init__(self, read: Callable[[int], Awaitable[bytes]], mtu: int) -> None:
self.read = read
self.mtu = mtu
self.codec_capabilities = codec_capabilities
@property
def packets(self):
async def generate_packets():
# pylint: disable=import-outside-toplevel
from .avdtp import MediaPacket # Import here to avoid a circular reference
sequence_number = 0
sample_count = 0
@@ -981,11 +851,13 @@ class OpusParser:
raise ValueError(f"version {version} not supported")
header_type = self.HeaderType(header[5])
(granule_position,) = struct.unpack_from("<Q", header, 6)
(bitstream_serial_number,) = struct.unpack_from("<I", header, 14)
(page_sequence_number,) = struct.unpack_from("<I", header, 18)
(crc_checksum,) = struct.unpack_from("<I", header, 22)
page_segments = header[26]
(
granule_position,
bitstream_serial_number,
page_sequence_number,
crc_checksum,
page_segments,
) = struct.unpack_from("<QIIIB", header, 6)
segment_table = await self.read(page_segments)
if header_type & self.HeaderType.FIRST:
@@ -1051,12 +923,9 @@ class OpusParser:
# -----------------------------------------------------------------------------
class OpusPacketSource:
def __init__(
self, read: Callable[[int], Awaitable[bytes]], mtu: int, codec_capabilities
) -> None:
def __init__(self, read: Callable[[int], Awaitable[bytes]], mtu: int) -> None:
self.read = read
self.mtu = mtu
self.codec_capabilities = codec_capabilities
@property
def packets(self):
@@ -1093,5 +962,7 @@ class OpusPacketSource:
# above
# -----------------------------------------------------------------------------
A2DP_VENDOR_MEDIA_CODEC_INFORMATION_CLASSES = {
OPUS_VENDOR_ID: {OPUS_CODEC_ID: OpusMediaCodecInformation}
OpusMediaCodecInformation.VENDOR_ID: {
OpusMediaCodecInformation.CODEC_ID: OpusMediaCodecInformation
}
}

96
bumble/avdtp.py
View File

@@ -17,12 +17,10 @@
# -----------------------------------------------------------------------------
from __future__ import annotations
import asyncio
import struct
import time
import logging
import enum
import warnings
from pyee import EventEmitter
from typing import (
Any,
Awaitable,
@@ -39,6 +37,8 @@ from typing import (
cast,
)
from pyee import EventEmitter
from .core import (
BT_ADVANCED_AUDIO_DISTRIBUTION_SERVICE,
InvalidStateError,
@@ -56,9 +56,11 @@ from .a2dp import (
SbcMediaCodecInformation,
VendorSpecificMediaCodecInformation,
)
from .rtp import MediaPacket
from . import sdp, device, l2cap
from .colors import color
# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
@@ -279,96 +281,6 @@ class RealtimeClock:
await asyncio.sleep(duration)
# -----------------------------------------------------------------------------
class MediaPacket:
@staticmethod
def from_bytes(data: bytes) -> MediaPacket:
version = (data[0] >> 6) & 0x03
padding = (data[0] >> 5) & 0x01
extension = (data[0] >> 4) & 0x01
csrc_count = data[0] & 0x0F
marker = (data[1] >> 7) & 0x01
payload_type = data[1] & 0x7F
sequence_number = struct.unpack_from('>H', data, 2)[0]
timestamp = struct.unpack_from('>I', data, 4)[0]
ssrc = struct.unpack_from('>I', data, 8)[0]
csrc_list = [
struct.unpack_from('>I', data, 12 + i)[0] for i in range(csrc_count)
]
payload = data[12 + csrc_count * 4 :]
return MediaPacket(
version,
padding,
extension,
marker,
sequence_number,
timestamp,
ssrc,
csrc_list,
payload_type,
payload,
)
def __init__(
self,
version: int,
padding: int,
extension: int,
marker: int,
sequence_number: int,
timestamp: int,
ssrc: int,
csrc_list: List[int],
payload_type: int,
payload: bytes,
) -> None:
self.version = version
self.padding = padding
self.extension = extension
self.marker = marker
self.sequence_number = sequence_number & 0xFFFF
self.timestamp = timestamp & 0xFFFFFFFF
self.timestamp_seconds = 0.0
self.ssrc = ssrc
self.csrc_list = csrc_list
self.payload_type = payload_type
self.payload = payload
def __bytes__(self) -> bytes:
header = bytes(
[
self.version << 6
| self.padding << 5
| self.extension << 4
| len(self.csrc_list),
self.marker << 7 | self.payload_type,
]
) + struct.pack(
'>HII',
self.sequence_number,
self.timestamp,
self.ssrc,
)
for csrc in self.csrc_list:
header += struct.pack('>I', csrc)
return header + self.payload
def __str__(self) -> str:
return (
f'RTP(v={self.version},'
f'p={self.padding},'
f'x={self.extension},'
f'm={self.marker},'
f'pt={self.payload_type},'
f'sn={self.sequence_number},'
f'ts={self.timestamp},'
f'ssrc={self.ssrc},'
f'csrcs={self.csrc_list},'
f'payload_size={len(self.payload)})'
)
# -----------------------------------------------------------------------------
class MediaPacketPump:
pump_task: Optional[asyncio.Task]

26
bumble/hci.py
View File

@@ -6065,6 +6065,32 @@ class HCI_Read_Remote_Version_Information_Complete_Event(HCI_Event):
'''
# -----------------------------------------------------------------------------
@HCI_Event.event(
[
('status', STATUS_SPEC),
('connection_handle', 2),
('unused', 1),
(
'service_type',
{
'size': 1,
'mapper': lambda x: HCI_QOS_Setup_Complete_Event.ServiceType(x).name,
},
),
]
)
class HCI_QOS_Setup_Complete_Event(HCI_Event):
'''
See Bluetooth spec @ 7.7.13 QoS Setup Complete Event
'''
class ServiceType(OpenIntEnum):
NO_TRAFFIC_AVAILABLE = 0x00
BEST_EFFORT_AVAILABLE = 0x01
GUARANTEED_AVAILABLE = 0x02
# -----------------------------------------------------------------------------
@HCI_Event.event(
[

12
bumble/host.py
View File

@@ -1106,6 +1106,18 @@ class Host(AbortableEventEmitter):
event.status,
)
def on_hci_qos_setup_complete_event(self, event):
if event.status == hci.HCI_SUCCESS:
self.emit(
'connection_qos_setup', event.connection_handle, event.service_type
)
else:
self.emit(
'connection_qos_setup_failure',
event.connection_handle,
event.status,
)
def on_hci_link_supervision_timeout_changed_event(self, event):
pass

110
bumble/rtp.py Normal file
View File

@@ -0,0 +1,110 @@
# Copyright 2024 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
from __future__ import annotations
import struct
from typing import List
# -----------------------------------------------------------------------------
class MediaPacket:
@staticmethod
def from_bytes(data: bytes) -> MediaPacket:
version = (data[0] >> 6) & 0x03
padding = (data[0] >> 5) & 0x01
extension = (data[0] >> 4) & 0x01
csrc_count = data[0] & 0x0F
marker = (data[1] >> 7) & 0x01
payload_type = data[1] & 0x7F
sequence_number = struct.unpack_from('>H', data, 2)[0]
timestamp = struct.unpack_from('>I', data, 4)[0]
ssrc = struct.unpack_from('>I', data, 8)[0]
csrc_list = [
struct.unpack_from('>I', data, 12 + i)[0] for i in range(csrc_count)
]
payload = data[12 + csrc_count * 4 :]
return MediaPacket(
version,
padding,
extension,
marker,
sequence_number,
timestamp,
ssrc,
csrc_list,
payload_type,
payload,
)
def __init__(
self,
version: int,
padding: int,
extension: int,
marker: int,
sequence_number: int,
timestamp: int,
ssrc: int,
csrc_list: List[int],
payload_type: int,
payload: bytes,
) -> None:
self.version = version
self.padding = padding
self.extension = extension
self.marker = marker
self.sequence_number = sequence_number & 0xFFFF
self.timestamp = timestamp & 0xFFFFFFFF
self.timestamp_seconds = 0.0
self.ssrc = ssrc
self.csrc_list = csrc_list
self.payload_type = payload_type
self.payload = payload
def __bytes__(self) -> bytes:
header = bytes(
[
self.version << 6
| self.padding << 5
| self.extension << 4
| len(self.csrc_list),
self.marker << 7 | self.payload_type,
]
) + struct.pack(
'>HII',
self.sequence_number,
self.timestamp,
self.ssrc,
)
for csrc in self.csrc_list:
header += struct.pack('>I', csrc)
return header + self.payload
def __str__(self) -> str:
return (
f'RTP(v={self.version},'
f'p={self.padding},'
f'x={self.extension},'
f'm={self.marker},'
f'pt={self.payload_type},'
f'sn={self.sequence_number},'
f'ts={self.timestamp},'
f'ssrc={self.ssrc},'
f'csrcs={self.csrc_list},'
f'payload_size={len(self.payload)})'
)

31
examples/run_a2dp_sink.py
View File

@@ -61,20 +61,23 @@ def codec_capabilities():
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_lists(
sampling_frequencies=[48000, 44100, 32000, 16000],
channel_modes=[
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
],
block_lengths=[4, 8, 12, 16],
subbands=[4, 8],
allocation_methods=[
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_SNR_ALLOCATION_METHOD,
],
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,
),

26
examples/run_a2dp_source.py
View File

@@ -33,8 +33,6 @@ from bumble.avdtp import (
Listener,
)
from bumble.a2dp import (
SBC_JOINT_STEREO_CHANNEL_MODE,
SBC_LOUDNESS_ALLOCATION_METHOD,
make_audio_source_service_sdp_records,
A2DP_SBC_CODEC_TYPE,
SbcMediaCodecInformation,
@@ -59,12 +57,12 @@ def codec_capabilities():
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_discrete_values(
sampling_frequency=44100,
channel_mode=SBC_JOINT_STEREO_CHANNEL_MODE,
block_length=16,
subbands=8,
allocation_method=SBC_LOUDNESS_ALLOCATION_METHOD,
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,
),
@@ -73,11 +71,9 @@ def codec_capabilities():
# -----------------------------------------------------------------------------
def on_avdtp_connection(read_function, protocol):
packet_source = SbcPacketSource(
read_function, protocol.l2cap_channel.peer_mtu, codec_capabilities()
)
packet_source = SbcPacketSource(read_function, protocol.l2cap_channel.peer_mtu)
packet_pump = MediaPacketPump(packet_source.packets)
protocol.add_source(packet_source.codec_capabilities, packet_pump)
protocol.add_source(codec_capabilities(), packet_pump)
# -----------------------------------------------------------------------------
@@ -97,11 +93,9 @@ async def stream_packets(read_function, protocol):
print(f'### Selected sink: {sink.seid}')
# Stream the packets
packet_source = SbcPacketSource(
read_function, protocol.l2cap_channel.peer_mtu, codec_capabilities()
)
packet_source = SbcPacketSource(read_function, protocol.l2cap_channel.peer_mtu)
packet_pump = MediaPacketPump(packet_source.packets)
source = protocol.add_source(packet_source.codec_capabilities, packet_pump)
source = protocol.add_source(codec_capabilities(), packet_pump)
stream = await protocol.create_stream(source, sink)
await stream.start()
await asyncio.sleep(5)

31
examples/run_avrcp.py
View File

@@ -60,20 +60,23 @@ def codec_capabilities():
return avdtp.MediaCodecCapabilities(
media_type=avdtp.AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=a2dp.A2DP_SBC_CODEC_TYPE,
media_codec_information=a2dp.SbcMediaCodecInformation.from_lists(
sampling_frequencies=[48000, 44100, 32000, 16000],
channel_modes=[
a2dp.SBC_MONO_CHANNEL_MODE,
a2dp.SBC_DUAL_CHANNEL_MODE,
a2dp.SBC_STEREO_CHANNEL_MODE,
a2dp.SBC_JOINT_STEREO_CHANNEL_MODE,
],
block_lengths=[4, 8, 12, 16],
subbands=[4, 8],
allocation_methods=[
a2dp.SBC_LOUDNESS_ALLOCATION_METHOD,
a2dp.SBC_SNR_ALLOCATION_METHOD,
],
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,
minimum_bitpool_value=2,
maximum_bitpool_value=53,
),

175
tests/a2dp_test.py
View File

@@ -33,20 +33,16 @@ from bumble.avdtp import (
Protocol,
Listener,
MediaCodecCapabilities,
MediaPacket,
AVDTP_AUDIO_MEDIA_TYPE,
AVDTP_TSEP_SNK,
A2DP_SBC_CODEC_TYPE,
)
from bumble.a2dp import (
AacMediaCodecInformation,
OpusMediaCodecInformation,
SbcMediaCodecInformation,
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_SNR_ALLOCATION_METHOD,
)
from bumble.rtp import MediaPacket
# -----------------------------------------------------------------------------
# Logging
@@ -125,12 +121,12 @@ def source_codec_capabilities():
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_discrete_values(
sampling_frequency=44100,
channel_mode=SBC_JOINT_STEREO_CHANNEL_MODE,
block_length=16,
subbands=8,
allocation_method=SBC_LOUDNESS_ALLOCATION_METHOD,
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,
),
@@ -142,20 +138,23 @@ def sink_codec_capabilities():
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_lists(
sampling_frequencies=[48000, 44100, 32000, 16000],
channel_modes=[
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
],
block_lengths=[4, 8, 12, 16],
subbands=[4, 8],
allocation_methods=[
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_SNR_ALLOCATION_METHOD,
],
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,
),
@@ -273,7 +272,125 @@ async def test_source_sink_1():
# -----------------------------------------------------------------------------
async def run_test_self():
def test_sbc_codec_specific_information():
sbc_info = SbcMediaCodecInformation.from_bytes(bytes.fromhex("3fff0235"))
assert (
sbc_info.sampling_frequency
== SbcMediaCodecInformation.SamplingFrequency.SF_44100
| SbcMediaCodecInformation.SamplingFrequency.SF_48000
)
assert (
sbc_info.channel_mode
== SbcMediaCodecInformation.ChannelMode.MONO
| SbcMediaCodecInformation.ChannelMode.DUAL_CHANNEL
| SbcMediaCodecInformation.ChannelMode.STEREO
| SbcMediaCodecInformation.ChannelMode.JOINT_STEREO
)
assert (
sbc_info.block_length
== SbcMediaCodecInformation.BlockLength.BL_4
| SbcMediaCodecInformation.BlockLength.BL_8
| SbcMediaCodecInformation.BlockLength.BL_12
| SbcMediaCodecInformation.BlockLength.BL_16
)
assert (
sbc_info.subbands
== SbcMediaCodecInformation.Subbands.S_4 | SbcMediaCodecInformation.Subbands.S_8
)
assert (
sbc_info.allocation_method
== SbcMediaCodecInformation.AllocationMethod.SNR
| SbcMediaCodecInformation.AllocationMethod.LOUDNESS
)
assert sbc_info.minimum_bitpool_value == 2
assert sbc_info.maximum_bitpool_value == 53
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,
)
assert sbc_info == sbc_info2
assert bytes(sbc_info2) == bytes.fromhex("3fff0235")
# -----------------------------------------------------------------------------
def test_aac_codec_specific_information():
aac_info = AacMediaCodecInformation.from_bytes(bytes.fromhex("f0018c83e800"))
assert (
aac_info.object_type
== 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
== AacMediaCodecInformation.SamplingFrequency.SF_44100
| AacMediaCodecInformation.SamplingFrequency.SF_48000
)
assert (
aac_info.channels
== AacMediaCodecInformation.Channels.MONO
| AacMediaCodecInformation.Channels.STEREO
)
assert aac_info.vbr == 1
assert aac_info.bitrate == 256000
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,
)
assert aac_info == aac_info2
assert bytes(aac_info2) == bytes.fromhex("f0018c83e800")
# -----------------------------------------------------------------------------
def test_opus_codec_specific_information():
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
== 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 async_main():
test_sbc_codec_specific_information()
test_aac_codec_specific_information()
test_opus_codec_specific_information()
await test_self_connection()
await test_source_sink_1()
@@ -281,4 +398,4 @@ async def run_test_self():
# -----------------------------------------------------------------------------
if __name__ == '__main__':
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())
asyncio.run(run_test_self())
asyncio.run(async_main())

3
tests/avdtp_test.py
View File

@@ -23,13 +23,12 @@ from bumble.avdtp import (
AVDTP_MEDIA_TRANSPORT_SERVICE_CATEGORY,
AVDTP_SET_CONFIGURATION,
Message,
MediaPacket,
Get_Capabilities_Response,
Set_Configuration_Command,
Set_Configuration_Response,
ServiceCapabilities,
MediaCodecCapabilities,
)
from bumble.rtp import MediaPacket
# -----------------------------------------------------------------------------

51
web/speaker/speaker.py
View File

@@ -28,26 +28,18 @@ from bumble.avdtp import (
AVDTP_AUDIO_MEDIA_TYPE,
Listener,
MediaCodecCapabilities,
MediaPacket,
Protocol,
)
from bumble.a2dp import (
make_audio_sink_service_sdp_records,
MPEG_2_AAC_LC_OBJECT_TYPE,
A2DP_SBC_CODEC_TYPE,
A2DP_MPEG_2_4_AAC_CODEC_TYPE,
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_SNR_ALLOCATION_METHOD,
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
SbcMediaCodecInformation,
AacMediaCodecInformation,
)
from bumble.utils import AsyncRunner
from bumble.codecs import AacAudioRtpPacket
from bumble.hci import HCI_Reset_Command
from bumble.rtp import MediaPacket
# -----------------------------------------------------------------------------
@@ -130,10 +122,12 @@ class Speaker:
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_MPEG_2_4_AAC_CODEC_TYPE,
media_codec_information=AacMediaCodecInformation.from_lists(
object_types=[MPEG_2_AAC_LC_OBJECT_TYPE],
sampling_frequencies=[48000, 44100],
channels=[1, 2],
media_codec_information=AacMediaCodecInformation(
object_type=AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC,
sampling_frequency=AacMediaCodecInformation.SamplingFrequency.SF_48000
| AacMediaCodecInformation.SamplingFrequency.SF_44100,
channels=AacMediaCodecInformation.Channels.MONO
| AacMediaCodecInformation.Channels.STEREO,
vbr=1,
bitrate=256000,
),
@@ -143,20 +137,23 @@ class Speaker:
return MediaCodecCapabilities(
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_lists(
sampling_frequencies=[48000, 44100, 32000, 16000],
channel_modes=[
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
],
block_lengths=[4, 8, 12, 16],
subbands=[4, 8],
allocation_methods=[
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_SNR_ALLOCATION_METHOD,
],
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,
),