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

Implement basic support for multiple broadcasters

Browse Source
This commit is contained in:
pstruebi
2025-01-25 17:06:54 +01:00
parent 292cacb397
commit 36af8c3eb4
1 changed files with 111 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
auracast/auracast.py
View File

@@ -121,115 +121,124 @@ async def run_broadcast(
logger.error(color('Periodic advertising not supported', 'red'))
return
with wave.open(big_config[0].broacast_wav_file_path, 'rb') as wav:
logger.info('Encoding wav file into lc3...')
logger.info('Frame rate of .wav file is: %s', wav.getframerate())
encoder = lc3.Encoder(
frame_duration_us=global_config.frame_duration_us,
sample_rate_hz=global_config.auracast_sampling_rate_khz,
num_channels=1,
input_sample_rate_hz=wav.getframerate(),
)
frames = list[bytes]()
while pcm := wav.readframes(encoder.get_frame_samples()):
frames.append(
encoder.encode(pcm, num_bytes=global_config.octets_per_frame, bit_depth=wav.getsampwidth() * 8)
)
del encoder
print('Encoding complete.')
# Config advertising set
bap_sampling_freq = getattr(bap.SamplingFrequency, f"FREQ_{global_config.auracast_sampling_rate_khz}")
basic_audio_announcement = bap.BasicAudioAnnouncement(
presentation_delay=global_config.presentation_delay_us,
subgroups=[
bap.BasicAudioAnnouncement.Subgroup(
codec_id=hci.CodingFormat(codec_id=hci.CodecID.LC3),
codec_specific_configuration=bap.CodecSpecificConfiguration(
sampling_frequency=bap_sampling_freq,
frame_duration=bap.FrameDuration.DURATION_10000_US,
octets_per_codec_frame=global_config.octets_per_frame,
),
metadata=le_audio.Metadata(
[
le_audio.Metadata.Entry(
tag=le_audio.Metadata.Tag.LANGUAGE, data=big_config[0].broadcast_language.encode()
),
le_audio.Metadata.Entry(
tag=le_audio.Metadata.Tag.PROGRAM_INFO, data=big_config[0].broadcast_program_info.encode()
),
]
),
bis=[
bap.BasicAudioAnnouncement.BIS(
index=1,
codec_specific_configuration=bap.CodecSpecificConfiguration(
audio_channel_allocation=bap.AudioLocation.FRONT_LEFT
),
),
],
bigs = {}
for i, conf in enumerate(big_config):
bigs[f'big{i}'] = {}
with wave.open(conf.broacast_wav_file_path, 'rb') as wav:
logger.info('Encoding wav file into lc3...')
logger.info('Frame rate of .wav file is: %s', wav.getframerate())
encoder = lc3.Encoder(
frame_duration_us=global_config.frame_duration_us,
sample_rate_hz=global_config.auracast_sampling_rate_khz,
num_channels=1,
input_sample_rate_hz=wav.getframerate(),
)
],
)
logging.info('Setup Advertising')
broadcast_audio_announcement = bap.BroadcastAudioAnnouncement(big_config[0].broadcast_id)
advertising_set0 = await device.create_advertising_set(
advertising_parameters=bumble.device.AdvertisingParameters(
advertising_event_properties=bumble.device.AdvertisingEventProperties(
is_connectable=False
),
primary_advertising_interval_min=100,
primary_advertising_interval_max=200,
advertising_sid=0
# TODO: use 2mbit phy
),
advertising_data=(
broadcast_audio_announcement.get_advertising_data()
+ bytes(
core.AdvertisingData(
[(core.AdvertisingData.BROADCAST_NAME, big_config[0].broadcast_name.encode())]
frames = list[bytes]()
while pcm := wav.readframes(encoder.get_frame_samples()):
frames.append(
encoder.encode(pcm, num_bytes=global_config.octets_per_frame, bit_depth=wav.getsampwidth() * 8)
)
)
),
periodic_advertising_parameters=bumble.device.PeriodicAdvertisingParameters(
periodic_advertising_interval_min=80,
periodic_advertising_interval_max=160,
),
periodic_advertising_data=basic_audio_announcement.get_advertising_data(),
auto_restart=True,
auto_start=True,
)
del encoder
logger.info('Encoding complete.')
logging.info('Start Periodic Advertising')
await advertising_set0.start_periodic()
bigs.update()
bigs[f'big{i}']['frames'] = frames
logging.info('Setup BIG')
big0 = await device.create_big(
advertising_set0,
parameters=bumble.device.BigParameters(
num_bis=1,
sdu_interval=global_config.frame_duration_us,
max_sdu=global_config.octets_per_frame, # is this octets per frame ?
max_transport_latency=65,
rtn=4,
broadcast_code=(
bytes.fromhex(big_config[0].broadcast_code) if big_config[0].broadcast_code else None
),
),
)
logging.info('Setup ISO Data Path')
for bis_link in big0.bis_links:
await bis_link.setup_data_path(
direction=bis_link.Direction.HOST_TO_CONTROLLER
# Config advertising set
bigs[f'big{i}']['basic_audio_announcement'] = bap.BasicAudioAnnouncement(
presentation_delay=global_config.presentation_delay_us,
subgroups=[
bap.BasicAudioAnnouncement.Subgroup(
codec_id=hci.CodingFormat(codec_id=hci.CodecID.LC3),
codec_specific_configuration=bap.CodecSpecificConfiguration(
sampling_frequency=bap_sampling_freq,
frame_duration=bap.FrameDuration.DURATION_10000_US,
octets_per_codec_frame=global_config.octets_per_frame,
),
metadata=le_audio.Metadata(
[
le_audio.Metadata.Entry(
tag=le_audio.Metadata.Tag.LANGUAGE, data=conf.broadcast_language.encode()
),
le_audio.Metadata.Entry(
tag=le_audio.Metadata.Tag.PROGRAM_INFO, data=conf.broadcast_program_info.encode()
),
]
),
bis=[
bap.BasicAudioAnnouncement.BIS(
index=1,
codec_specific_configuration=bap.CodecSpecificConfiguration(
audio_channel_allocation=bap.AudioLocation.FRONT_LEFT
),
),
],
)
],
)
logging.info('Setup Advertising')
bigs[f'big{i}']['broadcast_audio_announcement'] = bap.BroadcastAudioAnnouncement(conf.broadcast_id)
advertising_set = await device.create_advertising_set(
advertising_parameters=bumble.device.AdvertisingParameters(
advertising_event_properties=bumble.device.AdvertisingEventProperties(
is_connectable=False
),
primary_advertising_interval_min=100,
primary_advertising_interval_max=200,
advertising_sid=0
# TODO: use 2mbit phy
),
advertising_data=(
bigs[f'big{i}']['broadcast_audio_announcement'].get_advertising_data()
+ bytes(
core.AdvertisingData(
[(core.AdvertisingData.BROADCAST_NAME, conf.broadcast_name.encode())]
)
)
),
periodic_advertising_parameters=bumble.device.PeriodicAdvertisingParameters(
periodic_advertising_interval_min=80,
periodic_advertising_interval_max=160,
),
periodic_advertising_data=bigs[f'big{i}']['basic_audio_announcement'].get_advertising_data(),
auto_restart=True,
auto_start=True,
)
bigs[f'big{i}']['advertising_set'] = advertising_set
frames_iterator = itertools.cycle(frames)
logging.info('Start Periodic Advertising')
await bigs[f'big{i}']['advertising_set'].start_periodic()
logging.info('Setup BIG')
big = await device.create_big(
bigs[f'big{i}']['advertising_set'] ,
parameters=bumble.device.BigParameters(
num_bis=1,
sdu_interval=global_config.frame_duration_us,
max_sdu=global_config.octets_per_frame, # is this octets per frame ?
max_transport_latency=65,
rtn=4,
broadcast_code=(
bytes.fromhex(conf.broadcast_code) if conf.broadcast_code else None
),
),
)
bigs[f'big{i}']['big'] = big
logging.info('Setup ISO Data Path')
for bis_link in big.bis_links:
await bis_link.setup_data_path(
direction=bis_link.Direction.HOST_TO_CONTROLLER
)
bigs[f'big{i}']['frames_iterator'] = itertools.cycle(frames)
logging.info("Broadcasting...")
def on_packet_complete(event):
frame = next(frames_iterator)
big0.bis_links[0].write(frame)
for val in bigs.values():
frame = next(val['frames_iterator'] )
val['big'].bis_links[0].write(frame)
device.host.on('hci_number_of_completed_packets_event', on_packet_complete)
@@ -259,7 +268,8 @@ if __name__ == "__main__":
global_conf = auracast_config.global_base_config
bigs = [
auracast_config.broadcast_de
#auracast_config.broadcast_de,
auracast_config.broadcast_en
]
global_conf.octets_per_frame=60# 48kbps@24kHz