vocal_helper.diar_cloud module

vocal_helper.diar_cloud

Hosted diarization: pyannoteAI, pre-recorded and live. The paid half of the diarization stage, mirroring vocal_helper.diar on the other side of the network.

Extracted from notes_helper.cloud_transcribe and notes_helper.live_cloud.diar_cloud (same author), where this code was written against the real APIs.

Why the paid diarizer is the one worth paying for

Measured on this suite’s own material, the free local path does not lose on the text — its transcription confidence sits close to the hosted one. It collapses on the speakers: a six-voice broadcast panel came back as roughly two clusters, hundreds of turns merged into one.

Diarization is also, by a wide margin, the cheaper of the two hosted stages. So the interesting purchase is not “buy the cloud”, it is buy the diarization and keep the local transcription — the hybrid that vocal_helper.cloud.CloudOfflinePipeline exposes as a first-class mode, because an economy of that size has no business hiding in a configuration key.

Why no speaker count is passed by default

Discovering how many people are talking is exactly what this provider is good at, and pinning the count can wedge the segmentation on a wrong one. Left alone, the call is a pure function of the audio, so the ledger keys on the bytes alone and a recording diarizes once, ever.

Three opt-in hints exist, all part of the ledger key so a hinted run is cached separately from the auto-detected one, and so the hint actually takes effect:

min_speakers — prefer this one. A roster is a floor, never a cap: these voices are known to exist, and an unnamed extra one stays possible. It lets the provider keep separating genuinely distinct voices instead of merging them to fit a number.
max_speakers — an upper bound, for when over-splitting (one voice fragmented across several clusters) is the known failure instead.
num_speakers — an exact count. The most constraining of the three: it will merge or split to hit the number even when the audio disagrees. Use it only when the total is genuinely fixed and an extra voice is impossible.

Which model, and why it is named here rather than left to a default

Verified against the provider’s documentation and the installed client on 2026-11-01, because the answer had changed since this code was first written.

Batch: ``precision-3``. It is the provider’s most accurate batch model, and the only one with tunable voice-activity and cross-talk sensitivity. The client library defaults to precision-2, the previous generation — so a caller who says nothing pays the full price for the older model. That is why the default is set here explicitly: on a paid call, taking a library’s default is not a neutral act. Published comparisons give precision-3 as 34.2 % more accurate than community-1 against 28 % for precision-2, and the provider’s own launch note puts precision-3 at 10.4 % lower average error than precision-2.

Live: ``live-1``. Not a preference — it is the only streaming model the service offers.

One consequence for the ledger: the model is part of the billing key. A run asking for precision-3 must not be served a cached precision-2 result, or the upgrade would be invisible and unpaid-for in the worst way: the better model appears to have changed nothing.

Note that confidence is not supported by precision-3 (turnLevelConfidence is), so a caller wanting per-segment confidence has to choose between the newest model and that one output.

Live-1’s envelope, which the caller must respect

16 kHz mono, chunks of about 100 ms, under 300 ms of latency, at most 8 speakers and at most 10 parallel streams, for sessions up to 5 hours. It is diarization-only: the official pattern is to send the same stream to a transcriber and to Live-1, then fuse — which is what vocal_helper.fusion does, and why it is not optional.

Usage example

>>> parse_pyannote_message({"type": "diarization_speaker_start",
...                         "data": {"timestamp": 1.5, "speaker": "SPEAKER_00"}})
DiarEvent(kind='start', timestamp=1.5, speaker='SPEAKER_00', message=None)
>>> parse_pyannote_message({"type": "keepalive"}) is None
True

vocal_helper.diar_cloud.BATCH_MODEL = 'precision-3': The most accurate batch model. The client library would default to precision-2.

class vocal_helper.diar_cloud.DiarEvent(kind, timestamp, speaker, message=None)[source]

Bases: object

One speaker-turn boundary, parsed and transport-agnostic.

Parameters:

kind (str)
timestamp (float | None)
speaker (str | None)
message (str | None)

kind: str: "start", "end" or "error".

message: str | None = None

speaker: str | None

timestamp: float | None

vocal_helper.diar_cloud.LIVE_MODEL = 'live-1': The only streaming model the service offers. A constraint, not a choice.

vocal_helper.diar_cloud.PCM_CHUNK_BYTES = 6400: 1600 samples × 4 bytes (float32) mono at 16 kHz, i.e. one 100 ms chunk.

vocal_helper.diar_cloud.PCM_CHUNK_MS = 100: The cadence the protocol is tuned for (not enforced client-side).

vocal_helper.diar_cloud.PCM_SAMPLE_RATE = 16000: Hz — the only sample rate Live-1 accepts.

class vocal_helper.diar_cloud.PyannoteAiLiveDiar(ws)[source]

Bases: object

Stream PCM over an open WebSocket and yield parsed speaker events.

The connection is injected, never opened here, for the same reason as vocal_helper.asr_cloud.GladiaLiveASR: production passes a real connection, a test passes a mock, and this class touches neither the network nor a credential.

