play_stream

Signature:

BoosterAudioManager.play_stream(
    *,
    sample_rate: int = 16000,
    channels: int = 1,
    sample_format: PcmSampleFormat = "S16LE",
    volume: float | None = None,
    queue_size: int = 64,
    overflow: OverflowPolicy = "block",
) -> AudioPlaybackStreamHandle

Description:

Opens a writable PCM playback stream, suitable for dynamic data sources such as TTS and WebSocket audio.

Parameters:

Parameter	Type	Default	Description
sample_rate	int	16000	Sample rate in Hz.
channels	int	1	Number of audio channels.
sample_format	PcmSampleFormat	"S16LE"	PCM sample format; see "Sensor and State Snapshot / PcmSampleFormat".
volume	float | None	None	Playback volume in the range 0.0 to 1.0.
queue_size	int	64	Write queue size; must be greater than 0.
overflow	OverflowPolicy	"block"	Policy when the queue is full; see "Tasks and Subscriptions / OverflowPolicy".

Returns:

AudioPlaybackStreamHandle; see "Tasks and Subscriptions / AudioPlaybackStreamHandle"

Raises:

ValueError: Invalid sample format, sample rate, channel count, volume, queue size, or overflow policy.

RuntimeError: An audio task in the same group is already running.

Example:

import time
from boosteros.robots.booster import BoosterRobot

robot = BoosterRobot()

robot.audio_manager.start_recording()
time.sleep(5.0)
audio = robot.audio_manager.stop_recording()

stream = robot.audio_manager.play_stream(
    sample_rate=audio.sample_rate,
    channels=audio.channels,
    sample_format=audio.sample_format,
)
try:
    chunk_bytes = 4096
    for start in range(0, len(audio.data), chunk_bytes):
        stream.write(audio.with_data(audio.data[start : start + chunk_bytes]))
finally:
    stream.close()

status = stream.wait(timeout=audio.duration.seconds + 10.0)
print(status)
# Example output: 'SUCCEEDED'