speaker_helper package

Subpackages

Submodules

Module contents

speaker-helper — professional text-to-speech, offline and streaming.

Module summary

speaker-helper is the inverse of vocal-helper: it turns text into speech. A small, typed API over three jobs — say one sentence, stream a reply as it is written, and narrate a document with the timed track that lets a page follow along — plus a CLI, a REST API, an MCP surface and a skill.

The default backend is kokoro: 82M parameters through ONNX, Apache-2.0 in code and weights, 54 voices, running in this process. Nothing to launch, nothing to connect to, and a time-to-first-audio that does not include an HTTP round trip. openvoice and chatterbox clone a voice from a recording; mock is deterministic and serverless, for tests.

Usage example

>>> from speaker_helper import Speaker, Settings
>>> spk = Speaker(Settings.from_mapping({"backend": "mock", "language": "fr"}))
>>> # spk.save("Bonjour le monde.", "hello.wav")

See EXAMPLES.md for a runnable cookbook.

Author

Warith HARCHAOUI — https://linkedin.com/in/warith-harchaoui

class speaker_helper.AudioResult(wav_bytes, sample_rate, duration_s, compute_s, text, voice_id, language)[source]

Bases: object

A synthesised audio payload and its measured metadata.

Parameters:

wav_bytes (bytes) – The complete WAV file contents (RIFF header + PCM), ready to write to disk or stream to a client.
sample_rate (int) – Sample rate of the audio in Hz.
duration_s (float) – Audio duration in seconds.
compute_s (float) – Wall-clock seconds spent producing this audio (network + synthesis).
text (str) – The text that was synthesised.
voice_id (str) – The voice used.
language (str) – The language used.

Notes

rtf (real-time factor) is derived, not stored: compute_s / duration_s. A value below 1.0 means synthesis was faster than real time — the operating point speaker-helper aims for.

compute_s: float

duration_s: float

language: str

property rtf: float

Real-time factor compute_s / duration_s (inf if silent).

Returns:: Below 1.0 means faster than real time.
Return type:: float

sample_rate: int

text: str

voice_id: str

wav_bytes: bytes

class speaker_helper.LanguageProfile(language, engine='kokoro', voice_id='', normalize=True, first_chunk_sentences=1, stream_concurrency=1, measured_rtf=None, measured_quality=None)[source]

Bases: object

Hyperparameters and a measured operating point for one language.

Parameters:

language (str) – ISO-639-1 code this profile targets.
engine (str) – Engine/model id to use for this language.
voice_id (str) – Preset voice id; empty means “auto-pick a voice for the language”.
normalize (bool) – Ask the engine to loudness-normalise the output.
first_chunk_sentences (int) – Streaming producer granularity — sentences in the first emitted chunk. 1 minimises time-to-first-audio.
stream_concurrency (int) – Streaming consumer parallelism — chunks synthesised at once. 1 is a strict pipeline (lowest first-audio latency).
measured_rtf (float or None) – Mean real-time factor measured for this language, or None if untuned.
measured_quality (float or None) – Quality score in [0, 1] measured for this language, or None.

apply(settings=None)[source]

Return settings (or fresh defaults) with this profile applied.

Parameters:: settings (Settings or None) – Base configuration to derive from; None uses defaults.
Returns:: A copy carrying this profile’s language, engine, voice, and producer granularity. stream_concurrency is a Speaker argument, not a Settings field, so pass it separately (see from_profile()).
Return type:: Settings

engine: str = 'kokoro'

first_chunk_sentences: int = 1

language: str

measured_quality: float | None = None

measured_rtf: float | None = None

normalize: bool = True

stream_concurrency: int = 1

voice_id: str = ''

with_measurement(report)[source]

Return a copy with measured_rtf/measured_quality from a report.

Parameters:: report (EvalReport) – A report for this language (duck-typed: needs mean_rtf and quality).
Returns:: A copy carrying the measured operating point.
Return type:: LanguageProfile

class speaker_helper.MockEngine(settings)[source]

Bases: object

A deterministic, serverless TTS backend for tests and evaluation.

It produces a real (decodable) WAV whose duration is proportional to the input length at a fixed speaking rate, and reports a compute time derived from a configured real-time factor. No audio model runs and no network call is made, so behaviour is reproducible and fast — ideal for CI, for exercising the evaluation layer’s anomaly checks, and for proving that engine-agnostic code paths work against any backend.

Parameters:: settings (Settings) – Used for language and engine labels and for the mock knobs (rtf, chars_per_sec, sample_rate, amplitude, frequency_hz) read via Settings.mock.

Notes

The audio is a quiet sine tone (not silence) so evaluation’s “empty audio” and “clipping” checks see a realistic, non-degenerate signal. Set mock.amplitude above 1.0 to deliberately synthesise a clipping signal for testing the anomaly detector.

