capture_helper package

Submodules

Module contents

Capture Helper

Local-first, library-shaped camera / microphone capture layer for the AI Helpers stack. It ships the INPUT layer: cross-platform device enumeration + selection (cameras / microphones) and a pair of iterators that bridge those devices to the rest of the suite’s contracts — plus a live multi-source scene configurator (a browser GUI served by the FastAPI app) that turns a visual layout of live sources into a reusable JSON scene artifact the CLI / API can replay headless.

The capture layer

  • SourceKind — literal "camera" | "microphone".

  • Source — typed dict describing one device.

  • MicFrame — typed dict for one PCM frame (mirrors podcast_helper.streaming.PcmFrame).

  • list_sources() — best-effort cross-platform device enumeration via ffmpeg -list_devices (avfoundation / v4l2 / dshow / pulse / alsa). Returns [] rather than raising on unsupported platforms.

  • pick_source() — pick the first device of a given kind matching optional name_substring / index filters.

  • iter_camera_frames() — synchronous generator yielding (H, W, 3) BGR uint8 numpy arrays — same shape and dtype as video_helper.extract_frames().

  • iter_mic_audio() — async generator yielding MicFrame — same shape as podcast_helper.extract_audio_stream().

The scene configurator (additive, does not touch the iterators)

This is an early-stage project. The public iterator contracts above are stable; the scene / preview surface is new and additive.

Usage example

>>> import asyncio, capture_helper as ch
>>> cam = ch.pick_source("camera")
>>> for frame in ch.iter_camera_frames(cam, output_width=640, output_height=360,
...                                    fps=30, max_frames=300):
...     # frame.shape == (360, 640, 3), dtype uint8, BGR.
...     do_something(frame)
>>>
>>> async def listen():
...     mic = ch.pick_source("microphone")
...     async for f in ch.iter_mic_audio(mic, target_sample_rate=16000, frame_ms=20):
...         # f["pcm"].shape == (320,) — 20ms @ 16kHz mono.
...         await asr.feed(f["pcm"])
>>> asyncio.run(listen())

Author

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

class capture_helper.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
class capture_helper.ResolvedSceneSource[source]

Bases: TypedDict

A SceneSource paired with the live device it resolved to.

Keys

scene_sourceSceneSource

The original recipe from the scene.

resolvedSource | None

The live capture_helper.Source matched on this machine, or None when no device satisfied the recipe’s selectors.

errorstr | None

Human-readable reason the recipe could not be resolved, or None on success.

error: str | None
resolved: Source | None
scene_source: SceneSource
class capture_helper.Scene[source]

Bases: TypedDict

A named, serialisable multi-source composition.

Keys

format_versionint

On-disk schema version (SCENE_FORMAT_VERSION).

namestr

Human name for the scene (also the suggested filename stem).

width, heightint

Canvas size in pixels. Source tiles are positioned within this box.

sourceslist[SceneSource]

Ordered list of placed sources (draw order for cameras is by z, ties broken by list order).

format_version: int
height: int
name: str
sources: list[SceneSource]
width: int
class capture_helper.SceneSource[source]

Bases: TypedDict

One source placed on a scene canvas.

A SceneSource is a recipe, not a live handle: it records how to re-select the device (kind + optional name_substring / index), where it sits on the canvas (x/y/w/h/z), and the capture parameters to pass to the iterators when the scene is run.

Keys

idstr

Stable, GUI-generated identifier (uuid4 hex) so the front-end can address a placed source across drag / resize / delete without relying on list order.

kind"camera" | "microphone"

Which iterator this source drives.

labelstr

Human-friendly name shown in the GUI (e.g. "host cam"). Free text.

name_substringstr | None

Case-insensitive substring used to re-resolve the device via capture_helper.pick_source(). None = no name constraint.

indexint | None

Exact device index used to re-resolve the device. None = no index constraint. When both are None the first device of kind is used.

x, yint

Top-left position of the source’s tile on the canvas, in pixels.

w, hint

Tile size in pixels. For microphones (no visual) these may be 0.

zint

Stacking order — higher draws on top. Cameras only.

paramsdict

Free-form capture parameters forwarded to the iterator. For cameras: fps, output_width, output_height, pad_color. For microphones: target_sample_rate, frame_ms, to_mono. Unknown keys are ignored by the runner, so the GUI can round-trip extras safely.

h: int
id: str
index: int | None
kind: Literal['camera', 'microphone']
label: str
name_substring: str | None
params: dict[str, Any]
w: int
x: int
y: int
z: int
class capture_helper.Source[source]

Bases: TypedDict

One capture device, normalised across platforms.

Keys

kind"camera" | "microphone"

Audio / video distinction.

namestr

