Audio Capture Sources

A capture source owns an audio input device independently of any voice session, so that a consumer which must listen before a session exists — a wake word, a level meter — no longer has to hand the device over at the moment the person starts speaking.

Normative reference: RFC Section 12.12. Narrative guide: Shared Microphone Capture.

Quick start

from roomkit.voice.capture import LocalMicSource
from roomkit.voice.backends.local import LocalAudioBackend

mic = LocalMicSource(sample_rate=24000, backlog_seconds=10)
mic.start()

# Listen with no session in sight. Enqueue only — never block the callback.
detector = mic.subscribe(enqueue, name="wakeword")

# The backend becomes a subscriber rather than the device's owner.
transport = LocalAudioBackend(source=mic)

mark = mic.mark()                       # at SPEECH_START
await channel.start_session(            # once the trigger matched
    room_id, participant_id, connection=None,
    metadata={"capture_since": mark},
)

Install the device implementation with pip install roomkit[local-audio].

See the full example for a complete runnable script.

Source ABC

AudioCaptureSource

AudioCaptureSource(*, sample_rate=16000, channels=1, sample_width=2, block_duration_ms=20, backlog_seconds=DEFAULT_BACKLOG_SECONDS, max_backlog_bytes=None)

Bases: ABC

A continuous source of AudioFrame shared by several consumers.

Lifetime is explicit: start() and stop() are the only things that acquire and release the device. Dropping to zero subscribers does not stop capture, and gaining one does not start it — which is what makes it safe for a wake-word detector to detach for the duration of a session.

Frames are emitted raw. Echo cancellation is per-session (it needs the reference signal of what that session plays) and belongs downstream, in the backend, after fan-out. A subscriber still attached while a session is playing hears the far end unattenuated. Resampling is likewise the subscriber's concern: a source has one format.

sample_rate property

sample_rate

Sample rate of every frame this source emits.

channels property

channels

Channel count of every frame this source emits.

sample_width property

sample_width

Bytes per sample of every frame this source emits.

block_duration_ms property

block_duration_ms

Nominal duration of one captured block, in milliseconds.

input_latency_ms property

input_latency_ms

Device input latency, when the implementation can report one.

Used to seed AEC stream delay. None means unknown.

start abstractmethod

start()

Acquire the device and begin capturing. Idempotent.

stop abstractmethod

stop()

Release the device and stop capturing. Idempotent.

close

close()

Stop capturing and drop every subscriber. Idempotent.

mark

mark()

Name the current position in the backlog, for later replay.

subscribe

subscribe(callback, *, since=None, name=None)

Attach a consumer, optionally replaying the backlog from since.

The callback runs synchronously on the capture thread. It MUST NOT perform unbounded work: enqueue the frame and return. The source guarantees no isolation between subscribers — one slow subscriber degrades capture for all of them.

Ordering is total: every replayed frame is delivered before every live one, including frames captured during the replay itself.

Parameters:

Name	Type	Description	Default
callback	CaptureFrameCallback	Invoked once per frame.	required
since	CaptureMark | None	Replay from this mark before going live. A mark whose position has been evicted replays what remains and sets truncated on the returned subscription.	None
name	str | None	Diagnostic label used in logs and slow-subscriber warnings.	None

Returns:

Type	Description
CaptureSubscription	The subscription handle.

Raises:

Type	Description
ValueError	If since was issued by a different source.

Marks and subscriptions

CaptureMark dataclass

CaptureMark(sequence, source_id)

A position in a source's backlog.

Opaque: the fields are an implementation detail and MUST NOT be interpreted by callers. Obtain one from :meth:AudioCaptureSource.mark and hand it back to :meth:AudioCaptureSource.subscribe.

sequence instance-attribute

sequence

Index of the next frame the source will capture.

source_id instance-attribute

source_id

Identity of the issuing source, so a foreign mark is rejected.

CaptureSubscription

CaptureSubscription(source, subscriber, name)

Handle on one subscriber's attachment to a capture source.

name instance-attribute

name = name

Diagnostic label, as passed to subscribe().

replayed_bytes instance-attribute

replayed_bytes = 0

Bytes delivered from the backlog — audio captured before subscribing.

truncated instance-attribute

truncated = False

True when the mark had already been evicted and replay is partial.

active property

active

False once :meth:unsubscribe has been called.

unsubscribe

unsubscribe()

Detach. Idempotent, and never stops the source itself.

Implementations

LocalMicSource

LocalMicSource(*, input_device=None, sample_rate=16000, channels=1, block_duration_ms=20, backlog_seconds=DEFAULT_BACKLOG_SECONDS, max_backlog_bytes=None)

Bases: AudioCaptureSource

Capture source owning the system microphone.

input_latency_ms property

input_latency_ms

PortAudio's reported input latency, once the stream is open.

MockCaptureSource

MockCaptureSource(*, sample_rate=16000, channels=1, block_duration_ms=20, backlog_seconds=DEFAULT_BACKLOG_SECONDS, max_backlog_bytes=None)

Bases: AudioCaptureSource

Capture source driven by explicit feed() calls.

Frames are dispatched on the calling thread, which stands in for the capture thread. Tests that need to exercise the catch-up path can feed from a second thread while a subscriber is replaying.

started instance-attribute

started = False

True between start() and stop().

start_count instance-attribute

start_count = 0

How many times start() actually acquired the device.

block_bytes property

block_bytes

Byte length of one nominal block at this source's format.

feed

feed(data)

Dispatch one frame, as a capture device would.

feed_blocks

feed_blocks(count, *, fill=0)

Dispatch count blocks of identical filler, returning what was sent.

Each block carries a distinct byte value derived from its index, so tests can assert on exact ordering rather than on totals alone.

The subscriber contract

Fan-out is synchronous on the capture thread — that is what keeps echo cancellation's capture/reference timing in step. A subscriber must not perform unbounded work in the callback: enqueue the frame and return. The source guarantees no isolation between subscribers, times each callback, and warns by name when one runs long.

Frames are emitted raw, pre-AEC: echo cancellation is per-session and applied downstream in the backend. A subscriber left attached during a call hears the far end unattenuated.