async aclose()[source]

No-op: the mock holds no resources.

Return type:: None

async clone_voice(name, samples, *, language=None)[source]

Pretend to clone a voice; return a deterministic id (no audio model).

Validates that at least one sample is provided (matching the real engine’s contract) but performs no synthesis-model work, so cloning is testable without a server.

Parameters:

name (str)
samples (list[VoiceSample])
language (str | None)

Return type:

str

async health()[source]

Return a static healthy document identifying the mock backend.

Return type:: dict

async list_voices(engine=None)[source]

Return a small fixed set of fake voices for engine.

Parameters:: engine (str | None)
Return type:: VoiceList

async synthesize(text, *, language=None, voice=None, speed=1.0)[source]

Synthesise text into a deterministic sine-tone WAV.

Honouring speed is what makes this backend worth narrating with: a pace of fast has to shorten the audio, or the cues it produces are not the ones a real engine would produce and the timeline tests pass against a shape nothing ships.

Parameters:

text (str) – Text to “speak” (must be non-empty). Only its length drives the output duration; content is not spoken.
language (str or None) – Override the configured language label on the result.
voice (str or None) – Reported back on the result; nothing here varies with it.
speed (float) – Divides the duration, the way a faster reading would.

Returns:

A decodable WAV plus duration and a compute time equal to duration * rtf (so result.rtf reproduces the configured RTF).

Return type:

AudioResult

Raises:

ValueError – If text is empty or whitespace-only.

class speaker_helper.NarrationResult(wav_bytes, sample_rate, duration_s, compute_s, cues=<factory>, voice_id='', language='', detected_language='', language_fallback=False, ignored_hints=<factory>)[source]

Bases: object

A narrated document: the audio, and where every block landed in it.

Parameters:

wav_bytes (bytes) – The complete WAV file (RIFF header + PCM).
sample_rate (int) – Sample rate in Hz.
duration_s (float) – Length of the audio.
compute_s (float) – Wall-clock seconds spent synthesising.
cues (list of SpeechCue) – Where each segment is spoken. Free to produce — the offsets were already computed to lay the audio out — and the only thing that lets a page follow a narration instead of merely playing it.
voice_id (str) – The voice that read it.
language (str) – The language it was phonemised in. Getting this wrong is not an accent: French read as en-us is an American reading French spelling aloud, letter patterns and all.
detected_language (str) – What the text turned out to be written in, when nobody named a voice. Empty when the caller chose one — this is the detector’s answer, not a restatement of the request.
language_fallback (bool) – True when the text is in a language the backend has no voice for, so it is being read by a voice that does not speak it. Said out loud, because the audio alone will not make it obvious to whoever generated it and does not speak the language either.
ignored_hints (list of str) – Dials the segments asked for and this backend does not have. Empty on a plain narration; non-empty means the caller should know which engine does have them.

compute_s: float

cues: list[SpeechCue]

detected_language: str = ''

duration_s: float

ignored_hints: list[str]

language: str = ''

language_fallback: bool = False

property rtf: float: Real-time factor compute_s / duration_s (inf if silent).

sample_rate: int

to_vtt()[source]

The timed track, as WebVTT.

Return type:: str

voice_id: str = ''

wav_bytes: bytes

class speaker_helper.OperatingPoint(engine, quality, mean_rtf, quality_source='prior', rtf_source='unknown', backend='kokoro', language=None)[source]

Bases: object

One engine’s position on the quality↔speed plane, with provenance.

Parameters:

engine (str) – Engine/model id (e.g. "kokoro").
quality (float) – Quality score in [0, 1] to maximise (measured chrF or a prior).
mean_rtf (float or None) – Mean real-time factor to minimise; None when never measured (the engine then cannot be routed for the online condition).
quality_source (str) – "measured_chrf" or "prior" — how quality was obtained.
rtf_source (str) – "measured", "estimate", or "unknown" — how mean_rtf was obtained.
backend (str) – The backend that serves this engine (default "kokoro").
language (str or None) – Language the measurement pertains to, when language-specific.

backend: str = 'kokoro'

engine: str

language: str | None = None

mean_rtf: float | None

quality: float

quality_source: str = 'prior'

rtf_source: str = 'unknown'

class speaker_helper.RouteDecision(condition, engine, backend, mode, first_chunk_sentences, stream_concurrency, quality, mean_rtf, quality_source, rtf_source, confidence, justification, candidates_considered, pareto_size)[source]

Bases: object

A concrete, justified operating point the caller can act on.

Parameters:

