Standard ASRstandard_asr.audio.wire
PCM wire codec for streaming: the canonical pcm_s16le encoding, and helpers for
encoding/decoding between float32 waveforms and little-endian 16-bit PCM bytes.
Canonical streaming wire encoding and PCM codec.
This module is the single source of truth for the float32 <-> 16-bit PCM
conversion the streaming wire protocol is pinned to. The conversion is byte-pinned
so that every language's wire implementation produces identical PCM and a
cross-language conformance test sees no +-1 LSB noise; defining it once here
(rather than letting each engine re-derive it) is what keeps the Python and wire
layers isomorphic across languages.
Public surface:
CANONICAL_WIRE_ENCODING-- the canonical encoding identifier ("pcm_s16le"), so applications and plugins stop hardcoding the string and risking a silent mismatch against an engine'swire_encodings.pcm16_encode-- float waveform -> canonicalpcm_s16lebytes.pcm16_decode-- canonicalpcm_s16lebytes -> float32 waveform.require_float_waveform/to_int16_pcm-- the lower-level canonical-quantization primitives (a dtype guard + the round-half float->int16 conversion with a non-finite count), shared bypcm16_encodeand the WAV encoders so the spec-pinned quantization is defined exactly once.
The encode/decode pair deliberately uses the spec's asymmetric scale factors
(x 32767 on encode, / 32768 on decode); the ~-0.00027 dB round-trip
attenuation is intentional and identical across the WAV encoder, the WAV reader,
and this codec because all three call the helpers defined here.
CANONICAL_WIRE_ENCODING#attributeSource
CANONICAL_WIRE_ENCODING: Final = 'pcm_s16le'pcm16_decode#functionSource
def pcm16_decode(data: bytes) -> NDArray[np.float32]Decode canonical pcm_s16le bytes to a float32 waveform in [-1, 1].
The inverse of pcm16_encode: read the bytes as little-endian signed
16-bit codes and scale by / 32768 (the deliberate 32767/32768
round-trip asymmetry the spec pins, shared with the WAV reader). An empty
data decodes to an empty array. Frames whose length is not a whole number
of 16-bit samples are rejected loudly rather than silently dropping a byte.
| Name | Type | Description |
|---|---|---|
data | bytes | Little-endian signed 16-bit PCM bytes (interleaved if multi-channel; de-interleaving is the caller's responsibility). |
NDArray[np.float32]The decoded
float32waveform.
AudioProcessingErrorIf
len(data)is not a multiple of 2 bytes.
pcm16_encode#functionSource
def pcm16_encode(samples: NDArray[np.floating]) -> bytesEncode a float waveform in [-1, 1] to canonical pcm_s16le bytes.
The canonical streaming wire encoding: clip to [-1, 1],
quantize with round-half (x 32767), serialize little-endian. Non-finite
samples are sanitized (NaN->0, +-Inf->+-full-scale) so the cast never emits
garbage. An empty array encodes to b"". Use this in a streaming engine
instead of re-implementing the quantization (astype(np.int16) truncation
silently doubles the quantization error and diverges from the wire contract).
| Name | Type | Description |
|---|---|---|
samples | NDArray[np.floating] | Float waveform (an amplitude in roughly |
bytesThe little-endian signed 16-bit PCM byte payload.
AudioProcessingErrorIf
samplesis not a floating-point dtype.
require_float_waveform#functionSource
def require_float_waveform(audio: NDArray[np.floating]) -> NDArray[np.floating]Reject a non-floating array before it is treated as a [-1, 1] waveform.
The public encoders are typed NDArray[np.floating], but np.asarray
does not enforce dtype at runtime, so a dynamic / un-type-checked caller can
pass an integer PCM array. np.clip(int_codes, -1, 1) would crush every
sample to a square wave (for example, 1000 -> 1.0), encoding completely corrupted
PCM that still 'succeeds' -- a silent wrong result. The engine input boundary
(AudioArray.__post_init__) already rejects non-floating dtypes; this gives
the standalone codec/encoder helpers the same guard. The conversion is
audio.astype(np.float32) / 32768.0 for int16 codes (mirrors the decode
scaling).
| Name | Type | Description |
|---|---|---|
audio | NDArray[np.floating] | The array to validate. |
NDArray[np.floating]The array unchanged, narrowed to a floating dtype.
AudioProcessingErrorIf
audiois not a floating-point dtype.
to_int16_pcm#functionSource
def to_int16_pcm(audio: NDArray[np.floating]) -> tuple[NDArray[np.int16], int]Convert a float waveform in [-1, 1] to signed 16-bit PCM.
Non-finite samples are sanitized first (NaN->0, +Inf->+1, -Inf->-1),
because np.clip does NOT replace NaN (np.clip(nan, -1, 1) == nan) and
casting NaN to int16 is undefined behavior that emits garbage PCM -- a
silent-wrong-result. This mirrors the decode paths, which already sanitize.
The count of replaced samples is returned so the caller can emit a
non_finite_audio diagnostic (the sanitize is correct and necessary, but
the fact that it happened MUST be visible to the caller -- explicit over
implicit). Clipping then happens before the cast (NumPy 1.x/2.x
defensive), and quantization uses
round-half (np.rint) rather than truncation so the canonical encoder's
quantization error stays bounded by 0.5 LSB instead of 1 LSB.
| Name | Type | Description |
|---|---|---|
audio | NDArray[np.floating] | Float waveform array. |
NDArray[np.int16]A
(pcm, sanitized_non_finite)pair: the signed 16-bit PCM array andintthe number of non-finite samples that were replaced.