vocal_helper.sources module

vocal_helper.sources

Async iterators that produce vocal_helper.types.PcmFrame events from a few canonical inputs.

Four sources ship in v0.1.0 :

  • from_microphone() — wraps capture_helper.iter_mic_audio (optional dependency ; requires vocal-helper[mic]).

  • from_url() — wraps podcast_helper.extract_audio_stream to consume any URL yt-dlp can reach (YouTube / Vimeo / Twitch VOD or live, SoundCloud, podcast RSS feeds, direct HLS / m3u8, direct audio files) as a paced PCM stream (optional dependency ; requires vocal-helper[stream]).

  • from_wav_file() — replays a 16 kHz mono WAV at real_time=True (default) so the downstream stages see the same pacing as a live stream, or as fast as possible when real_time=False (for offline batch tests).

  • from_numpy_array() — yields frames from an existing PCM buffer ; useful for unit tests and synthetic streams.

The contract is the same in all three cases :

async for frame in source(…):

# frame[“pcm”].shape == (frame_samples,) # frame[“sample_rate”] == 16000 # frame[“t0”] == seconds since first frame await queue.put(frame)

Author

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

async vocal_helper.sources.from_microphone(*, sample_rate=16000, frame_ms=20, device_name=None, device_index=None)[source]

Yield PCM frames from the system’s default (or named) microphone.

Requires the mic extra (pip install vocal-helper[mic]).

Parameters:

  • sample_rate (int) – Target rate in Hz. Default 16 kHz.

  • frame_ms (int) – Frame length in milliseconds. 20 ms is the default — matches capture_helper and lines up with Silero VAD’s 32 ms window cleanly.

  • device_name (str, optional) – Substring of the device name to pick.

  • device_index (int, optional) – Concrete device index (taken from capture_helper.list_sources).

Return type:

AsyncIterator[PcmFrame]

async vocal_helper.sources.from_numpy_array(pcm, *, sample_rate=16000, frame_ms=20, real_time=False)[source]

Yield PCM frames from an in-memory mono float32 buffer.

Parameters:

  • pcm (ndarray[tuple[Any, ...], dtype[float32]])

  • sample_rate (int)

  • frame_ms (int)

  • real_time (bool)

Return type:

AsyncIterator[PcmFrame]

async vocal_helper.sources.from_url(url, *, realtime=True, speed=1.0, headers=None, cookies_from_browser=None, record_to=None)[source]

Yield PCM frames from any URL (YouTube / Twitch / RSS / direct audio).

Async wrapper over podcast_helper.extract_audio_stream() that enforces the speech-pipeline contract every downstream stage relies on. We deliberately do NOT expose sample_rate, frame_ms or to_mono kwargs : vocal-helper hardcodes the only configuration that keeps Silero VAD, pyannote/embedding and whisper.cpp all happy at once.

The enforced contract

  • target_sample_rate = 16_000 — Silero VAD ONNX v5 and pyannote/embedding are both trained at 16 kHz. Any other value would force re-sampling inside the stages.

  • to_mono = True — pyannote, whisper and the VAD all expect a single channel.

  • frame_ms = 20 — 320 samples ; aligns with Silero’s native 32 ms window with one carry-over slice, so no re-buffering.

  • dtype = float32 ∈ [-1, +1] — what whisper.cpp and pyannote ingest natively. ffmpeg’s libswresample applies the Shannon-Nyquist anti-aliasing low-pass at the new Nyquist when down-sampling, so spectral aliasing is impossible.

Callers that need a different shape should drop the helper and call podcast_helper.extract_audio_stream directly — those callers are by definition outside the speech pipeline.

Requires podcast-helper (pip install vocal-helper[stream]) plus ffmpeg and yt-dlp on PATH.

param url:

File path, file:// URL, direct audio URL (MP3 / M4A / Opus / WAV / HLS m3u8), RSS feed URL (auto-picks latest episode), or any yt-dlp-supported URL — YouTube, Vimeo, SoundCloud, Twitch VOD and live, etc. Spotify-protected and Apple Podcasts catalog URLs raise NotImplementedError with hints.

type url:

str

param realtime:

Pace decoding at wall-clock (ffmpeg’s -re). Set False for burst-decoding a VOD (offline benchmarking). Live sources pace themselves, so podcast-helper forces this to False internally.

type realtime:

bool, default True

param speed:

Playback rate for VOD only. Implemented via ffmpeg’s atempo= filter so pitch is preserved. 2.0 doubles ASR throughput on long episodes ; 0.5 slows down for proofing. Raises ValueError on live streams.

type speed:

float, default 1.0

param headers:

HTTP headers ffmpeg should send. Merged on top of yt-dlp’s per-source headers (your keys win).

type headers:

dict[str, str], optional

param cookies_from_browser:

"firefox" / "chrome" / "safari" / etc. — used by yt-dlp for age-gated, members-only or private content.

type cookies_from_browser:

str, optional

param record_to:

If set, ffmpeg writes a parallel compressed archive of the same audio to this path while the live PCM stream is consumed. Single decode, two encoder paths. See podcast_helper docs for the codec-by-extension table.

type record_to:

str or Path, optional

Yields:

PcmFrame – t0 in seconds since the source started, sample_rate == 16_000, pcm mono float32 of length 320 (= 16 000 × 20 / 1 000).

raises RuntimeError:

If the upstream frame violates the speech contract — wrong sample rate, wrong dtype, non-mono. Fail-loud is the point : a silent contract drift would corrupt every downstream stage.

Examples

>>> import asyncio, vocal_helper as voh
>>> async def main():
...     pipeline = voh.Pipeline(
...         source=lambda: voh.sources.from_url(
...             "https://www.youtube.com/watch?v=YE7VzlLtp-4",
...         ),
...         config=voh.PipelineConfig(
...             diar={"backend": "pyannote"},
...             llm={"model": "gemma4:e4b"},
...         ),
...     )
...     async for ev in pipeline.run():
...         if isinstance(ev, dict) and "text" in ev:
...             print(f"[{ev['t0']:.1f}s {ev['speaker']}] {ev['text']}")
>>> asyncio.run(main())
Parameters:
Return type:

AsyncIterator[PcmFrame]

async vocal_helper.sources.from_wav_file(path, *, sample_rate=16000, frame_ms=20, real_time=True)[source]

Yield PCM frames from a mono WAV file.

Pacing :

  • real_time=True — sleep frame_ms between frames so the downstream pipeline sees the same cadence as a live source.

  • real_time=False — yield as fast as possible. Useful for benchmarking and unit tests.

Parameters:
Return type:

AsyncIterator[PcmFrame]

vocal_helper.sources.probe_duration_s(path)[source]

Best-effort media duration in seconds, for the diarization router.

The router (vocal_helper.router.select_diarization()) needs the audio length to choose the offline backend — short/dense → NeMo Sortformer, long → pyannote. This is a cheap metadata read (ffprobe via audio_helper), not a full decode, so it is safe to call before building the pipeline. Any failure (unreadable file, missing probe backend) returns None, which the router treats as “unknown length” and routes to the robust long-form branch.

Parameters:

path (str or pathlib.Path) – Path to any ffmpeg-decodable media file (wav / mp3 / m4a / video / …).

Returns:

Duration in seconds when it can be read and is positive, else None.

Return type:

float or None

Examples

>>> probe_duration_s("/nonexistent.wav") is None
True