Crate pcmflux 

pcmflux: PulseAudio/PipeWire audio capture with Opus encoding, plus mic-uplink playback, exposed as a pure-Rust PyO3 extension (the audio sibling of pixelflux).

The capture thread pulls S16LE PCM fragments from a PulseAudio record stream, reassembles them into fixed-size Opus frames, encodes them (mono/stereo via the opus crate, 5.1/7.1 surround via the multistream API), optionally wraps them in RFC 2198 RED framing — redundant copies of recent frames, so a client on a lossy transport rebuilds a dropped packet from the next one it receives instead of stalling for a retransmit — and hands each frame to a delivery thread that runs the Python callback off the audio path, so a slow or GIL-blocked callback can never stall the PulseAudio pump. The playback path mirrors this in reverse: it decodes the Opus mic uplink and writes PCM into a virtual sink.

Concurrency design (the invariants below are load-bearing):

• A lifecycle mutex serializes joining/reassigning the capture thread.
• A single stop_state atomic is the one source of truth (0 = running, -1 = external stop, a positive value = self-stop recorded under the issuing thread’s tid — the delivery thread’s for a callback-issued stop). The external -1 is stored INSIDE that lock immediately before join, so a stop can never be lost between observing a live thread and asking it to stop. A re-entrant self-start undoes only its own self-stop via one compare-exchange, so a racing external stop is never clobbered.
• The PulseAudio mainloop is pumped with a bounded ~20ms timeout, so a stop is observed within ~20ms even if the audio source delivers no data (is wedged).
• The GIL is released around join, because joining the capture thread transitively joins the delivery thread, whose in-flight Python callback needs the GIL; holding it while joining would deadlock.
• A callback may itself call stop/start; the callback runs on the delivery thread, so that re-entrant case is detected via the delivery thread’s OS tid (the capture thread’s tid is checked too) and short-circuits without joining — a join from inside the callback would cycle (stopper joins capture, capture joins delivery, delivery is the stopper).

Macros

elog 🔒

Non-panicking eprintln! replacement — the stderr sibling of plog!.

plog 🔒

Non-panicking println! replacement that swallows write errors (e.g. EPIPE) instead of panicking, so a broken output pipe can’t unwind the capture thread or a callback.

Structs

AudioCapture 🔒

Python-facing capture handle. Owns the Shared lifecycle state and exposes start_capture / stop_capture / update_audio_bitrate / is_capturing to Python.

AudioCaptureSettings 🔒

Python-facing capture/encode configuration read by start_capture.

AudioFrame 🔒

Zero-copy buffer-protocol result type handed to the Python callback.

AudioPlayback 🔒

Python-facing mic-playback handle, symmetric to AudioCapture.

AudioPlaybackSettings 🔒

Python-facing mic-playback configuration read by AudioPlayback.start.

BufferPool 🔒

Recycles outgoing frame buffers from dropped AudioFrames back to the capture thread, so the steady-state emit path allocates nothing.

DeliveryRing 🔒

Bounded, drop-oldest hand-off from the capture thread to the Python delivery thread, so a slow or GIL-blocked callback can never stall the PulseAudio pump.

DeliveryThread 🔒

Owns the delivery thread for the lifetime of one capture run and tears it down on Drop, so teardown also happens when the capture thread UNWINDS.

Inner 🔒

Shared state for one capture (or playback) run: the lifecycle state machine plus the per-frame settings mirrors the worker reads on the hot path — all lock-free except the rarely touched last_error.

MultiOpus 🔒

Owning wrapper over a raw OpusMSEncoder (the surround multistream encoder).

OpusPlaybackDecoder 🔒

Turns the mic uplink back into PCM: the client always sends the mic as Opus, so every inbound packet must be decoded before it can be queued for the virtual sink. It decodes one packet to interleaved S16LE PCM, reusing a scratch buffer across calls to stay off the per-decode allocation path. Lives behind a Mutex on PbShared and is driven from write / write_red.

