ReferenceFull Library ReferenceNemo Voice AgentNemo Voice AgentUtils

nemo_voice_agent.utils.audio

View as Markdown

Module Contents

Classes

NameDescription
AudioStreamA class that simulates a realtime audio stream. It caches the input audio chunks
NoiseConfigA class that configures the noise for the audio stream.
NoiseGeneratorA class that generates noise audio by reading provided noise audio files.
SOXRAudioResamplerAn audio resampler that uses the SoX resampler library. It’s stateless and will return the result immediately.
SOXRAudioStreamResamplerA class that resamples an audio stream using the SoX resampler library.

Functions

NameDescription
audio_bytes_to_float32Convert PCM-16 audio bytes to float32 numpy array, clamped to -1.0 to 1.0.
audio_float32_to_bytesConvert float32 numpy array to PCM-16 audio bytes.

Data

STREAM_TIMEOUT_SECS

API

class nemo_voice_agent.utils.audio.AudioStream(
    chunk_size_in_seconds: float,
    input_sample_rate: int,
    output_sample_rate: int,
    stream_resampler: bool = True,
    tag: str = '',
    min_buffer_chunks: int = 5,
    drain_threshold: int = 5,
    min_sustain_chunks: int = 1,
    noise_config: typing.Optional[nemo_voice_agent.utils.audio.NoiseConfig] = None
)

A class that simulates a realtime audio stream. It caches the input audio chunks and resamples them to the output sample rate. Each time its get() function is called, it returns the next chunk of audio at the output sample rate. If the audio cache doesn’t have enough audio to fill the output chunk, it will append silence to the output chunk.

The class will be used in an asyncio context, where one thread is putting audio chunks into the cache and another thread is getting audio chunks from the cache.

_buffer_empty_count
= 0
_next_send_time
= 0
_prev_noise_scale
= 1.0
audio_cache
= asyncio.Queue()
current_buffer_size
int

Get the current size of the buffer.

gain_db
= self.noise_config.gain_db
noise_generator
output_buffer
= b''
output_chunk_bytes
resampler
nemo_voice_agent.utils.audio.AudioStream._augment_with_noise(
    audio_chunk: bytes,
    noise_chunk: typing.Optional[bytes] = None
) -> bytes

Augment audio with noise based on random SNR sampling.

This method mixes audio with noise according to a gain_db.

Parameters:

audio_chunk
bytes

Original audio bytes (16-bit signed integers)

noise_chunk
Optional[bytes]" default="None

Noise audio bytes (16-bit signed integers)

Returns: bytes

Mixed audio with noise as bytes

nemo_voice_agent.utils.audio.AudioStream._is_buffer_full() -> bool

Check if the buffer is full.

nemo_voice_agent.utils.audio.AudioStream._send_audio_sleep()
async

Simulate audio device timing by sleeping between audio chunks.

nemo_voice_agent.utils.audio.AudioStream.get_nowait() -> typing.Tuple[bytes, bool]
async

Get the next output chunk of audio, immediately padding with silence if no audio is available.

nemo_voice_agent.utils.audio.AudioStream.get_output_chunk(
    audio_chunk: bytes,
    noise_chunk: typing.Optional[bytes] = None
) -> bytes

Pad audio chunk with silence/noise if shorter than expected output chunk size.

If noise_chunk is provided and noise_generator is available, the audio will be augmented with noise based on SNR. Otherwise, it pads with silence (zeros).

Parameters:

audio_chunk
bytes

Audio bytes (16-bit signed integers)

noise_chunk
Optional[bytes]" default="None

Optional noise bytes for augmentation

Returns: bytes

Audio chunk padded to output_chunk_bytes

nemo_voice_agent.utils.audio.AudioStream.get_wait(
    timeout: float = None,
    no_wait: bool = False
) -> typing.Tuple[bytes, bool]
async

Get the next output chunk of audio, WAITING for audio to be available.

Unlike get(), this method will block and wait for audio to arrive rather than immediately padding with silence. This prevents gaps in audio when packets arrive in bursts (common in WebSocket/network scenarios).

Use this for continuous audio streaming where you want smooth audio without artificial gaps.

Returns: Tuple[audio_chunk, has_speech]: Tuple containing the audio chunk bytes and a boolean indicating if there’s speech in the chunk

Parameters:

timeout
float" default="None

Maximum time to wait in seconds (None = no wait)

no_wait
bool" default="False

If True, only tries to read the audio cache once, and returns silence immediately if no audio is available.

nemo_voice_agent.utils.audio.AudioStream.put(
    audio_chunk: bytes
)
async

Put an audio chunk into the audio cache after resampling.

Parameters:

audio_chunk
bytes

Input audio chunk at input_sample_rate