Human-readable name as reported by the OS driver. Use this when passing to a future iter_camera_frames / iter_mic_audio — the per-OS ffmpeg input string is built internally.

indexint

Numeric device index in ffmpeg’s listing (0-based). Some OS backends only accept the index; others accept name OR index.

platformstr

"darwin" / "linux" / "windows" — useful for callers that branch on the underlying driver (avfoundation / v4l2 / dshow / pulse).

driverstr

ffmpeg input format flag — "avfoundation" / "v4l2" / "dshow" / "pulse" / "alsa".

driver: str
index: int
kind: Literal['camera', 'microphone']
name: str
platform: str
capture_helper.add_source(scene, *, kind, name_substring=None, index=None, label=None, x=0, y=0, w=0, h=0, z=0, params=None, source_id=None)[source]

Return a copy of scene with one additional placed source.

The input scene is not mutated — a shallow copy with a new sources list is returned, so callers holding the old scene keep it intact (matches the immutable-update convention used across the suite’s config helpers).

Parameters:

  • scene (Scene) – The scene to extend.

  • kind ("camera" | "microphone") – Source kind.

  • name_substring (str, optional) – Case-insensitive device-name filter recorded for later resolution.

  • index (int, optional) – Exact device-index filter recorded for later resolution.

  • label (str, optional) – Display name. Defaults to a generated "<kind> N" label.

  • x (int) – Canvas placement (see SceneSource).

  • y (int) – Canvas placement (see SceneSource).

  • w (int) – Canvas placement (see SceneSource).

  • h (int) – Canvas placement (see SceneSource).

  • z (int) – Canvas placement (see SceneSource).

  • params (dict, optional) – Capture parameters forwarded to the iterator at run time.

  • source_id (str, optional) – Explicit id (uuid4 hex). Generated when omitted — the GUI usually supplies its own so the round-trip is stable.

Returns:

A new scene including the added source.

Return type:

Scene

Examples

>>> s = new_scene("demo")
>>> s2 = add_source(s, kind="camera", label="cam", w=1280, h=720)
>>> len(s["sources"]), len(s2["sources"])
(0, 1)
capture_helper.ffmpeg_input_args(source)[source]

Build the ffmpeg -f <driver> -i <spec> argument pair for source.

Encapsulates the per-OS quirks so capture_helper.camera and capture_helper.mic don’t have to know about them:

  • avfoundation (macOS) addresses devices as "<video_idx>:<audio_idx>" where "none" opts out of the other track.

  • v4l2 (Linux) uses the device path (/dev/videoN) reported by list_sources().

  • pulse (Linux) and alsa (Linux) take the source/device name.

  • dshow (Windows) prefixes the device name with video= or audio=.

Parameters:

source (Source) – A device dict returned by list_sources() / pick_source().

Returns:

Two elements: ["-f", "<driver>"] followed by ["-i", "<spec>"] — ready to splice into an ffmpeg command line.

Return type:

list[str]

Raises:

ValueError – If the source’s driver is unknown to this helper.

capture_helper.frame_to_jpeg(frame, *, quality=5)[source]

Encode a single BGR frame to a JPEG byte string via ffmpeg.

Parameters:

  • frame (numpy.ndarray) – A (H, W, 3) uint8 array in OpenCV BGR channel order — exactly what capture_helper.iter_camera_frames() yields.

  • quality (int, default 5) – ffmpeg -q:v value (2 best … 31 worst). Lower = larger file, better image. 5 is a good preview trade-off.

Returns:

The JPEG-encoded image.

Return type:

bytes

Raises:

Examples

>>> import numpy as np
>>> jpg = frame_to_jpeg(np.zeros((16, 16, 3), dtype=np.uint8))
>>> jpg[:2] == b"\xff\xd8"   # JPEG SOI marker
True
capture_helper.iter_camera_frames(source, *, width=None, height=None, fps=None, output_width=None, output_height=None, pad_color='black', max_frames=None)[source]

Yield live camera frames as numpy BGR uint8 arrays.

Wraps ffmpeg with the per-OS input driver picked up from source["driver"] and emits (H, W, 3) uint8 arrays in OpenCV’s BGR channel order — the same shape and dtype video_helper.extract_frames() yields. Consumers built for the file-based path therefore plug in unchanged.