condition (str) – The condition this decision answers.
engine (str) – The chosen engine and the backend that serves it.
backend (str) – The chosen engine and the backend that serves it.
mode (str) – "streaming" (online) or "offline".
first_chunk_sentences (int) – Streaming producer/consumer knobs implied by the condition (only meaningful when mode == "streaming").
stream_concurrency (int) – Streaming producer/consumer knobs implied by the condition (only meaningful when mode == "streaming").
quality (float) – The chosen point’s quality score.
mean_rtf (float or None) – The chosen point’s mean RTF (None only in the offline, RTF-unknown fallback).
quality_source (str) – Provenance of the two axes (see OperatingPoint).
rtf_source (str) – Provenance of the two axes (see OperatingPoint).
confidence (str) – "high" (both axes measured), "medium" (one measured), or "low" (both from priors/estimates).
justification (str) – A human-readable, number-citing explanation of why this point won.
candidates_considered (int) – How many candidates were weighed and how many survived to the Pareto front — for auditability.
pareto_size (int) – How many candidates were weighed and how many survived to the Pareto front — for auditability.

apply(settings=None)[source]

Return settings (or fresh defaults) reconfigured for this decision.

Parameters:: settings (Settings or None) – Base configuration to derive from; None uses defaults.
Returns:: A copy carrying the chosen backend, engine, mode, and (for streaming) the first-chunk granularity. stream_concurrency is a Speaker argument, not a Settings field, so read it off this decision separately (see from_route()).
Return type:: Settings

backend: str

candidates_considered: int

condition: str

confidence: str

engine: str

first_chunk_sentences: int

justification: str

mean_rtf: float | None

mode: str

pareto_size: int

quality: float

quality_source: str

rtf_source: str

stream_concurrency: int

to_dict()[source]

Return a JSON-serialisable dict of the decision (for API / CLI).

Return type:: dict

class speaker_helper.RouteRequest(condition, language='fr', rtf_ceiling=0.8, rtf_budget=None, quality_floor=None, require_measured_rtf=True)[source]

Bases: object

What the caller needs; the router turns it into a justified choice.

Parameters:

condition (str) – "online_realtime" (delay-bounded streaming) or "offline" (throughput / quality). See CONDITIONS.
language (str) – ISO-639-1 target language (drives the default catalogue and profile).
rtf_ceiling (float) – Online only: the hard mean_rtf bound an engine must beat to be feasible. Defaults to DEFAULT_RTF_CEILING (0.8).
rtf_budget (float or None) – Offline only: optional patience cap on mean_rtf (None = no cap, quality is all that matters).
quality_floor (float or None) – Reject any candidate below this quality, in both conditions (None = no floor).
require_measured_rtf (bool) – Online only: refuse engines whose RTF was never measured (default True — do not claim an untimed engine keeps up).

condition: str

language: str = 'fr'

quality_floor: float | None = None

require_measured_rtf: bool = True

rtf_budget: float | None = None

rtf_ceiling: float = 0.8

class speaker_helper.Segment(text, say='', pause_before_ms=0.0, pause_after_ms=0.0, pace='normal', emotion='', intensity='')[source]

Bases: object

One block of a document, as the narrator receives it.

Parameters:

text (str) – The page’s words. This is what the cue carries, and therefore what a read-along highlights. Never substituted.
say (str) – What the engine should actually pronounce, when that differs — a formula read as a sentence, a footnote marker dropped. Empty means “say text”, which is the ordinary case.
pause_before_ms (float) – Silence the document’s structure asks for around this block — the gap that sets a heading apart from the paragraph before it. Requested by the caller, applied here.
pause_after_ms (float) – Silence the document’s structure asks for around this block — the gap that sets a heading apart from the paragraph before it. Requested by the caller, applied here.
pace (str) – "slow", "normal" or "fast". See PACE_SPEED.
emotion (str) – Asked for by some documents, honoured by some engines. A backend that has no such dial reports the name in NarrationResult.ignored_hints rather than pretending.
intensity (str) – Asked for by some documents, honoured by some engines. A backend that has no such dial reports the name in NarrationResult.ignored_hints rather than pretending.

Notes

pause_before_ms is skipped on the first segment: a “pause before” with nothing before it is dead air at the head of the file.

emotion: str = ''

classmethod from_mapping(raw)[source]

Build a segment from a plain dict, ignoring keys it does not know.

The wire format between a document tool and this one is JSON, and the document side is free to carry extra keys (block ids, heading levels) that are none of a narrator’s business. Unknown keys are dropped here rather than rejected, so adding one upstream is not a breaking change.

Parameters:: raw (dict[str, Any])
Return type:: Segment

intensity: str = ''

property is_speakable: bool

Whether anything here could be pronounced at all.

False for a segment that is only punctuation. Such a segment is not dropped — see this module’s docstring — it becomes a silence that keeps its rank.

pace: str = 'normal'

pause_after_ms: float = 0.0

pause_before_ms: float = 0.0

say: str = ''

property spoken: str

say when set, else text.

Type:: What to hand the engine

text: str