Parameters:: ws (Any) – An open connection exposing await ws.send(...) and async iteration.

Examples

>>> import asyncio, json
>>> 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": "diarization_speaker_start",
...                               "data": {"timestamp": 0.0, "speaker": "S0"}})
...         return gen()
>>> async def demo():
...     ws = FakeWs()
...     diar = PyannoteAiLiveDiar(ws)
...     await diar.end_stream()
...     return [e.speaker async for e in diar.events()], ws.sent
>>> asyncio.run(demo())
(['S0'], ['{"type": "end_of_stream"}'])

async end_stream()[source]

Signal end of stream, so the server flushes its final events and closes.

Return type:: None

async events()[source]

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

Return type:: AsyncIterator[DiarEvent]

async send_audio(pcm_f32le)[source]

Send one raw PCM chunk; Live-1 expects roughly 100 ms per chunk.

Parameters:: pcm_f32le (bytes)
Return type:: None

vocal_helper.diar_cloud.open_pyannoteai_live_stream()[source]

Open a Live-1 session and return its identifier and socket URL.

Returns:: {"id", "url"}.
Return type:: dict of str
Raises:: RuntimeError – When PYANNOTEAI_API_KEY is not configured.

vocal_helper.diar_cloud.parse_pyannote_message(raw)[source]

Parse one decoded Live-1 message, or None when it is not an event.

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

Deliberately tolerant about where the fields sit. The payload may be wrapped in data or flat at the top level, the clock may be called timestamp, time, start or end, and the label speaker, speaker_label or label. A stricter reader is not safer here: it returns no event rather than an error, so the whole session yields no speaker turns and nothing says why — which is exactly what happened before this lookup existed.

Parameters:: raw (dict) – One decoded JSON message.
Returns:: None for messages that carry no boundary and no error.
Return type:: DiarEvent or None

Examples

>>> parse_pyannote_message({"type": "diarization_speaker_end",
...                         "data": {"timestamp": 9.0, "speaker": "S1"}}).kind
'end'

La forme de l’exemple officiel : enveloppe data, horloge start.

>>> event = parse_pyannote_message({"type": "diarization_speaker_start",
...                                 "data": {"speaker": "SPEAKER_0", "start": 1.5}})
>>> (event.kind, event.speaker, event.timestamp)
('start', 'SPEAKER_0', 1.5)

À plat, sans enveloppe :

>>> event = parse_pyannote_message({"type": "diarization_speaker_start",
...                                 "label": "S0", "time": 2.0})
>>> (event.speaker, event.timestamp)
('S0', 2.0)

>>> parse_pyannote_message({"type": "error", "message": "quota"}).message
'quota'

vocal_helper.diar_cloud.pcm_f32_to_wire_bytes(pcm)[source]

Mono float32 samples to Live-1’s f32le wire format.

A reinterpretation, not a conversion: the source frame contract already yields float32, so there is no resampling and no requantisation to get wrong here.

Parameters:: pcm (numpy.ndarray) – Mono float32 samples.
Returns:: Little-endian 32-bit floats.
Return type:: bytes

Examples

>>> import numpy as np
>>> len(pcm_f32_to_wire_bytes(np.zeros(1600, dtype=np.float32))) == PCM_CHUNK_BYTES
True

vocal_helper.diar_cloud.pyannoteai_diarize(*, audio, model='precision-3', num_speakers=None, min_speakers=None, max_speakers=None, vad_sensitivity=None, crosstalk_sensitivity=None)[source]

Diarize one file with pyannoteAI. Billed once per audio, model and hint set.

Parameters:

audio (str, keyword-only) – Media path. Converted to 16 kHz mono MP3 before upload, so a video recording does not send its video track for no benefit.
model (str, optional) – Defaults to BATCH_MODEL ("precision-3"), the most accurate one. The client library’s own default is the previous generation, which is why this is set here rather than omitted. Part of the ledger key, so an upgrade is not served a stale cheaper result.
num_speakers (int, optional) – The three opt-in hints. See the module docstring for which to reach for; the short version is prefer ``min_speakers``.
min_speakers (int, optional) – The three opt-in hints. See the module docstring for which to reach for; the short version is prefer ``min_speakers``.
max_speakers (int, optional) – The three opt-in hints. See the module docstring for which to reach for; the short version is prefer ``min_speakers``.
vad_sensitivity (float, optional) – precision-3 only: how eagerly speech and overlapping speech are detected. Left unset unless asked for, because the provider’s defaults are tuned and a guess here moves every segment boundary at once.
crosstalk_sensitivity (float, optional) – precision-3 only: how eagerly speech and overlapping speech are detected. Left unset unless asked for, because the provider’s defaults are tuned and a guess here moves every segment boundary at once.

Returns:

Segments, each with start, end and speaker.

Return type:

list of dict

Raises:

RuntimeError – When the key is missing, or the response carries no segments. The result document’s shape has moved between versions, so three known layouts are tried before giving up — and giving up says so rather than returning an empty list, which would read downstream as “nobody spoke”.