nemo_voice_agent.utils.audio.AudioStream.resample(
    audio_chunk: bytes
) -> bytes

Resample an audio chunk from input sample rate to output sample rate.

Parameters:

audio_chunk
bytes

Raw audio bytes (16-bit signed integers)

Returns: bytes

Resampled audio bytes (16-bit signed integers)

class nemo_voice_agent.utils.audio.NoiseConfig(
    noise_files: typing.Optional[typing.Union[typing.List[str], str]] = None,
    gain_db: float = 0.0,
    max_noise_duration: typing.Optional[float] = 600.0,
    random_offset: bool = True,
    random_white_noise: bool = False,
    white_noise_db: typing.Optional[float] = -90.0
)
Dataclass

A class that configures the noise for the audio stream.

gain_db
float = 0.0
max_noise_duration
Optional[float] = 600.0
noise_files
Optional[Union[List[str], str]] = None
random_offset
bool = True
random_white_noise
bool = False
white_noise_db
Optional[float] = -90.0
nemo_voice_agent.utils.audio.NoiseConfig.to_dict() -> dict

Convert the noise configuration to a dictionary.

class nemo_voice_agent.utils.audio.NoiseGenerator(
    noise_audio_files: typing.Union[typing.List[str], str],
    sample_rate: int,
    max_duration: typing.Optional[float] = None,
    random_offset: bool = True,
    random_white_noise: bool = False,
    white_noise_db: typing.Optional[float] = None
)

A class that generates noise audio by reading provided noise audio files.

current_position
= 0
noise_audio_data
= self.load_audio_files()
nemo_voice_agent.utils.audio.NoiseGenerator.generate_random_white_noise() -> numpy.ndarray

Generate random white noise with the given duration and sample rate.

Returns: np.ndarray

np.ndarray: Float32 white noise in [-1.0, 1.0], length max_duration * sample_rate. Scaled by white_noise_db when set (amplitude = 10^(white_noise_db/20)).

nemo_voice_agent.utils.audio.NoiseGenerator.get_noise_chunk(
    chunk_size_in_seconds: float
) -> numpy.ndarray

Get the next noise audio segment of chunk size chunk_size_in_seconds, and return the chunk. If the noise audio data is less than the chunk size, restart from the beginning.

Parameters:

chunk_size_in_seconds
float

Duration of the noise chunk to return in seconds.

Returns: np.ndarray

np.ndarray: Noise audio chunk of the requested duration.

nemo_voice_agent.utils.audio.NoiseGenerator.get_noise_chunk_bytes(
    chunk_size_in_seconds: float
) -> bytes

Get the next noise audio segment of chunk size chunk_size_in_seconds, and return the chunk as Int16 bytes.

nemo_voice_agent.utils.audio.NoiseGenerator.load_audio_files() -> numpy.ndarray

Load the noise audio files.

class nemo_voice_agent.utils.audio.SOXRAudioResampler(
    in_sample_rate: int,
    out_sample_rate: int,
    quality: str = 'VHQ',
    args = (),
    kwargs = {}
)

An audio resampler that uses the SoX resampler library. It’s stateless and will return the result immediately.

nemo_voice_agent.utils.audio.SOXRAudioResampler.resample(
    audio: bytes
) -> bytes

Resample audio data using SoX resampler library.

Parameters:

audio
bytes

Input audio data as raw bytes (16-bit signed integers).

Returns: bytes

Resampled audio data as raw bytes (16-bit signed integers).

class nemo_voice_agent.utils.audio.SOXRAudioStreamResampler(
    in_sample_rate: int,
    out_sample_rate: int,
    quality: str = 'VHQ',
    args = (),
    kwargs = {}
)

A class that resamples an audio stream using the SoX resampler library.

resampler
nemo_voice_agent.utils.audio.SOXRAudioStreamResampler._should_flush()

Check if the resampler should be flushed.

nemo_voice_agent.utils.audio.SOXRAudioStreamResampler.resample(
    audio: bytes
)

Resample an audio chunk using the SoX resampler library. Args: audio: The audio chunk to resample. Returns: The resampled audio chunk.

nemo_voice_agent.utils.audio.SOXRAudioStreamResampler.reset()

Reset the resampler.

nemo_voice_agent.utils.audio.audio_bytes_to_float32(
    audio_bytes: bytes
) -> numpy.ndarray

Convert PCM-16 audio bytes to float32 numpy array, clamped to -1.0 to 1.0.

nemo_voice_agent.utils.audio.audio_float32_to_bytes(
    audio_float32: numpy.ndarray
) -> bytes

Convert float32 numpy array to PCM-16 audio bytes.

nemo_voice_agent.utils.audio.STREAM_TIMEOUT_SECS = 0.2