vocal_helper.fusion module

vocal_helper.fusion

Putting words and speakers back together.

Both paid modes split the two questions “what was said?” and “who said it?” between two providers, so something has to join the answers. That is this module, and in the live mode it is not optional: Gladia does not diarize live, so a live transcript without fusion has no speakers at all.

Extracted from notes_helper.live_cloud.fusion (same author), widened to cover the pre-recorded path as well.

Two topologies, one arithmetic

Live: the transcriber emits word-level events as it goes, the diarizer emits speaker boundaries as it goes, and nothing is sorted or complete. Boundaries are paired into intervals, then each final utterance takes the speaker it overlaps most.

Pre-recorded: the transcriber returns whole utterances with its own speaker labels, and the diarizer returns better ones. Each utterance is relabelled by the diarization segment it overlaps most, which is how a hybrid run gets local transcription timing with hosted speaker quality.

Either way the operation is the same: attribute an interval to whoever occupies it most. Doing it by overlap rather than by midpoint matters, because an utterance that straddles a turn change belongs to whoever held most of it, not to whoever happened to hold its centre.

Only finals are fused

Partial transcripts exist for display: they are revised as the model hears more. Fusing them would put retracted words in the record. The contract is “final wins”, which is also what any downstream report already expects.

An interval still open when fusion runs

Live diarization emits a start long before the matching end: the speaker is still talking. Such a turn is treated as covering up to now (by default, the utterance’s own end), which is the streaming equivalent of the pre-recorded path’s whole-file window. Without it, the current speaker would be the one speaker who never gets credited.

Usage example

>>> from vocal_helper.diar_cloud import DiarEvent
>>> events = [DiarEvent("start", 0.0, "S0"), DiarEvent("end", 2.0, "S0"),
...           DiarEvent("start", 2.0, "S1")]
>>> [(t.speaker, t.start, t.end) for t in build_speaker_turns(events)]
[('S0', 0.0, 2.0), ('S1', 2.0, None)]

class vocal_helper.fusion.SpeakerTurn(speaker, start, end=None)

Bases: object

One diarized interval. end is None while the speaker is still talking.

Parameters:

• speaker (str)

• start (float)

• end (float | None)

end: float | None = None

speaker: str

start: float

vocal_helper.fusion.build_speaker_turns(diar_events)

Pair start and end events into intervals.

Three defensive rules, because a live stream is not a well-formed document.

An end with no open start for that speaker is ignored: the protocol should never emit it, and a malformed or mocked stream must not crash fusion.

A second start for a speaker whose turn is still open closes the stale one first. A real stream keeps labels stable for a whole session, so two consecutive start events for the same speaker mean the first one’s end was dropped — not that one person is somehow speaking twice at once.

An end carrying no speaker label closes the only open turn when there is exactly one, and is ignored otherwise. The provider does omit the label on some end frames, and when a single turn is open there is nothing to guess; with several open, guessing would silently attribute speech to the wrong person, which is worse than losing a boundary.

Parameters:

diar_events (list of DiarEvent) – Events in arrival order. Events missing a speaker or a timestamp are skipped.

Returns:

In the order the turns opened. The last turn of each speaker may still be open.

Return type:

list of SpeakerTurn

Examples

>>> from vocal_helper.diar_cloud import DiarEvent
>>> turns = build_speaker_turns([DiarEvent("start", 1.0, "A"),
...                              DiarEvent("start", 3.0, "A")])
>>> [(t.start, t.end) for t in turns]
[(1.0, 3.0), (3.0, None)]
>>> build_speaker_turns([DiarEvent("end", 1.0, "ghost")])
[]

Une fin sans étiquette, un seul tour ouvert : il n’y a rien à deviner.

>>> turns = build_speaker_turns([DiarEvent("start", 0.0, "A"),
...                              DiarEvent("end", 2.0, None)])
>>> [(t.speaker, t.start, t.end) for t in turns]
[('A', 0.0, 2.0)]

Deux tours ouverts : on perd la frontière plutôt que de l’attribuer au hasard.

>>> turns = build_speaker_turns([DiarEvent("start", 0.0, "A"),
...                              DiarEvent("start", 1.0, "B"),
...                              DiarEvent("end", 2.0, None)])
>>> [t.end for t in turns]
[None, None]

vocal_helper.fusion.fuse_live_transcripts(transcripts, diar_events, *, now=None, engine='gladia:solaria-1+pyannoteai:live-1')

Label each final live utterance with the speaker it overlaps most.

Parameters:

• transcripts (list of AsrTranscript) – Everything the transcriber emitted. Partials are ignored.

• diar_events (list of DiarEvent) – Everything the diarizer emitted.

• now (float, keyword-only, optional) – The moment fusion runs, used to close still-open turns. Defaults to each utterance’s own end.

• engine (str, keyword-only, optional) – Recorded on every utterance, so a record names what produced it.

Returns:

In transcript order. speaker is "?" when no turn overlapped at all, which is honest: an unknown speaker is not speaker zero.

Return type:

list of CloudUtterance

Examples

>>> from vocal_helper.asr_cloud import AsrTranscript, AsrWord
>>> from vocal_helper.diar_cloud import DiarEvent
>>> finals = [AsrTranscript("final", [AsrWord("oui", 0.1, 0.5, 0.9)])]
>>> events = [DiarEvent("start", 0.0, "S0"), DiarEvent("end", 1.0, "S0")]
>>> out = fuse_live_transcripts(finals, events)
>>> out[0]["speaker"], out[0]["text"], out[0]["word_confidence"]
('S0', 'oui', [0.9])
>>> fuse_live_transcripts([AsrTranscript("partial", [])], events)
[]

vocal_helper.fusion.relabel_utterances(utterances, diar_segments, *, engine=None)

Replace each utterance’s speaker with the diarization segment it overlaps most.

This is the pre-recorded fusion, and the whole point of the hybrid mode: keep the transcriber’s words and timings, take the diarizer’s speakers.

Parameters:

• utterances (list of CloudUtterance) – From any transcriber, with whatever speakers it guessed.

• diar_segments (list of dict) – From the diarizer, each with start, end and speaker. Order is irrelevant, and segments may genuinely overlap on cross-talk.

• engine (str, keyword-only, optional) – When given, recorded on every utterance, replacing whatever was there.

Returns:

The same utterances, relabelled. An utterance that overlaps no segment keeps its original speaker rather than losing it: a gap in the diarization is not evidence about who was talking.

Return type:

list of CloudUtterance

Examples

>>> utts = [{"t0": 0.0, "t1": 1.0, "speaker": "spk_0", "text": "a",
...          "words": [(0.0, 1.0, "a")], "language": None}]
>>> segs = [{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_01"}]
>>> relabel_utterances(utts, segs)[0]["speaker"]
'SPEAKER_01'
>>> relabel_utterances(utts, [])[0]["speaker"]
'spk_0'