Parameters:

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

  • width (int, optional) – Capture-side resolution request. Passed to ffmpeg as -video_size WxH before the input — the OS driver picks the closest supported mode if the exact value isn’t available. Leave None to use the driver’s default.

  • height (int, optional) – Capture-side resolution request. Passed to ffmpeg as -video_size WxH before the input — the OS driver picks the closest supported mode if the exact value isn’t available. Leave None to use the driver’s default.

  • fps (float, optional) – Capture-side frame rate request. Same caveat as width / height. Leave None for the driver default.

  • output_width (int, optional) –

    Post-decode output size. Behaviour mirrors video_helper.extract_frames():

    • both set → scale-fit (aspect preserved) then pad with pad_color to exactly output_width × output_height;

    • one set → scale that axis, preserve aspect on the other;

    • neither set → native camera frame size.

  • output_height (int, optional) –

    Post-decode output size. Behaviour mirrors video_helper.extract_frames():

    • both set → scale-fit (aspect preserved) then pad with pad_color to exactly output_width × output_height;

    • one set → scale that axis, preserve aspect on the other;

    • neither set → native camera frame size.

  • pad_color (str, optional) – Colour name ("black" / "white" / "#RRGGBB" / …) used when scale-fit-and-pad applies. Default "black".

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

Yields:

numpy.ndarray – Successive frames as (H, W, 3) BGR uint8 arrays. Same convention as OpenCV and video_helper.extract_frames().

Raises:

  • ValueError – If source["kind"] isn’t "camera".

  • FileNotFoundError – If ffmpeg isn’t on PATH.

  • RuntimeError – If ffmpeg exits before the first frame is decoded (typical when the OS denied camera permission or the device is in use by another process).

Return type:

Iterator[ndarray]

Examples

>>> cam = ch.pick_source("camera")
>>> for frame in ch.iter_camera_frames(cam,
...                                    output_width=224, output_height=224,
...                                    max_frames=10):
...     # frame.shape == (224, 224, 3), dtype uint8, BGR.
...     model(frame)
capture_helper.iter_camera_jpeg(source, *, output_width=480, output_height=270, fps=10.0, quality=5, max_frames=None)[source]

Yield a stream of JPEG-encoded frames from a live camera.

Each yielded item is a complete JPEG suitable for concatenation into an multipart/x-mixed-replace MJPEG HTTP response (see capture_helper.api()).

Parameters:

  • source (Source) – A "camera" device.

  • output_width (int) – Preview size per frame.

  • output_height (int) – Preview size per frame.

  • fps (float, default 10.0) – Capture-side frame rate — kept low so the preview is cheap.

  • quality (int, default 5) – JPEG quality per frame.

  • max_frames (int, optional) – Stop after this many frames. None = until the consumer disconnects.

Yields:

bytes – Successive JPEG images.

Raises:

ValueError – If source is not a camera.

Return type:

Iterator[bytes]

Examples

>>> cam = ...
>>> for jpg in iter_camera_jpeg(cam, max_frames=3):
...     handle(jpg)
async capture_helper.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())
capture_helper.list_sources(kind=None)[source]

Enumerate available capture devices on the current OS.

Parameters:

kind ("camera" | "microphone" | None) – Filter to one kind. None (default) returns both.

Returns:

Possibly empty if ffmpeg is missing or the OS driver returns nothing parseable. Never raises.

Return type:

list[Source]

Examples

>>> from capture_helper import list_sources
>>> for s in list_sources("microphone"):
...     print(f"[{s['index']}] {s['name']} (driver={s['driver']})")
capture_helper.load_scene(path)[source]

Load and validate a scene from a JSON file.

Parameters:

path (str or Path) – Path to a scene JSON file.

Returns:

The parsed, validated scene.

Return type:

Scene

Raises:

Examples

>>> import tempfile, os
>>> p = os.path.join(tempfile.mkdtemp(), "demo.scene.json")
>>> _ = save_scene(new_scene("demo"), p)
>>> load_scene(p)["name"]
'demo'
async capture_helper.mic_level(source, *, target_sample_rate=16000, frame_ms=20, window_ms=200)[source]

Capture a short slice of live audio and reduce it to a level reading.

Drives capture_helper.iter_mic_audio() for roughly window_ms of audio, then returns RMS / peak levels — the GUI polls this to animate a VU meter without holding a long-lived stream open.

Parameters:

  • source (Source) – A "microphone" device.

  • target_sample_rate (int, default 16000) – Sample rate for the capture (matches the iterator default).

  • frame_ms (int, default 20) – Per-frame duration fed to the iterator.

  • window_ms (int, default 200) – Approximate total capture window in milliseconds.

Returns:

{"rms": …, "rms_dbfs": …, "peak": …, "peak_dbfs": …}.

Return type:

dict[str, float]

Raises:

ValueError – If source is not a microphone.

Examples

>>> import asyncio
>>> mic = ...
>>> asyncio.run(mic_level(mic))
capture_helper.new_scene(name='untitled', *, width=1280, height=720)[source]

Create an empty scene with a named canvas.

Parameters:

  • name (str, default "untitled") – Human name for the scene; also the suggested filename stem.

  • width (int) – Canvas size in pixels. Defaults to 1280x720.

  • height (int) – Canvas size in pixels. Defaults to 1280x720.

Returns:

