auracaster/bumble-auracast
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 1 Wiki Activity

add str and scc testcases

Browse Source
This commit is contained in:
Paul Obernesser
2025-12-08 11:41:29 +01:00
parent 6c7b74a0b2
commit 1e30c7e1d7
4 changed files with 157 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/auracast/auracast_config.py
+1
View File
@@ -59,6 +59,7 @@ class AuracastBigConfig(BaseModel):
loop: bool = True
precode_wav: bool = False
iso_que_len: int = 64
num_bis: int = 1 # 1 = mono (FRONT_LEFT), 2 = stereo (FRONT_LEFT + FRONT_RIGHT)
class AuracastBigConfigDeu(AuracastBigConfig):
id: int = 12
src/auracast/multicast.py
+71 -26
View File
@@ -257,6 +257,20 @@ def run_async(async_command: Coroutine) -> None:
color('!!! An error occurred while executing the command:', 'red'), message
)
def _build_bis_list(num_bis: int) -> list:
"""Build BIS list for BasicAudioAnnouncement based on num_bis (1=mono, 2=stereo)."""
locations = [bap.AudioLocation.FRONT_LEFT, bap.AudioLocation.FRONT_RIGHT]
return [
bap.BasicAudioAnnouncement.BIS(
index=idx + 1,
codec_specific_configuration=bap.CodecSpecificConfiguration(
audio_channel_allocation=locations[idx]
),
)
for idx in range(num_bis)
]
async def init_broadcast(
device,
global_config : auracast_config.AuracastGlobalConfig,
@@ -311,14 +325,7 @@ async def init_broadcast(
octets_per_codec_frame=global_config.octets_per_frame,
),
metadata=metadata,
bis=[
bap.BasicAudioAnnouncement.BIS(
index=1,
codec_specific_configuration=bap.CodecSpecificConfiguration(
audio_channel_allocation=bap.AudioLocation.FRONT_LEFT
),
),
],
bis=_build_bis_list(conf.num_bis),
)
],
)
@@ -384,7 +391,7 @@ async def init_broadcast(
big = await device.create_big(
bigs[f'big{i}']['advertising_set'],
parameters=bumble.device.BigParameters(
num_bis=1,
num_bis=conf.num_bis,
sdu_interval=global_config.qos_config.iso_int_multiple_10ms*10000, # Is the same as iso interval
max_sdu=global_config.octets_per_frame,
max_transport_latency=global_config.qos_config.max_transport_latency_ms,
@@ -402,11 +409,18 @@ async def init_broadcast(
direction=bis_link.Direction.HOST_TO_CONTROLLER
)
iso_queue = bumble.device.IsoPacketStream(big.bis_links[0], conf.iso_que_len)
# Create ISO queue(s) - one per BIS
iso_queues = [
bumble.device.IsoPacketStream(link, conf.iso_que_len)
for link in big.bis_links
]
logging.info('Setup ISO Data Path')
bigs[f'big{i}']['iso_queue'] = iso_queue
bigs[f'big{i}']['iso_queues'] = iso_queues
bigs[f'big{i}']['num_bis'] = conf.num_bis
# Keep backward compat: iso_queue points to first queue
bigs[f'big{i}']['iso_queue'] = iso_queues[0]
if global_config.debug:
logging.info(f'big{i} parameters are:')
@@ -638,8 +652,16 @@ class Streamer():
pcm_format = await audio_input.open()
if pcm_format.channels != 1:
logging.info("Input device provides %d channels – will down-mix to mono for LC3", pcm_format.channels)
num_bis = big.get('num_bis', 1)
if num_bis == 2 and pcm_format.channels < 2:
logging.error("Stereo (num_bis=2) requires at least 2 input channels, got %d", pcm_format.channels)
return
if pcm_format.channels != num_bis:
if num_bis == 1:
logging.info("Input device provides %d channels – will down-mix to mono for LC3", pcm_format.channels)
else:
logging.info("Input device provides %d channels – using first %d for stereo", pcm_format.channels, num_bis)
if pcm_format.sample_type == audio_io.PcmFormat.SampleType.INT16:
pcm_bit_depth = 16
elif pcm_format.sample_type == audio_io.PcmFormat.SampleType.FLOAT32:
@@ -648,20 +670,26 @@ class Streamer():
logging.error("Only INT16 and FLOAT32 sample types are supported")
return
encoder = lc3.Encoder(
frame_duration_us=global_config.frame_duration_us,
sample_rate_hz=global_config.auracast_sampling_rate_hz,
num_channels=1,
input_sample_rate_hz=pcm_format.sample_rate,
)
# Create one encoder per BIS (mono: 1 encoder, stereo: 2 encoders)
encoders = [
lc3.Encoder(
frame_duration_us=global_config.frame_duration_us,
sample_rate_hz=global_config.auracast_sampling_rate_hz,
num_channels=1,
input_sample_rate_hz=pcm_format.sample_rate,
)
for _ in range(num_bis)
]
lc3_frame_samples = encoder.get_frame_samples() # number of the pcm samples per lc3 frame
lc3_frame_samples = encoders[0].get_frame_samples() # number of the pcm samples per lc3 frame
big['pcm_bit_depth'] = pcm_bit_depth
big['channels'] = pcm_format.channels
big['lc3_frame_samples'] = lc3_frame_samples
big['lc3_bytes_per_frame'] = global_config.octets_per_frame
big['audio_input'] = audio_input
big['encoder'] = encoder
big['encoders'] = encoders
# Keep backward compat
big['encoder'] = encoders[0]
big['precoded'] = False
logging.info("Streaming audio...")
@@ -686,8 +714,8 @@ class Streamer():
# Ensure frames generator exists (so we can aclose() on stop)
frames_gen = big.get('frames_gen')
if frames_gen is None:
# For stereo, request frame_samples per channel (interleaved input)
frames_gen = big['audio_input'].frames(big['lc3_frame_samples'])
big['frames_gen'] = frames_gen
# Initialize perf tracking bucket per BIG
@@ -713,14 +741,31 @@ class Streamer():
# Measure LC3 encoding time
t1 = time.perf_counter()
lc3_frame = big['encoder'].encode(
pcm_frame, num_bytes=big['lc3_bytes_per_frame'], bit_depth=big['pcm_bit_depth']
)
num_bis = big.get('num_bis', 1)
if num_bis == 1:
# Mono: single encoder, single queue
lc3_frame = big['encoder'].encode(
pcm_frame, num_bytes=big['lc3_bytes_per_frame'], bit_depth=big['pcm_bit_depth']
)
lc3_frames_out = [lc3_frame]
else:
# Stereo: split interleaved PCM into L/R, encode separately
pcm_array = np.frombuffer(pcm_frame, dtype=np.int16)
channels_in = big['channels']
lc3_frames_out = []
for ch_idx, encoder in enumerate(big['encoders']):
# Extract channel (interleaved: L,R,L,R,... or L,R,C,... for >2 ch)
ch_pcm = pcm_array[ch_idx::channels_in].tobytes()
lc3_frame = encoder.encode(
ch_pcm, num_bytes=big['lc3_bytes_per_frame'], bit_depth=big['pcm_bit_depth']
)
lc3_frames_out.append(lc3_frame)
dt_enc = time.perf_counter() - t1
# Measure write blocking time
t2 = time.perf_counter()
await big['iso_queue'].write(lc3_frame)
for q_idx, lc3_frame in enumerate(lc3_frames_out):
await big['iso_queues'][q_idx].write(lc3_frame)
dt_write = time.perf_counter() - t2
# Total loop duration
src/qualification/BAP/BSRC/SCC/test_scc.py
+42
View File
@@ -0,0 +1,42 @@
"""
For BV36-C and BV 37-C to success just restart the stream while the testcase is running
"""
import logging
import os
from auracast.auracast_config import AuracastGlobalConfig, AuracastBigConfig, AuracastQosHigh
from auracast.multicast import broadcast, run_async
if __name__ == "__main__":
logging.basicConfig(
level=os.environ.get("LOG_LEVEL", logging.INFO),
format="%(module)s.py:%(lineno)d %(levelname)s: %(message)s",
)
# Ensure relative audio paths like in AuracastBigConfig work (./auracast/...) from src/auracast/
os.chdir(os.path.join(os.path.dirname(__file__), "../../../../auracast"))
# Start from default global config
config = AuracastGlobalConfig()
# Use same QoS profile as multicast main
config.qos_config = AuracastQosHigh()
# Transport similar to multicast main; adjust if needed for your setup
# config.transport = "auto" # let multicast auto-detect
config.transport = "serial:/dev/ttyAMA3,1000000,rtscts" # Raspberry Pi default
# Default BIG, only modify the random address as requested
big = AuracastBigConfig()
big.random_address = "F1:F1:F2:F3:F4:F5"
big.audio_source = "file:./testdata/announcement_en.wav"
big.id = 12
run_async(
broadcast(
config,
[big],
)
)
src/qualification/BAP/BSRC/STR/test_str.py
+43
View File
@@ -0,0 +1,43 @@
"""
For BV36-C and BV 37-C to success just restart the stream while the testcase is running
"""
import logging
import os
from auracast.auracast_config import AuracastGlobalConfig, AuracastBigConfig, AuracastQosHigh
from auracast.multicast import broadcast, run_async
if __name__ == "__main__":
logging.basicConfig(
level=os.environ.get("LOG_LEVEL", logging.INFO),
format="%(module)s.py:%(lineno)d %(levelname)s: %(message)s",
)
# Ensure relative audio paths like in AuracastBigConfig work (./auracast/...) from src/auracast/
os.chdir(os.path.join(os.path.dirname(__file__), "../../../../auracast"))
# Start from default global config
config = AuracastGlobalConfig()
# Use same QoS profile as multicast main
config.qos_config = AuracastQosHigh()
# Transport similar to multicast main; adjust if needed for your setup
# config.transport = "auto" # let multicast auto-detect
config.transport = "serial:/dev/ttyAMA3,1000000,rtscts" # Raspberry Pi default
# Stereo BIG with 2 BISes (FRONT_LEFT + FRONT_RIGHT)
big = AuracastBigConfig()
big.random_address = "F1:F1:F2:F3:F4:F5"
big.audio_source = "file:./testdata/announcement_en_stereo.wav"
big.id = 12
big.num_bis = 2 # stereo: 2 BISes
run_async(
broadcast(
config,
[big],
)
)