PaCaptureSession 🔒

Own one whole capture run end to end on a dedicated thread — connect PulseAudio, encode, and deliver — until stopped. It runs on its own thread because the PulseAudio mainloop must be pumped continuously and independently of Python: sharing the caller’s thread would tie capture cadence to the GIL and let any Python stall starve the audio. The body handed to spawn_worker.

PaPlaybackSession 🔒

One PulseAudio session for playback: mainloop, context, and the playback stream, all recreated together on reconnect — the mirror of PaCaptureSession, with the same drop-order requirement (stream first, then its context, then the mainloop).

PbSettings 🔒

Playback settings, snapshotted from the Python AudioPlaybackSettings at start so the playback thread owns an immutable copy for the run’s lifetime.

PbShared 🔒

Per-AudioPlayback shared handle — the mirror of Shared for the playback path.

PlayQueue 🔒

Bounded, drop-oldest byte queue for the mic-PCM handoff into the virtual “input” sink — the whole of the playback path’s buffering in a single bound.

PoolTaker 🔒

Sole-consumer view over the shared BufferPool for the capture thread.

RunState 🔒

Per-run encode/deliver state, living on the capture thread’s stack for the lifetime of one capture. Holds the encoder, the frame-reassembly buffers, the outgoing buffer recycler, the RED redundancy history, and the running debug-log counters.

Settings 🔒

Capture/encode settings, snapshotted from AudioCaptureSettings at start.

Shared 🔒

Per-AudioCapture shared handle: the lock-free Inner state plus the lifecycle-locked join handle for the capture thread.

Enums

PcmEncoder 🔒

One encode surface over both Opus APIs: the opus crate for mono/stereo, and the raw multistream C API for 6/8-channel surround.

SessionOpenError 🔒

Why a session failed to open: drives the retry policy of the caller.

Constants

MAX_OPUS_PACKET 🔒

Max Opus output packet size in bytes per stream; sizes the emit buffer pool.

OPUS_BITRATE_MAX 🔒

OPUS_BITRATE_MIN 🔒

Opus target-bitrate bounds (bits/s); the initial opus_bitrate and every live update are clamped into this range so the encoder never sees a value it would reject.

PUMP_TIMEOUT_US 🔒

PulseAudio mainloop pump timeout (~20 ms) — upper bound on how long a pending stop can go unobserved even when the audio source delivers nothing.

RED_BLOCK_PT 🔒

RED block payload type.

RED_MAX_DISTANCE 🔒

Max redundant copies per frame and RED history depth.

RED_MAX_LEN 🔒

Max RED block length — 10-bit ceiling in bytes.

RED_MAX_OFFSET 🔒

Max RED timestamp offset — 14-bit ceiling in 48 kHz samples from the primary.

RED_PREFIX_MAX 🔒

Worst-case byte length of the WS frame prefix (tag + n_red + pts + headers + payloads).

STOP_EXTERNAL 🔒

stop_state sentinel: authoritative external stop (-1). Any positive value instead is the OS tid of a capture thread that self-stopped from its callback.

STOP_NONE 🔒

stop_state sentinel: no stop pending (running).

ST_FAILED 🔒

start_state: the last run ended in error (startup or mid-run); last_error says why.

ST_IDLE 🔒

start_state: no worker has run yet, or the last run stopped cleanly.

ST_RUNNING 🔒

start_state handshake: the hot loop is running.

ST_STARTING 🔒

start_state handshake: startup in progress.

Statics

PLAYBACK_REGISTRY 🔒

Process-wide registry of live playbacks, swept by the same atexit sweep as captures. Holds Weak references so it keeps nothing alive on its own.

REGISTRY 🔒

Process-wide registry of live captures, swept at interpreter exit. Holds Weak references so it keeps nothing alive on its own.

Functions

__pyfunction__stop_all_captures 🔒 ^⚠