A scene with no sources yet.

Return type:

Scene

Raises:

ValueError – If width or height is not a positive integer.

Examples

>>> s = new_scene("demo", width=640, height=360)
>>> s["width"], s["height"], s["sources"]
(640, 360, [])
capture_helper.pick_source(kind, *, name_substring=None, index=None)[source]

Pick a single capture device matching the given constraints.

Runs list_sources() for kind and applies each non-None filter as a hard predicate. Returns the first remaining candidate (OS-listing order — usually “built-in first, peripherals after”). Raises ValueError if nothing matches.

Parameters:

  • kind ("camera" | "microphone") – Which kind of device to pick.

  • name_substring (str, optional) – Case-insensitive substring filter against Source["name"]. Useful to disambiguate when several devices of the same kind are present ("BlackHole", "USB", "FaceTime", …).

  • index (int, optional) – Exact match against Source["index"]. When the OS reports stable indices (avfoundation, dshow), this picks a specific device unambiguously.

Returns:

The chosen device.

Return type:

Source

Raises:

ValueError – If no device matches (either the catalog is empty, or every candidate is filtered out by the constraints).

Examples

>>> from capture_helper import pick_source
>>> cam = pick_source("camera")                  # first available camera
>>> mic = pick_source("microphone", name_substring="BlackHole")
>>> usb_cam = pick_source("camera", index=1)
capture_helper.resolve_scene_sources(scene)[source]

Match every scene source against the current machine’s live devices.

A scene authored elsewhere (or before a device was plugged in) may not fully resolve here. Rather than raising on the first miss, this returns one ResolvedSceneSource per recipe so the GUI / CLI can report exactly which tiles are live and which are dangling.

Parameters:

scene (Scene) – The scene to resolve.

Returns:

One entry per source in scene["sources"], in the same order.

Return type:

list[ResolvedSceneSource]

Examples

>>> res = resolve_scene_sources(new_scene("empty"))
>>> res
[]
capture_helper.rms_dbfs(pcm)[source]

Compute linear RMS and its dBFS equivalent for a PCM block.

Parameters:

pcm (numpy.ndarray) – Float32 samples in [-1, 1] (any shape; flattened internally).

Returns:

(rms_linear, rms_dbfs). rms_dbfs is clamped at -120.0 for digital silence to avoid -inf.

Return type:

tuple[float, float]

Examples

>>> import numpy as np
>>> lin, db = rms_dbfs(np.zeros(100, dtype=np.float32))
>>> lin == 0.0 and db == -120.0
True
capture_helper.save_scene(scene, path)[source]

Serialise a scene to a JSON file on disk.

Parameters:

  • scene (Scene) – The scene to persist. Validated before writing.

  • path (str or Path) – Destination path. Parent directories are created if missing.

Returns:

The absolute path the scene was written to.

Return type:

str

Raises:

ValueError – If scene fails validate_scene().

Examples

>>> import tempfile, os
>>> p = os.path.join(tempfile.mkdtemp(), "demo.scene.json")
>>> _ = save_scene(new_scene("demo"), p)
>>> os.path.exists(p)
True
capture_helper.scene_from_available_devices(name='auto')[source]

Build a starter scene from whatever devices this machine reports.

Convenience used by the GUI’s “auto-populate” button and by the docs: grabs the first camera (full-canvas) and the first microphone (if any) so the user has something live to arrange instead of a blank canvas.

Parameters:

name (str, default "auto") – Name for the generated scene.

Returns:

A scene seeded with up to one camera and one microphone. Empty of sources when the host reports no devices (e.g. headless CI).

Return type:

Scene

Examples

>>> isinstance(scene_from_available_devices(), dict)
True
capture_helper.snapshot_jpeg(source, *, output_width=480, output_height=270, quality=5)[source]

Grab one live frame from a camera and return it as JPEG bytes.

Parameters:
Returns:

A single JPEG image.

Return type:

bytes

Raises:

  • ValueError – If source is not a camera.

  • RuntimeError – If no frame could be captured (permission denied / device busy).

Examples

>>> cam = ...
>>> jpg = snapshot_jpeg(cam)
capture_helper.validate_scene(scene)[source]

Validate an in-memory object as a well-formed Scene.

Cheap structural checks only — device availability is NOT verified here (that is resolve_scene_sources()’ job, and depends on the host). This guards the save/load boundary so a corrupt or hand-edited file surfaces a clear error instead of an obscure KeyError deep in the runner.

Parameters:

scene (Any) – Candidate object (typically freshly parsed JSON).

Returns:

The same object, once every invariant holds (returned for chaining).

Return type:

Scene

Raises:

ValueError – If any required key is missing or has the wrong type / value.

Examples

>>> validate_scene(new_scene("ok"))["name"]
'ok'