record_stream

Signature:

BoosterAudioManager.record_stream(
    sample_rate: int = 16000,
    channels: int = 1,
    sample_format: PcmSampleFormat = "S16LE",
    *,
    use_naec: bool = False,
    chunk_bytes: int | None = None,
    stop_event: threading.Event | None = None,
) -> Iterator[AudioData]

Description:

Reads microphone input as a generator, continuously yielding AudioData audio chunks.

Parameters:

Parameter	Type	Default	Description
sample_rate	int	16000	Sample rate in Hz.
channels	int	1	Number of channels.
sample_format	PcmSampleFormat	"S16LE"	PCM sample format, see "Perception and State Data / PcmSampleFormat".
use_naec	bool	False	Whether to prioritize NAEC audio.
chunk_bytes	int | None	None	When not None, re-chunks audio by the specified byte count.
stop_event	threading.Event | None	None	External stop signal.

Returns:

Iterator[AudioData], see "Perception and State Data / AudioData"

Raises:

ValueError: Invalid sample format, sample rate, channels, or chunk_bytes.

RuntimeError: Audio subsystem is closed, or an error occurred during capture.

Example:

import threading
from boosteros.robots.booster import BoosterRobot

robot = BoosterRobot()
stop_event = threading.Event()
for index, chunk in enumerate(
    robot.audio_manager.record_stream(chunk_bytes=4096, stop_event=stop_event)
):
    print(chunk.duration.seconds)
    # Example output: 0.128
    if index >= 2:
        stop_event.set()