vocal_helper.asr_cloud module

vocal_helper.asr_cloud

Hosted transcription: Gladia (pre-recorded and live) and ElevenLabs Scribe (pre-recorded). The paid half of the transcription stage, mirroring vocal_helper.asr on the other side of the network.

Extracted from notes_helper.cloud_transcribe and notes_helper.live_cloud.asr_cloud (same author), where this code was written, measured and debugged against the real APIs. It belongs in the library that owns transcription rather than in a product sitting on top of one.

Opt-in, never a fallback

A missing key raises. There is no path where an absent credential degrades to a local engine, and no path where a present credential turns a free run into a billed one: the caller asks for a cloud mode by name or gets the local one. Every billed call goes through vocal_helper.wallet.paid(), so the same audio is never charged twice.

Two models, two endpoints, and the mistake that is easy to make

Re-verified against the provider’s OpenAPI specification on 2026-11-01, because the answer is the kind that changes and the cost of being wrong is a rejected paid job. Gladia exposes transcription twice, and the model names are not interchangeable.

  • Pre-recorded (POST /v2/upload then /v2/pre-recorded) accepts solaria-3, the newer model, optimised for English, French, German, Spanish and Italian, and it does diarize.

  • Live (POST /v2/live then a WebSocket) accepts solaria-1 and nothing else: the specification’s streaming-model enumeration has a single member. An earlier version of this code set the live model to solaria-3 on the reasonable-sounding assumption that newer is better; the API rejects it. There is no solaria-2, and solaria-3 remains asynchronous only, so “take the best available” and “take the only one available” give the same answer here.

And the premise of the whole live design: Gladia does not diarize live (confirmed against Gladia’s own documentation — real-time diarization is not supported). Speaker identity for the live paid mode must come from pyannoteAI Live-1 in vocal_helper.diar_cloud and be fused in vocal_helper.fusion, never from Gladia alone.

Neither live endpoint has a Python software development kit; both are wire-protocol only, which is why the clients below are hand-written.

Why a glossary is not part of the ledger key

Both providers accept a list of domain terms to bias recognition toward: Gladia calls it a custom vocabulary, ElevenLabs calls it keyterm prompting. Neither joins the ledger key. Editing a glossary and re-running the same audio reuses the prior transcript rather than paying again: a misspelt proper noun is cheaper to repair downstream than to re-transcribe, and when a quota is spent, a cache miss does not merely cost money — it silently fails over and changes the result.

Bias strength matters too. At Gladia’s default intensity the model force-fits boosted terms, “hearing” a glossary proper noun where none was spoken. The default here is lower, which keeps the spelling help and cuts the hallucinated insertions.

Usage example

>>> parse_gladia_message({"type": "final", "utterances": [
...     {"words": [{"word": "bonjour", "time_begin": 0.0,
...                 "time_end": 0.6, "confidence": 0.98}]}]})
AsrTranscript(kind='final', words=[AsrWord(word='bonjour', time_begin=0.0, time_end=0.6, confidence=0.98)])
>>> parse_gladia_message({"type": "session_start"}) is None
True
>>> sanitize_keyterms(["Arcom", "a (very) long & messy term", "x"])
['Arcom', 'a long et messy term']
class vocal_helper.asr_cloud.AsrTranscript(kind, words=<factory>)[source]

Bases: object

One parsed live transcript event, partial or final.

Parameters:
kind: str

"partial" (for display) or "final" (for the record).

words: list[AsrWord]
class vocal_helper.asr_cloud.AsrWord(word, time_begin, time_end, confidence=1.0)[source]

Bases: object

One transcribed word with its span and the model’s confidence in it.

Parameters:
confidence: float = 1.0
time_begin: float
time_end: float
word: str
vocal_helper.asr_cloud.GLADIA_LIVE_MODEL = 'solaria-1'

The ONLY model the live endpoint accepts (verified 2026-11-01). A constraint, not a choice.

vocal_helper.asr_cloud.GLADIA_MAX_S = 8000.0

