vocal_helper.cloud module

vocal_helper.cloud

The two paid entry points, siblings of the two free ones.

vocal-helper has four modes on two independent axes. Local or cloud asks where the audio goes; offline or live asks when the answer arrives. Confusing the two is the classic mistake, so they are named separately and none of them is reachable by accident:

Four names rather than one factory taking a string, deliberately: a billed mode must not be reachable through a typo, and the two local names already have live callers that must keep working untouched.

There is no automatic failover between the columns. A missing key makes a paid mode raise, never degrade: a run that silently fell back would produce a different answer at a different quality under the same command, which is the worst possible outcome for anything meant to be reproducible.

The hybrid is not a class

Measured on this suite’s material, the free path does not lose on the words, it collapses on the speakers. Diarization is also the cheaper hosted stage. So the purchase worth making is local transcription with hosted diarization, and on this corpus it changes the bill by roughly a factor of six.

That mode needs no new orchestration, because OfflinePipeline already composes source → diarization → transcription. It is one extra diarization backend:

voh.OfflinePipeline(source=..., config=voh.OfflinePipelineConfig(
    diar={"backend": "pyannoteai"},   # hosted speakers
    asr={"language": "fr"},           # local words
))

Expressing it that way rather than as a third class is what keeps the matrix honest: the hybrid is genuinely the local offline pipeline with one stage bought, not a fourth topology.

Usage example

>>> config = CloudOfflineConfig(transcriber="gladia", languages=["fr"])
>>> config.transcriber, config.diarizer
('gladia', 'pyannoteai')
>>> CloudOfflineConfig(transcriber="whisper").transcriber
Traceback (most recent call last):
    ...
ValueError: unknown transcriber 'whisper'; choose from elevenlabs, gladia

class vocal_helper.cloud.CloudLivePipeline(asr_ws, diar_ws, *, pre_send=None)

Bases: object

Paid live transcription and diarization: one PCM stream, two providers, fused.

The topology is a fan-out, not a chain: the same audio goes to Gladia (for words) and to pyannoteAI Live-1 (for speakers), concurrently with receiving from both, and the two answers are joined at the end. It has to be this way, because Gladia does not diarize live.

Both connections are injected rather than opened here, which is what makes this class testable against a local mock server and keeps credentials out of it. A caller owns the async with websockets.connect(...) scope for each, using the URLs from vocal_helper.asr_cloud.open_gladia_live_session() and vocal_helper.diar_cloud.open_pyannoteai_live_stream().

Parameters:

• asr_ws (Any) – Open connection to the transcription socket.

• diar_ws (Any) – Open connection to the diarization socket.

• pre_send (callable, optional) – async (ndarray) -> ndarray, applied to every frame before it leaves the machine. Identity by default. This is the anonymisation injection point, and the reason it lives here is a separation of duties: this library provides transport and a hook, and never personal-data policy, which belongs to the caller who knows what must be removed.

Examples

>>> import asyncio, json, numpy as np
>>> class FakeWs:
...     def __init__(self, replies): self.replies, self.sent = replies, []
...     async def send(self, payload): self.sent.append(payload)
...     def __aiter__(self):
...         async def gen():
...             for r in self.replies: yield json.dumps(r)
...         return gen()
>>> async def demo():
...     asr = FakeWs([{"type": "final", "words": [
...         {"word": "oui", "time_begin": 0.1, "time_end": 0.4}]}])
...     diar = FakeWs([{"type": "diarization_speaker_start",
...                     "data": {"timestamp": 0.0, "speaker": "S0"}},
...                    {"type": "diarization_speaker_end",
...                     "data": {"timestamp": 1.0, "speaker": "S0"}}])
...     async def source():
...         yield np.zeros(1600, dtype=np.float32)
...     out = await CloudLivePipeline(asr, diar).run(source())
...     return [(u["speaker"], u["text"]) for u in out], len(asr.sent), len(diar.sent)
>>> asyncio.run(demo())
([('S0', 'oui')], 1, 1)

async run(pcm_source, *, stop=None)

Fan pcm_source out to both providers and return the fused utterances.

Parameters:

• pcm_source (async iterator of ndarray) – Mono float32 chunks, from any of vocal_helper.sources.

• stop (asyncio.Event, keyword-only, optional) – Set externally — a “stop” button, a maximum-duration timer — to end the session without waiting for the source to exhaust itself. Neither a microphone nor an open-ended live URL ends on its own, so a live session without this has no way to finish.

Returns:

The fused record. Partial transcripts are excluded: they get revised, and a record must not contain retracted words.

Return type:

list of CloudUtterance

Notes

Three tasks run concurrently: sending audio, draining transcripts, draining speaker events. The two drain loops end when their socket closes, and this method does not close either one — the caller owns that scope. End of stream is signalled to the diarizer (which documents a message for it); Gladia documents no equivalent, so closing the socket is what ends it.

