capture_helper.mic module

capture_helper.mic

Live microphone-PCM async iterator that bridges capture_helper.list_sources() output with the same PCM-frame contract that podcast_helper.extract_audio_stream() uses for URL-based audio. The intent is: if you can drop a podcast URL into an ASR / VAD pipeline, you can drop a live mic in.

Implementation

  • Shells out to ffmpeg with the right per-OS input driver (-f avfoundation / -f pulse / -f alsa / -f dshow) built by capture_helper.sources.ffmpeg_input_args().

  • Asks ffmpeg to resample to target_sample_rate and downmix to mono (or preserve the source channel count) and emit raw f32le PCM to stdout. We read fixed-size byte blocks and yield MicFrame-typed dicts.

  • Async generator — matches podcast_helper.extract_audio_stream()’s shape so consumers can async for frame in ... regardless of whether the source is a URL or a local microphone.

Usage example

>>> import asyncio, capture_helper as ch
>>> async def main():
...     mic = ch.pick_source("microphone")
...     async for frame in ch.iter_mic_audio(mic, target_sample_rate=16000,
...                                          to_mono=True, frame_ms=20):
...         # frame["pcm"]: np.float32, shape (320,) for 20ms @ 16kHz mono
...         await asr.feed(frame["pcm"])
>>> asyncio.run(main())

Author

Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/

class capture_helper.mic.MicFrame[source]

Bases: TypedDict

One PCM frame from a live microphone, in absolute capture time.

Structurally identical to podcast_helper.streaming.PcmFrame so downstream consumers (VAD, ASR, dB-meter, …) don’t have to branch on the source type.

Keys

t_abs_sfloat

Seconds since ffmpeg latched onto the device. Monotonic.

pcmNDArray[np.float32]

Audio samples in [-1.0, 1.0], float32.

  • Shape (n_samples,) when to_mono=True (default).

  • Shape (n_samples, n_channels) when to_mono=False and the source has more than one channel.

voicedbool | None

Always None here — VAD downstream fills it in if used.

pcm: ndarray[tuple[Any, ...], dtype[float32]]
t_abs_s: float
voiced: bool | None
async capture_helper.mic.iter_mic_audio(source, *, target_sample_rate=16000, to_mono=True, frame_ms=20, max_frames=None)[source]

Yield PCM frames from a live microphone.

Parameters:

  • source (Source) – Device dict returned by pick_source() / list_sources(). Its kind must be "microphone".

  • target_sample_rate (int, default 16000) – Exact output sample rate in Hz. ffmpeg resamples via libswresample with an anti-aliasing low-pass at the new Nyquist — Shannon-correct and more than enough for ASR / VAD / ML pipelines.

  • to_mono (bool, default True) –

    • True: ffmpeg standard downmix to one channel; pcm.shape == (n_samples,).

    • False: preserve the source’s native channel count; pcm.shape == (n_samples, n_channels) (only set when the source actually has more than one channel).

  • frame_ms (int, default 20) – Frame duration in milliseconds. 20 matches Silero VAD’s native frame size, avoiding a downstream re-buffer.

  • max_frames (int, optional) – Stop after yielding this many frames. None (default) = unbounded — runs until the source disconnects or the consumer breaks the async for.

Yields:

MicFrame – Successive PCM frames in absolute capture time.

Raises:

  • ValueError – If source["kind"] isn’t "microphone", or if frame_ms / target_sample_rate is non-positive.

  • FileNotFoundError – Raised by ffmpeg / asyncio if ffmpeg isn’t on PATH.

Return type:

AsyncIterator[MicFrame]

Examples

>>> import asyncio, capture_helper as ch
>>> async def main():
...     mic = ch.pick_source("microphone")
...     async for frame in ch.iter_mic_audio(mic):
...         # frame["pcm"]: np.float32 (320,) for 20ms @ 16kHz mono
...         pass
>>> asyncio.run(main())