hearing.capture

Audio capture and channel handling for the hearing pipeline.

This module owns turning an audio source into per-channel float arrays, and the all-important channel split: mic and system audio are kept on separate channels so that “who is local” is known for free (see the hearing-diarization skill for what we do with that, and hearing-audio-capture for the macOS mechanics — BlackHole + Aggregate Device, or Core Audio taps).

The generic, channel-agnostic conditioning (decode, down-mix, resample, energy) now lives in scribed.audio and is re-exported here for backward compatibility — hearing owns only the multi-channel / “me vs them” concerns.

Functions

split_channels(data, *[, mic_channels, ...])

Split a multi-channel array into per-Channel mono arrays.

Classes

ChannelSplitFileCapture(path[, ...])

A batch CaptureSource over an audio file.

DeviceCapture([device, block_ms, ...])

Live macOS capture from an audio device (the milestone-2 capture source).

StreamingFileCapture(path[, block_ms, ...])

A streaming CaptureSource over a file.

class hearing.capture.ChannelSplitFileCapture(path, mic_channels=(0,), system_channels=(1,))[source]

Bases: object

A batch CaptureSource over an audio file.

Reads the file once and yields (channel, mono_frames) for each channel present. The live counterpart (device/tap) implements the same interface; swapping it in is the only change needed to go live — see the hearing-live-pipeline skill.

Typical use (needs a real audio file, so it is not run as a doctest):

ChannelSplitFileCapture("meeting.wav").frames()
async astream()[source]

Stream the file’s channels in one shot (whole clip per channel).

Provided so a file satisfies the async CaptureSource too; for a realistic block-by-block live simulation use StreamingFileCapture.

Return type:

AsyncIterator[tuple[Channel, ndarray]]

channels()[source]

The per-channel mono arrays at the file’s native sample rate.

Return type:

dict[Channel, ndarray]

frames()[source]

Yield (channel, mono_frames) once per channel (whole clip).

Return type:

Iterator[tuple[Channel, ndarray]]

class hearing.capture.DeviceCapture(device=None, block_ms=200, mic_channels=(0, 1), system_channels=(2, 3), sample_rate=48000, channels_count=4)[source]

Bases: object

Live macOS capture from an audio device (the milestone-2 capture source).

Point this at an Aggregate Device that combines your microphone and BlackHole (so mic and system audio land on separate, non-mixed channels — see the hearing-audio-capture skill for the Audio MIDI Setup). Reads multi-channel blocks and splits them into mic/system channels.

Requires pip install 'hearing[capture]' (sounddevice). Needs real audio hardware to exercise; constructing it is cheap (the device opens in astream). For a hardware-free run of the live loop, use StreamingFileCapture.

async astream()[source]

Open the device and yield (channel, block) blocks until cancelled.

Return type:

AsyncIterator[tuple[Channel, ndarray]]

static list_devices()[source]

Return available audio devices (helps pick the Aggregate Device).

class hearing.capture.StreamingFileCapture(path, block_ms=200, realtime=False, mic_channels=(0,), system_channels=(1,))[source]

Bases: object

A streaming CaptureSource over a file.

Yields (channel, block) in block_ms chunks, time-ordered across channels, so the live pipeline can be exercised end-to-end without a mic or BlackHole. With realtime=True it paces playback in real time (for demos); the default streams as fast as possible (for tests).

async astream()[source]

Yield (channel, block) blocks, all channels per time step.

Return type:

AsyncIterator[tuple[Channel, ndarray]]

channels()[source]

The per-channel mono arrays at the file’s native sample rate.

Return type:

dict[Channel, ndarray]

frames()[source]

Batch view: whole clip per channel.

Return type:

Iterator[tuple[Channel, ndarray]]

hearing.capture.split_channels(data, *, mic_channels=(0,), system_channels=(1,))[source]

Split a multi-channel array into per-Channel mono arrays.

An aggregate device does not mix — mic lands on one set of channels and system audio on another — so we can slice the columns apart. A mono file (or single-column) has no split; it is returned as {Channel.MIXED: ...}.

Parameters:
  • data (ndarray) – shape (n_samples,) or (n_samples, n_channels).

  • mic_channels (Sequence[int]) – column indices carrying the local microphone.

  • system_channels (Sequence[int]) – column indices carrying captured system audio.

Return type:

dict[Channel, ndarray]

Returns:

Mapping from Channel to a mono float32 array. Channels whose indices are out of range are silently omitted.