Just under Gladia’s ~8100 s (~135 min) per-file cap; longer media is chunked and stitched.

vocal_helper.asr_cloud.GLADIA_PRERECORDED_MODEL = 'solaria-3'

Pre-recorded default. Override with VOH_GLADIA_MODEL (solaria-1 for 100+ languages).

class vocal_helper.asr_cloud.GladiaLiveASR(ws)[source]

Bases: object

Stream PCM over an open WebSocket and yield parsed transcripts.

The connection is injected, never opened here: production passes a real websockets connection to the URL from open_gladia_live_session(), and a test points the same class at a local mock server. So this class touches neither the network, nor a credential, nor the optional dependency — it only uses the send / async-iteration interface any such connection provides.

Parameters:

ws (Any) – An open connection exposing await ws.send(bytes) and async iteration over received frames.

Examples

>>> import asyncio
>>> class FakeWs:
...     def __init__(self): self.sent = []
...     async def send(self, payload): self.sent.append(payload)
...     def __aiter__(self):
...         async def gen():
...             yield json.dumps({"type": "final", "words": [
...                 {"word": "salut", "time_begin": 0.0, "time_end": 0.4}]})
...         return gen()
>>> async def demo():
...     ws = FakeWs()
...     asr = GladiaLiveASR(ws)
...     await asr.send_audio(b"\x00\x00")
...     return [t.words[0].word async for t in asr.transcripts()], ws.sent
>>> asyncio.run(demo())
(['salut'], [b'\x00\x00'])
async send_audio(pcm)[source]

Send one raw PCM chunk, in the encoding declared at session creation.

Parameters:

pcm (bytes)

Return type:

None

async transcripts()[source]

Yield each transcript event as the server emits it, until the socket closes.

Return type:

AsyncIterator[AsrTranscript]

vocal_helper.asr_cloud.PCM_SAMPLE_RATE = 16000

Hz. Gladia also accepts 8000/32000/44100/48000, declared at session init.

vocal_helper.asr_cloud.elevenlabs_transcribe(*, audio, language=None, keyterms=None, diarize=True, num_speakers=None, model=None)[source]

Transcribe one file with ElevenLabs Scribe, word by word. Billed once per audio.

Scribe bundles diarization into this same response, unlike Gladia, which is transcription-only and always pairs with a separate diarization call. So diarize and num_speakers genuinely change what comes back, and both are part of the ledger key: a run with a forced speaker count must not silently reuse an auto-detected result.

Scribe handles long files in one call, so no chunking is needed here.

Parameters:

  • audio (str, keyword-only) – Media path. Converted to 16 kHz mono MP3 before upload.

  • language (str, optional) – ISO-639 code. None auto-detects.

  • keyterms (list of str, optional) – Domain terms to bias toward. Sanitised by sanitize_keyterms(); not part of the ledger key.

  • diarize (bool, optional) – Ask Scribe for a per-word speaker. A caller may instead relabel with pyannoteAI downstream.

  • num_speakers (int, optional) – Force an exact count. Part of the ledger key.

  • model (str, optional) – Defaults to ELEVENLABS_STT_MODEL or scribe_v1.

Returns:

Scribe’s response: a flat word stream with timings and speakers.

Return type:

dict

Raises:

RuntimeError – When the key is missing, or the request is refused.

vocal_helper.asr_cloud.gladia_transcribe(*, audio, languages=None, code_switching=True, custom_vocabulary=None)[source]

Transcribe one file with Gladia’s pre-recorded endpoint. Billed once per audio.

Parameters:

  • audio (str, keyword-only) – Media path. Converted to 16 kHz mono MP3 before upload.

  • languages (list of str, optional) – Expected languages. None lets Gladia detect.

  • code_switching (bool, optional) – Whether a single recording may switch between the declared languages.

  • custom_vocabulary (list of str, optional) – Domain terms and proper nouns to bias toward. Not part of the ledger key — see the module docstring.

Returns:

Gladia’s full result document.

Return type:

dict

Raises:

RuntimeError – When the key is missing, or when the upload, the job submission or the job itself fails. Each case says which, with the provider’s own message: a rejected oversized vocabulary and an exhausted quota are different problems and must not read alike.

vocal_helper.asr_cloud.gladia_utterances(audio, *, languages=None, code_switching=True, custom_vocabulary=None, overlap_s=120.0)[source]

Gladia utterances for a recording of any length.

A file under the cap is one call. A longer one is split into overlapping chunks, each transcribed separately, then stitched: chunk times are shifted to absolute, each chunk’s speaker labels are matched to the running result through the shared window, and only the turns past the overlap’s midpoint are appended, so the seam neither loses nor duplicates anything.

Parameters:
Returns:

Gladia utterances in absolute time, with globally consistent speakers.

Return type:

list of dict

vocal_helper.asr_cloud.open_gladia_live_session(*, sample_rate=16000)[source]

Open a Gladia live session and return its identifiers and socket URL.

Parameters:

sample_rate (int, keyword-only, optional) – The rate the stream will actually carry. Declared here, honoured by the socket; mismatching it corrupts the audio silently.

Returns:

{"id", "created_at", "url"}. The url carries a temporary token.

Return type:

dict of str

Raises:

RuntimeError – When GLADIA_API_KEY is not configured. The mode is unavailable rather than degraded.

vocal_helper.asr_cloud.parse_gladia_message(raw)[source]

Parse one decoded live message, or None when it is not a transcript.

Pure: no input/output, so the wire format is testable without a socket.

Parameters:

raw (dict) – One decoded JSON message from the live WebSocket.

Returns:

None for anything that is not a transcript event (session lifecycle, acknowledgements), so a caller can ignore the rest without a type check.

Return type:

AsrTranscript or None

Examples

>>> parse_gladia_message({"type": "partial", "words": [
...     {"word": "oui", "time_begin": 1.0, "time_end": 1.2}]}).kind
'partial'
>>> parse_gladia_message({"type": "final", "utterances": []}).words
[]
vocal_helper.asr_cloud.pcm_f32_to_wire_bytes(pcm)[source]

Mono float32 samples in [-1, 1] to Gladia live’s 16-bit PCM wire format.

Clamps before scaling, so a source that slightly overshoots [-1, 1] — a hot input, a resampling overshoot — cannot wrap around into a full-amplitude 16-bit value, which is heard as a click and transcribed as nothing.

Parameters:

pcm (numpy.ndarray) – Mono float32 samples.

Returns:

Little-endian signed 16-bit samples.

Return type:

bytes

Examples

>>> import numpy as np
>>> pcm_f32_to_wire_bytes(np.array([0.0, 1.0, -2.0], dtype=np.float32))
b'\x00\x00\xff\x7f\x01\x80'
vocal_helper.asr_cloud.sanitize_keyterms(terms, *, cap=1000, max_len=50, max_spaces=4)[source]

Make a keyterm list safe, so one bad term never rejects the whole call.

Scribe validates keyterms strictly and rejects the entire request when a single term breaks a rule — which, in a product with a local fallback, shows up as a mysteriously free run with a worse transcript. A glossary is data nobody controls (hand-written, model-extracted), so it is cleaned at this boundary rather than trusted: curly quotes normalised, parentheticals dropped, & spelled out, unsupported characters stripped, spaces collapsed, length truncated, and anything still too long or too wordy discarded.

Parameters:

  • terms (list of str or None) – The raw glossary.

  • cap (int, keyword-only, optional) – Maximum number of terms to keep.

  • max_len (int, keyword-only, optional) – Maximum characters per term.

  • max_spaces (int, keyword-only, optional) – Maximum spaces per term, i.e. max_spaces + 1 words.

Returns:

Cleaned, de-duplicated case-insensitively, original order preserved.

Return type:

list of str

Examples

>>> sanitize_keyterms(None)
[]
>>> sanitize_keyterms(["Arcom", "ARCOM", "a"])
['Arcom']
>>> sanitize_keyterms(["L’Arcom (le régulateur)"])
["L'Arcom"]