class speaker_helper.Settings(backend='kokoro', engine='kokoro', voice='', voice_sample='', voice_id='', language='fr', normalize=True, mode='offline', first_chunk_sentences=1, mock=<factory>, clone=<factory>)[source]

Bases: object

Top-level speaker-helper configuration.

Parameters:

backend (str) – Which TTS backend to drive. "kokoro" (default) runs in this process and needs no server; "openvoice" and "chatterbox" clone a voice from a recording; "mock" is deterministic and serverless, for tests and the evaluation layer. The backend is an implementation detail — the rest of the package talks only to the TTSEngine protocol. See speaker_helper.engine.available_backends().
engine (str) – A label carried on results, for callers that record which model produced what. It does not select the backend; backend does.
voice (str) – Which built-in voice reads. Empty (default) means “read the text, detect the language, and pick the voice that speaks it” — which is not a convenience: the detected language is what the phonemiser works from, and French read as English is not French in an American accent, it is an American spelling the words out.
voice_sample (str) – Path to a reference recording, for the cloning backends. Kokoro refuses one rather than ignoring it.
voice_id (str) – Deprecated spelling of voice, kept because configuration files in the wild use it. voice wins when both are set.
language (str) – ISO-639-1 target language (e.g. "fr").
normalize (bool) – Loudness-normalise the output where the backend can.
mode (str) – "offline" or "streaming".
first_chunk_sentences (int) – Streaming only: sentences in the first emitted chunk (1 → lowest time-to-first-audio, which is the number a conversation feels).
mock (dict[str, Any])
clone (dict[str, Any])

Notes

Precedence, lowest to highest: dataclass defaults → settings.yaml → SPEAKER_HELPER_* environment variables.

backend: str = 'kokoro'

clone: dict[str, Any]

engine: str = 'kokoro'

first_chunk_sentences: int = 1

classmethod from_mapping(mapping)[source]

Build settings from a plain mapping (unknown keys ignored).

Parameters:: mapping (dict or None) – Nested mapping mirroring the dataclass fields.
Returns:: The populated settings, with environment overrides applied.
Return type:: Settings

classmethod from_yaml_text(text)[source]

Build settings from a YAML document string.

Parameters:: text (str) – YAML source.
Returns:: The populated settings.
Return type:: Settings

language: str = 'fr'

classmethod load(path=None)[source]

Load settings from a YAML file if present, else defaults + env.

Parameters:: path (str or Path or None) – Path to a settings.yaml. When None, ./settings.yaml is used if it exists; otherwise dataclass defaults apply. Environment overrides are always applied last.
Returns:: The populated settings.
Return type:: Settings

mock: dict[str, Any]

mode: str = 'offline'

normalize: bool = True

voice: str = ''

voice_id: str = ''

voice_sample: str = ''

class speaker_helper.Speaker(settings=None, *, engine=None, stream_concurrency=1)[source]

Bases: object

Turn text into speech via a pluggable TTS engine, offline or streaming.

The concrete engine (kokoro, chatterbox, mock, …) is selected by settings.backend and reached only through the TTSEngine protocol, so Speaker never depends on a specific backend.

Parameters:

settings (Settings or None) – Configuration; when None, Settings.load() is used (reads ./settings.yaml if present, then SPEAKER_HELPER_* env vars).
engine (TTSEngine or None) – Inject a ready backend instance (e.g. a mock in tests). When None (default), one is built from settings via create_engine().
stream_concurrency (int) – Maximum number of chunks synthesised in parallel in streaming mode. The default of 1 (a strict pipeline) gives the lowest time to first audio: the first chunk gets the whole engine to itself, and because kokoro synthesises faster than real time (RTF < 1) each later chunk is ready before playback of the previous one finishes. Raise it only to favour total throughput over first-audio latency.

Examples

>>> from speaker_helper import Speaker, Settings
>>> spk = Speaker(Settings.from_mapping({"engine": "kokoro", "language": "fr"}))
>>> spk.settings.engine
'kokoro'

async aclose()[source]

Release the underlying engine’s resources.

Return type:: None

property client: TTSEngine

Backward-compatible alias for engine.

Earlier releases exposed the backend as speaker.client (it was always an HTTP client). It is now any TTSEngine; this alias keeps old call sites working.

async clone_voice(name=None, samples=None, *, language=None)[source]

Clone a voice and make this speaker synthesise with it.

Parameters:

name (str or None) – Profile name for the clone (idempotency key). Defaults to the configured clone name, or "ref-fr-female".
samples (list of VoiceSample or None) – Reference recordings. When None, they are resolved from settings.clone — falling back to the bundled ref-fr-female reference. A sample without a transcript is transcribed with vocal-helper.
language (str or None) – Language of the cloned voice (defaults to the configured language).

Returns:

The cloned voice/profile id.

Return type:

str

Examples