_stop_all_captures 🔒

atexit sweep: stop and join every live capture and playback before interpreter shutdown, so no worker thread is still calling into Python during finalization.

await_start 🔒

Startup handshake shared by the capture and playback starts: with the GIL released, waits up to ~2 s for the worker this call spawned (spawned) to publish RUNNING or FAILED. A worker still STARTING when the window closes counts as started — its retry ladder legitimately runs longer than this — and a failure after that is exposed through state / last_error instead. On FAILED, join_failed_start tears down only that thread (identity-checked, sparing a concurrent winner) and the worker’s recorded error is raised as RuntimeError.

capture_run 🔒

check_opus_sample_rate 🔒

Opus codecs only run at 8, 12, 16, 24 or 48 kHz; anything else would fail encoder or decoder creation on the worker, so it is rejected up front.

extract_pb_settings 🔒

Read a Python AudioPlaybackSettings into a Rust PbSettings by attribute name, rejecting with ValueError what the playback thread could only fail on later.

extract_settings 🔒

Read a Python AudioCaptureSettings into a Rust Settings by attribute name, rejecting with ValueError what the worker could only fail on later.

gettid 🔒

Returns the calling thread’s OS tid (gettid syscall) so a stop/start issued from inside the Python callback can detect it is on the capture thread and avoid self-joining.

join_failed_start 🔒

Failed-start teardown: stop and join the thread spawned by THIS start attempt, but only if it still owns the slot.

multiopus_layout 🔒

Chromium’s multistream-Opus surround layout for a channel count, as (streams, coupled, mapping); None for anything but 5.1 (6) or 7.1 (8).

pa_capture_session_open 🔒

Open a capture session: mainloop + context + record stream driven to Ready on the bounded pump, honoring stop_pending at every turn. A NAMED device is validated by an introspect probe on every call (an async connect_record would not fail synchronously on a bad name, and on reconnect this is also what notices the device reappearing after an outage). device_was_present distinguishes initial bring-up from reconnect: a named device missing at startup is probably a misconfiguration, so it is reported as DeviceNotFound and the caller spends only a short window on it; the same device vanishing mid-run (PulseAudio/PipeWire restart kills every source) is transient and must be retried or audio never comes back.

pa_playback_session_open 🔒

Open a playback session: mainloop + context + playback stream driven to Ready on the bounded pump, honoring stop_pending at every turn. The mirror of pa_capture_session_open.

parse_device_name 🔒

Normalize a Python device_name (str | bytes | None) into Option<String>, mapping both None and the empty string to None (meaning the system default). Shared by the capture and playback settings extractors.

pcmflux 🔒

PyO3 module init: register the capture/playback classes and the atexit sweep.

playback_registry 🔒

Lazily initialize and return the playback registry.

playback_run 🔒

Drive one whole mic-playback run on the playback thread. The body handed to spawn_worker; the mirror of capture_run for the uplink.

pump 🔒

Run one bounded iteration of a PulseAudio standard mainloop: prepare(timeout_us) → poll → dispatch. Returns false if any stage errors.

registry 🔒

Lazily initialize and return the capture registry.

spawn_worker 🔒

Locked takeover + spawn of a worker thread, shared by the capture and playback starts. Returns the new thread’s id, or None if the spawn failed.

valid_opus_duration 🔒

True if ms is a valid Opus frame duration (2.5, 5, 10, 20, 40, or 60 ms).

value_error 🔒

Build a ValueError for a rejected settings field.

write_ws_prefix_into 🔒

Emit the RFC 2198 RED framing prefix into buf in-place and return its length, so the encoder can serialize the Opus packet directly after it with no scratch buffer and no per-frame allocation. The runtime counterpart of build_ws_body.

Type Aliases

FrameQueue 🔒

Backing store for the delivery ring: Some(queue) while open, None once closed so pop wakes and returns None for a clean shutdown.