vocal_helper.fanout module

vocal_helper.fanout

Fork one queue into several, so a producer never has to know who is listening.

Why the pipeline needed this

Pipeline wires its raw-audio queue to exactly one consumer, the voice-activity stage. That was enough while every diarizer in the package worked on voiced segments, and it stops being enough the moment one works on the audio itself: DiartDiarStage consumes PCM directly, and an end-of-stream batch diarizer wants the whole stream. Neither can be attached without a second reader, and making the source aware of them would put the knowledge of who consumes what back into the thing that should know least.

Ported from operator-assistance-streaming/streaming/multiplex.py, which is being absorbed.

The hazard, stated rather than hidden

Backpressure is honoured per sink, so the slowest sink paces every other one. That is the right default — nothing is silently lost — and it is a real trap: attach a display that stalls and the transcription stalls with it, which looks like a slow model and is not. The original carried that caveat in a docstring; here it is also a parameter, because the two cases want opposite answers.

A stage whose output must be complete (transcription, a stored record) is lossless: it has to wait. A sink whose only job is to show the present (a live meter, a preview pane) is better served by LOSSY, where a full queue drops the oldest frame rather than holding up the session. Choosing wrong is not a crash, it is a session that feels broken for a reason nobody can see — which is why the choice is explicit and has no silent default.

Measuring it

on_wait reports how long each put blocked, per sink. That number is the difference between “the model is slow” and “a consumer is starving everyone”, and it is what vocal_helper.live_metrics.LiveMetrics.record_queue_wait() was built to receive.

Author

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

vocal_helper.fanout.LOSSLESS = 'lossless'

wait for room. Nothing is lost, and this sink can pace the whole fan-out.

Type:

Sink policy

vocal_helper.fanout.LOSSY = 'lossy'

when full, drop the OLDEST pending item to make room. For consumers that show the present rather than record it — a stale frame on a live meter is worth less than a stalled session, and dropping the oldest keeps what is displayed as current as possible.

Type:

Sink policy

class vocal_helper.fanout.Sink(queue, *, policy='lossless', name='')[source]

Bases: object

One destination of a fan-out: a queue plus what to do when it is full.

Parameters:

  • queue (asyncio.Queue) – Where items go. Its maxsize is what makes the policy matter at all; an unbounded queue never blocks and never drops, it just grows until the process dies.

  • policy (str, optional) – LOSSLESS (default) or LOSSY.

  • name (str, optional) – Used in on_wait reports, so a queue-wait figure can be attributed to a consumer rather than to “somewhere in the pipeline”.

dropped

Items dropped because this sink was full. Non-zero is information, not an error — it says a lossy consumer could not keep up, which is exactly what it opted into.

async vocal_helper.fanout.fanout(source, sinks, *, on_wait=None)[source]

Pull from source and forward every item to every sink until the stream ends.

The terminating None sentinel always reaches every sink, whatever its policy: losing an end-of-stream marker is not a late frame, it is a consumer that waits forever. On a lossy sink it makes room for itself the same way data does, rather than blocking on a queue nobody is draining — the first version awaited it unconditionally and deadlocked. On cancellation the sentinel is pushed too, so no downstream stage is left hanging on a session that was torn down.

Parameters:

  • source (asyncio.Queue) – The producer’s queue. A bare queue in sinks is wrapped as a lossless Sink, so the simple call stays simple.

  • on_wait (callable, optional) – (sink_name, seconds) for every put that had to wait. Feed it to LiveMetrics.record_queue_wait() to find out which consumer is pacing the rest.

  • sinks (Sequence[Sink | Queue[T | None]])

Return type:

None