>>> from speaker_helper import Speaker, Settings, VoiceSample
>>> spk = Speaker(Settings.from_mapping({"backend": "mock"}))
>>> # give only the audio; the transcript is derived automatically:
>>> # await spk.clone_voice("my-voice", [VoiceSample("ref.wav", "")])

classmethod from_profile(profile, settings=None, *, engine=None)[source]

Build a Speaker from a per-language operating profile.

Applies the profile’s language/engine/voice/producer settings and its stream_concurrency (a consumer-parallelism argument, not a Settings field).

Parameters:

profile (LanguageProfile) – The profile to apply (see speaker_helper.profiles).
settings (Settings or None) – Base configuration to derive from; None uses defaults.
engine (TTSEngine or None) – Optional injected backend (e.g. a mock in tests).

Returns:

A speaker configured for the profile’s language.

Return type:

Speaker

classmethod from_route(condition, *, language='fr', settings=None, engine=None, **route_kwargs)[source]

Build a Speaker from a routed operating point.

Asks speaker_helper.router.route() to choose an engine + mode from measured quality↔speed evidence for the given condition ("online_realtime" or "offline"), applies that choice, and forwards the decision’s stream_concurrency.

Parameters:

condition (str) – "online_realtime" (delay-bounded streaming) or "offline".
language (str) – Target language driving the candidate catalogue.
settings (Settings or None) – Base configuration to reconfigure; None uses defaults.
engine (TTSEngine or None) – Optional injected backend (e.g. a mock in tests); when given, the routing choice still sets language/mode but this engine is used.
**route_kwargs – Extra RouteRequest fields (rtf_ceiling, rtf_budget, quality_floor, require_measured_rtf).

Returns:

A speaker configured for the routed operating point. The chosen RouteDecision is attached as speaker.route for inspection / logging.

Return type:

Speaker

async narrate(segments, *, language=None, voice=None)[source]

Read a document aloud and say where every block landed in the audio.

The difference from stream() is not the engine, it is the bookkeeping: a document has structure — pauses its headings ask for, blocks whose spoken form differs from their printed form — and a caller that wants to follow the reading needs to know which stretch of audio speaks which block. See speaker_helper.narration.

Parameters:

segments (list) – Segment objects, plain dicts, or bare strings — see coerce_segments().
language (str or None) – Override the configured ones for this reading.
voice (str or None) – Override the configured ones for this reading.

Returns:

The audio, the cues, and what the backend could not honour.

Return type:

NarrationResult

Raises:

ValueError – When no segment holds anything speakable, or when the laid-out audio and the timeline disagree.

Notes

Synthesis happens first, for every speakable block at once (bounded by stream_concurrency), and the layout second. Two passes rather than one because the sample rate is the backend’s answer, not ours, and the timeline cannot start counting before it knows what a sample is worth.

save(text, path, *, language=None)[source]

Synthesise text and write the WAV to path (blocking).

Parameters:

text (str) – The text to speak.
path (str or Path) – Destination .wav file.
language (str or None) – Override the configured target language.

Returns:

The synthesised audio (also written to path).

Return type:

AudioResult

async say(text, *, language=None)[source]

Synthesise text into one AudioResult (offline mode).

Parameters:

text (str) – The text to speak.
language (str or None) – Override the configured target language for this call.

Returns:

The synthesised audio and its metadata.

Return type:

AudioResult

say_sync(text, *, language=None)[source]

Blocking wrapper around say() for non-async callers.

Parameters:

text (str) – The text to speak.
language (str or None) – Override the configured target language.

Returns:

The synthesised audio.

Return type:

AudioResult

Raises:

RuntimeError – If called from within a running event loop (use await say).

async stream(text, *, language=None)[source]

Yield audio chunks as they are synthesised (streaming mode).

Parameters:

text (str) – The text to speak, split into sentence-sized chunks.
language (str or None) – Override the configured target language for this call.

Yields:

StreamChunk – Chunks in order. The first carries ttfa_s (time to first audio); the last has is_final=True.

Return type:

AsyncIterator[StreamChunk]

Notes

Synthesis is pipelined up to stream_concurrency chunks at a time so later chunks are produced while earlier ones are being consumed, but emission order matches the text.

async voices(engine=None)[source]

List preset voices for an engine (defaults to the configured one).

Parameters:: engine (str | None)
Return type:: VoiceList

async warmup()[source]

Prime the engine so the first real request is not the slow one.

Synthesises a tiny throwaway utterance, which forces backend-specific one-time costs (profile bootstrap, model download / load) to happen now rather than on a user’s first call. Failures are logged and swallowed — warm-up is best-effort and must never break startup.

Return type:: None

class speaker_helper.SpeechCue(index, text, start_seconds, end_seconds)[source]

Bases: object

One narrated block, with the slice of audio that speaks it.

Parameters:

index (int) – Position in the original segment list, not in the audio. The two differ: a segment whose text is empty after cleaning is never synthesised, so counting emitted pieces drifts from counting segments. Carrying the original index is what lets a page map a cue back to the block it came from.
text (str) – The page’s words — see Segment.
start_seconds (float) – Where this block is spoken, from the head of the file.
end_seconds (float) – Where this block is spoken, from the head of the file.

end_seconds: float

index: int

start_seconds: float

text: str

class speaker_helper.StreamChunk(seq, audio, ttfa_s=None, is_final=False)[source]

Bases: object

One chunk of a streaming synthesis (one sentence-sized unit).

Parameters:

seq (int) – Zero-based index of this chunk within the stream.
audio (AudioResult) – The synthesised audio for this chunk.
ttfa_s (float | None) – Time-to-first-audio in seconds, set only on the first chunk (seq == 0); None afterwards.
is_final (bool) – True for the last chunk of the stream.

audio: AudioResult

is_final: bool = False

seq: int

ttfa_s: float | None = None

class speaker_helper.TTSEngine(*args, **kwargs)[source]

Bases: Protocol

The minimal surface every speaker-helper backend must provide.

A backend is any object exposing these four coroutines. Everything else in the package is written against this protocol, never against a concrete engine, so the backend stays swappable.

Notes

@runtime_checkable lets isinstance(obj, TTSEngine) verify the method names are present (not their signatures) — handy in tests and factories.

async aclose()[source]

Release any held resources (HTTP clients, handles, …).

Return type:: None

async clone_voice(name, samples, *, language=None)[source]

Clone a voice from reference samples and return its id.

Implementations should be idempotent on name (reuse an existing clone rather than duplicating). A backend that cannot clone may raise NotImplementedError.

Parameters:

name (str)
samples (list[VoiceSample])
language (str | None)

Return type:

str

async health()[source]

Return a backend health document (must include a status key).

Return type:: dict

async list_voices(engine=None)[source]

List preset voices for an engine (or the configured one).

Parameters:: engine (str | None)
Return type:: VoiceList

async synthesize(text, *, language=None, voice=None, speed=1.0)[source]

Synthesise text into one AudioResult.

voice and speed arrived with narration, where a document picks a voice once and varies the pace block by block. They are keyword-only and both have defaults, so a backend written against the original two-argument signature still satisfies this protocol — narrate() offers them and withdraws them when the backend does not declare them.

Parameters:

text (str)
language (str | None)
voice (str | None)
speed (float)

Return type:

AudioResult

class speaker_helper.Timeline(sample_rate_hz)[source]

Bases: object

Accumulates durations while a backend accumulates audio.

Two jobs, one object: apply the pauses the segments ask for, and record where every segment landed. Both need the same running sample count, and keeping them apart is how they drift.

Deliberately numpy-free: it counts samples and the caller holds the arrays, so this module imports with nothing but the standard library — which is what lets a caller build and inspect a narration plan without the ONNX runtime anywhere near it.

Examples

>>> timeline = Timeline(1000)
>>> timeline.silence(500)
500
>>> timeline.speech(2000, index=0, text="Hello.")
>>> timeline.silence(250)
250
>>> [(c.start_seconds, c.end_seconds) for c in timeline.cues]
[(0.5, 2.5)]
>>> timeline.total_seconds
2.75

Parameters:: sample_rate_hz (int)

property cues: list[SpeechCue]: Every recorded cue, in playback order.

silence(milliseconds)[source]

Advance the clock by milliseconds; return the samples to append.

Returning the count rather than an array is what keeps this module numpy-free. A caller that ignores the return value desynchronises the timeline from its own audio, which is exactly what check_timeline_matches_audio() exists to catch.

Parameters:: milliseconds (float)
Return type:: int

speech(sample_count, *, index, text)[source]

Record that sample_count samples of audio speak text.

Parameters:

sample_count (int)
index (int)
text (str)

Return type:

None

property total_seconds: float: Length of the audio described so far, silences included.

class speaker_helper.Voice(voice_id, name, language, gender='', engine='')[source]

Bases: object

A preset voice offered by a backend.

Parameters:

voice_id (str) – Engine-specific identifier (e.g. "ff_siwis" for kokoro French).
name (str) – Human-readable display name.
language (str) – ISO-639-1 language code the voice speaks (e.g. "fr", "en").
gender (str) – Reported voice gender, or "" when the engine does not expose one.
engine (str) – Engine the voice belongs to (e.g. "kokoro").

engine: str = ''

gender: str = ''

language: str

name: str

voice_id: str

class speaker_helper.VoiceList(engine, voices=<factory>)[source]

Bases: object

Voices available for an engine.

Parameters:

engine (str) – The engine the voices belong to.
voices (list[Voice]) – The preset voices, possibly empty.

engine: str

voices: list[Voice]