class vocal_helper.cloud.CloudOfflineConfig(transcriber='gladia', diarizer='pyannoteai', diarizer_model='precision-3', languages=None, code_switching=True, glossary=None, num_speakers=None, min_speakers=None, max_speakers=None, overlap_s=120.0)

Bases: object

What a paid offline run should do.

Parameters:

• transcriber (str, optional) – "gladia" (default, solaria-3) or "elevenlabs" (Scribe).

• diarizer (str, optional) – "pyannoteai", or "" to keep the transcriber’s own speakers. Gladia diarizes its pre-recorded output, and Scribe bundles diarization too, but on a crowded panel the hosted diarizer is what the money is for — so it is on by default.

• diarizer_model (str, optional) – Defaults to the provider’s most accurate batch model. Named rather than left to the client library’s default, which is the previous generation: on a paid call, accepting a default is not a neutral act.

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

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

• glossary (list of str, optional) – Domain terms and proper nouns to bias toward. Never part of the billing ledger key, so editing it and re-running the same audio costs nothing.

• num_speakers (int, optional) – Diarization hints. Prefer ``min_speakers``: a roster is a floor, not a cap, and an exact count will merge or split to hit the number even when the audio disagrees.

• min_speakers (int, optional) – Diarization hints. Prefer ``min_speakers``: a roster is a floor, not a cap, and an exact count will merge or split to hit the number even when the audio disagrees.

• max_speakers (int, optional) – Diarization hints. Prefer ``min_speakers``: a roster is a floor, not a cap, and an exact count will merge or split to hit the number even when the audio disagrees.

• overlap_s (float, optional) – Shared audio between chunks when the media exceeds the provider’s cap.

Raises:

ValueError – On an unknown transcriber or diarizer, at construction time rather than halfway through a billed run.

Examples

>>> CloudOfflineConfig().transcriber
'gladia'
>>> CloudOfflineConfig(diarizer="").diarizer
''
>>> CloudOfflineConfig(diarizer="whisper")
Traceback (most recent call last):
    ...
ValueError: unknown diarizer 'whisper'; choose from '' or pyannoteai

code_switching: bool = True

diarizer: str = 'pyannoteai'

diarizer_model: str = 'precision-3'

glossary: list[str] | None = None

languages: list[str] | None = None

max_speakers: int | None = None

min_speakers: int | None = None

num_speakers: int | None = None

overlap_s: float = 120.0

transcriber: str = 'gladia'

class vocal_helper.cloud.CloudOfflinePipeline(config=None)

Bases: object

Paid offline transcription and diarization: a file in, utterances out.

Unlike the local pipelines, there is no PCM stream to drive: the providers take a whole file, so this class is a plain call rather than an async generator. The output type is the same either way, so a consumer does not branch on the mode.

Parameters:

config (CloudOfflineConfig, optional) – Defaults to Gladia solaria-3 plus pyannoteAI, which is the combination this suite measured and bought.

Examples

>>> pipeline = CloudOfflinePipeline()
>>> pipeline.config.transcriber, pipeline.config.diarizer
('gladia', 'pyannoteai')
>>> pipeline.engine_name()
'gladia:solaria-3+pyannoteai:precision-3'
>>> CloudOfflinePipeline(CloudOfflineConfig(diarizer="")).engine_name()
'gladia:solaria-3'

engine_name()

Name the engines that produced a result, for the record it ends up in.

Returns:

e.g. "gladia:solaria-3+pyannoteai:precision-3". The model, not just the provider: two runs through the same provider on different generations are not the same measurement, and a record that hid which one produced it could not be compared against itself a year later.

Return type:

str

run(audio)

Transcribe and diarize audio, returning what the engines said.

Raw, deliberately. Rows come back as the providers produced them: a speaker the diarizer discovered for four words is still here, a block that merges thirty seconds of speech is still one row, and a row whose times collapsed to an instant still has none. Those are facts about the engines, and a library that quietly edited them would be deciding, on a caller’s behalf, which voices count.

The next step is explicit, and almost always wanted:

rows = CloudOfflinePipeline(config).run(audio)
rows, speakers = voh.consolidate(rows, min_speaker_s=120.0)

See vocal_helper.turns.consolidate() for the five rules it applies and the complaint behind each.

Parameters:

audio (str) – Media path. Any container; the waveform is what gets uploaded.

Returns:

In time order, each naming the engine that produced it.

Return type:

list of CloudUtterance

Raises:

RuntimeError – When a required credential is missing, or a provider refuses the call. Never a silent fallback to a local engine.

vocal_helper.cloud.TRANSCRIBERS = ('gladia', 'elevenlabs')

The hosted transcribers. gladia is the default; elevenlabs is selectable.