abr_sdk.tts#

API for running TTS (text-to-speech) applications.

Author: Pawel Jaworski

Module Contents#

Classes#

Tts

Synthesize speech audio from input text.

Processor

Event loop driving a streaming TTS session.

API#

class abr_sdk.tts.Tts(lib_or_path: str | pathlib.Path | abr_sdk.core.Library, *, lib_search_paths: list[str | pathlib.Path] | None = None, use_default_lib_search_paths: bool = True, resources_dir: str | pathlib.Path | None = None, litert_delegate: str | None = None, cache_dir: str | pathlib.Path | None = None, logger: logging.Logger | None = None)#

Bases: abr_sdk.core.Application

Synthesize speech audio from input text.

Push raw UTF-8 text bytes and receive synthesized mono PCM audio, one signed 16-bit integer per sample in little-endian order at 16 kHz. The backend normalizes pushed text (numbers, dates, currency, acronyms, and abbreviations become speakable words) one sentence at a time, so raw text can be pushed as-is.

Streaming usage::

chunks: list[bytes] = []
with Tts("libabr-tts.so") as tts:
    tts.push(b"Hello world.", on_pcm=chunks.append)
    tts.wait_for_completion()
pcm = b"".join(chunks)

Initialization

text_input_buffer: abr_sdk.core.Buffer#

None

FIFO byte queue that receives the raw UTF-8 text pushed into the model.

pcm_output_buffer: abr_sdk.core.Buffer#

None

FIFO byte queue that emits the synthesized PCM audio as the model produces it.

__enter__() → abr_sdk.tts.Tts#
push(data: bytes, *, on_pcm: collections.abc.Callable[[bytes], None] | None = None, output_poll_timeout_ms: int = 0) → None#

Push UTF-8 text bytes into the TTS pipeline.

Advances the event loop until all input has been accepted by the pipeline. PCM bytes produced while advancing are delivered to on_pcm.

wait_for_completion() → None#

Block until the TTS pipeline finishes synthesizing all queued text.

Flushes the pipeline, then drives the event loop until the flush marker arrives on the PCM output buffer.

close() → None#

Close the TTS instance and release any active processor.

class abr_sdk.tts.Processor(tts: abr_sdk.tts.Tts, on_pcm: collections.abc.Callable[[bytes], None] | None = None)#

Event loop driving a streaming TTS session.

Initialization

process_and_wait_for_output(data: bytes | None, timeout_ms: int, drain_to_idle: bool) → None#

Push data into TTS and advance the event loop until input is drained.

When drain_to_idle is true, flushes the pipeline once all input is pushed and keeps polling until the flush marker arrives on the PCM output buffer.

close() → None#

Release the underlying event set; safe to call multiple times.

__enter__() → abr_sdk.tts.Processor#
__exit__(*exc: Any) → None#