class speaker_helper.VoiceSample(audio, reference_text)[source]

Bases: object

One reference recording used to clone a voice.

Parameters:

audio (bytes or str) – The reference audio: raw bytes, or a filesystem path to an audio file.
reference_text (str) – A transcript of exactly what is said in audio. Cloning engines align the audio to this text, so it must match the recording.

Notes

Keep samples short and clean (a few seconds of clear speech). Provide several for a more robust clone.

audio: bytes | str

filename()[source]

Return a filename to send with the upload (from the path, or a default).

Return type:: str

read_bytes()[source]

Return the audio as bytes, reading from disk if a path was given.

Return type:: bytes

reference_text: str

speaker_helper.available_backends()[source]

Return the sorted names of all registered backends.

Return type:: list[str]

speaker_helper.chunk_for_streaming(text, first_chunk_sentences=1)[source]

Group sentences into synthesis chunks, keeping the first one small.

Parameters:

text (str) – The text to synthesise.
first_chunk_sentences (int, optional) – How many sentences to place in the first chunk. Keeping this at 1 (the default) minimises TTFA; larger values reduce per-chunk overhead at the cost of a slower first emission.

Returns:

Chunks of text to synthesise in order. The first chunk holds first_chunk_sentences sentences; every remaining sentence is its own chunk so downstream audio keeps flowing steadily.

Return type:

list[str]

Examples

>>> chunk_for_streaming("A. B. C.", first_chunk_sentences=1)
['A.', 'B.', 'C.']
>>> chunk_for_streaming("A. B. C.", first_chunk_sentences=2)
['A. B.', 'C.']

speaker_helper.coerce_segments(raw)[source]

Accept segments as dicts, as Segment, or as bare strings.

Three shapes because three callers: a document tool writes JSON, a Python caller builds dataclasses, and somebody at a REPL has a list of sentences and no interest in any of this.

Parameters:: raw (list[Any])
Return type:: list[Segment]

speaker_helper.complexity(latex)[source]

How much structure a formula carries.

A rough count of the things that make a formula hard to follow by ear: commands, relations, nesting, sub- and superscripts. R scores zero and needs no help; a nested fraction of sums scores high and a symbol-by-symbol reading of it is sound without meaning.

Parameters:: latex (str)
Return type:: int

speaker_helper.create_engine(settings)[source]

Instantiate the backend named by settings.backend.

Parameters:: settings (Settings) – Configuration; its backend field selects the engine.
Returns:: A ready-to-use engine instance.
Return type:: TTSEngine
Raises:: ValueError – If settings.backend names no registered backend.

speaker_helper.cues_to_vtt(cues)[source]

Render narration cues as WebVTT.

Each cue carries its segment index as the WebVTT identifier — the line before the timings — which is what lets a page highlight the right block without matching on text. The text is collapsed onto one line because a blank line ends a cue, so a hard-wrapped payload would silently become two cues and shift every index after it.

Parameters:: cues (list[SpeechCue])
Return type:: str

speaker_helper.default_operating_points(language='fr')[source]

Build the fallback candidate set for language from profiles + priors.

The per-language profile supplies the measured kokoro operating point (RTF from native-MLX tuning; quality is the engine prior until a round-trip recalibrates it). Every other engine in the priors table contributes a quality-only candidate with an unknown RTF — enough to reason about offline quality, but deliberately not enough to claim it keeps up online.

Parameters:: language (str) – Language whose profile seeds the measured kokoro point.
Returns:: One candidate per known engine, provenance flagged on both axes.
Return type:: list of OperatingPoint

speaker_helper.profile_for(language)[source]

Return the profile for language (a default profile if none is defined).

Parameters:: language (str)
Return type:: LanguageProfile

speaker_helper.register_backend(name, factory)[source]

Register (or override) a backend factory under name.

Parameters:

name (str) – Backend identifier used in settings.backend (case-insensitive).
factory (callable) – Settings -> TTSEngine. Called lazily by create_engine().

Return type:

None

speaker_helper.resolve_spoken_math(segments, lang, *, narrator=None)[source]

Fill each segment’s say with a readable form of its notation.

A convenience over speaker_helper.speech_text.speakable(), which is where the work actually happens. Separate from Segment on purpose: a segment whose say was set deliberately — a pronunciation override, a formula a human already phrased — must not be overwritten, so this only fills the empty ones.

Parameters:

segments (list of Segment) – Left untouched; a new list is returned.
lang (str) – The language the page is written in. Not optional and not defaulted: a French page read with English operator names is worse than one that says nothing, because the listener hears fluent words in the wrong language and assumes they misheard.
narrator (callable or None) – (latex, lang) -> str | None, asked about formulas too complex to spell out. Optional — without one every formula gets the deterministic reading, which is the baseline, not a degraded mode.

Returns:

The same blocks, with say filled where it was empty and the notation differs from the words.

Return type:

list of Segment

speaker_helper.route(request, *, points=None, reports=None)[source]

Choose a justified engine + mode for request from the evidence.

Parameters:

request (RouteRequest) – The condition and its constraints.
points (sequence of OperatingPoint or None) – Explicit candidate operating points. When None and reports is also None, the built-in catalogue for request.language is used.
reports (sequence of EvalReport or None) – Measured reports to derive candidates from (highest confidence). Takes precedence over points when both are given.

Returns:

The chosen operating point, mode, streaming knobs, and a number-citing justification.

Return type:

RouteDecision

Raises:

ValueError – If request.condition is unknown, or no candidate satisfies the condition’s constraints (e.g. no engine keeps up online).

Examples

>>> route(RouteRequest(condition="offline", language="fr")).mode
'offline'

speaker_helper.route_settings(condition, *, language='fr', settings=None, **request_kwargs)[source]

Convenience: route for condition and return applied settings + decision.

Parameters:

condition (str) – "online_realtime" or "offline".
language (str) – Target language.
settings (Settings or None) – Base settings to reconfigure; None uses defaults.
**request_kwargs – Extra RouteRequest fields (rtf_ceiling, rtf_budget, quality_floor, require_measured_rtf).

Returns:

Settings reconfigured for the decision, and the decision itself (its stream_concurrency still needs forwarding to Speaker).

Return type:

tuple of (Settings, RouteDecision)

speaker_helper.speak_math(latex, lang='en')[source]

One formula, as a person would read it out.

R = T_{BA} \circ T_{AB} becomes “R equals T B A composed with T A B”.

Subscripts are read as their own letters rather than announced: “T A B” is how somebody says that at a blackboard, and “T subscript A B” is how nobody does. Powers keep their announcement, because “x 2” and “x squared” are different claims.

Parameters:

latex (str)
lang (str)

Return type:

str

speaker_helper.speakable(text, lang='en', *, narrator=None)[source]

The sentence as the voice should say it.

Only the notation changes; every other word is the page’s.

narrator is an optional (latex, lang) -> str | None. It is asked about formulas whose complexity() is above NARRATE_ABOVE, and its answer is used when it gives one. This is the difference between spelling a formula and reading it: “the sum over i of x i, all over n” is a transliteration, “the average of the x values” is a sentence, and only the second survives being heard once with no page in front of you.

It stays an injected callable rather than an import so this module needs no model, no network and no optional extra to be used or tested. Without one, every formula gets the deterministic reading — which is the baseline, not a degraded mode.

Parameters:

text (str)
lang (str)

Return type:

str

speaker_helper.split_sentences(text)[source]

Split text into sentence-sized, non-empty, stripped units.

Parameters:: text (str) – Arbitrary text, possibly multi-sentence and multi-line.
Returns:: Sentences in order, each stripped of surrounding whitespace, with their terminating punctuation preserved. Text without any terminator returns a single-element list (the whole stripped input); empty or whitespace-only input returns an empty list.
Return type:: list[str]

Examples

>>> split_sentences("Un. Deux. Trois.")
['Un.', 'Deux.', 'Trois.']
>>> split_sentences("no terminator here")
['no terminator here']
>>> split_sentences("   ")
[]

async speaker_helper.tune_profiles(base_settings, languages, **eval_kwargs)[source]

Measure a set of languages and return profiles carrying the results.

A thin convenience over run_multilang_eval(): it runs the evaluation per language and folds each report’s RTF/quality into the corresponding LanguageProfile.

Parameters:

base_settings (Settings) – Base configuration (backend, engine, connection).
languages (list of str) – Language codes to measure and profile.
**eval_kwargs – Forwarded to run_multilang_eval() (e.g. thresholds, transcriber, warmup).

Returns:

Mapping language -> LanguageProfile with measured RTF/quality.

Return type:

dict

speaker_helper.voice_for(language, gender='', *, index=0)[source]

Resolve a (language, gender) pair to a concrete kokoro voice id.

Parameters:

language (str) – ISO-639-1 language code.
gender (str, optional) – "male" or "female"; empty picks the first voice of the language.
index (int, optional) – Which matching voice to take when several exist, cycled modulo the count (so callers can hand out distinct voices to distinct speakers).

Returns:

A real kokoro voice_id. Falls back to the first voice of the language (ignoring gender), then to ff_siwis/am_michael-style defaults, so the result is always a usable id.

Return type:

str

speaker_helper.voices_for(language, gender='')[source]

Return the curated voices matching language and, if given, gender.

Parameters:

language (str) – ISO-639-1 language code (e.g. "fr", "en").
gender (str, optional) – "male" or "female" to filter; empty returns both genders.

Returns:

Matching voices, in catalogue order (may be empty).

Return type:

list of Voice