Live audio

kudio.core.stream

Audio streaming: playback, local recording, and remote (TCP) streams.

PyAudio is an optional dependency (pip install kudio[audio]); it is only imported when a stream/recorder is actually created, so the package can be used for offline processing without any audio hardware.

Classes:

Name	Description
Recorder	Blocking recorder for a fixed duration.
RemoteStreamReader	Receive an audio stream over TCP and feed it into an
LocalStreamReader	Stream from a local microphone into an AudioBuffer, with
StreamRecorder	Record for as long as you like, and watch the level while it happens.

Functions:

Name	Description
record	Blocking record for seconds via sounddevice (modern backend).
wave_decode	Raw int16 bytes -> array of shape (chunk_length, channels).
wave_encode	Multi-channel array -> interleaved int16 bytes.
play_audio	Play audio from a file path, an opened wave.Wave_read, or an ndarray.

Attributes:

Name	Type	Description
PA_FLOAT32		PortAudio's code for 32-bit float samples (pyaudio.paFloat32), for
PA_INT16		PortAudio's code for 16-bit integer samples (pyaudio.paInt16), the

PA_FLOAT32 module-attribute

PA_FLOAT32 = 1

PortAudio's code for 32-bit float samples (pyaudio.paFloat32), for LocalStreamReader(wav_format=...).

PA_INT16 module-attribute

PA_INT16 = 8

PortAudio's code for 16-bit integer samples (pyaudio.paInt16), the default wav_format.

Recorder

Recorder(
    device: Optional[int] = None,
    record_times: int = 3,
    rate: int = 16000,
    channels: int = 1,
    frames_per_buffer: int = 512,
)

Blocking recorder for a fixed duration.

Examples

>>> with Recorder(device=11, record_times=3, channels=6, rate=16000) as r:
...     wave_data = r.start_record()
...     r.save('kudio_test.wav')

Methods:

Name	Description
start_record	Record for record_times seconds; returns the waveform, with one
save	Write the last recording to file_dir (16-bit PCM).

Source code in kudio/core/stream.py

def __init__(self, device: Optional[int] = None, record_times: int = 3,
             rate: int = 16000, channels: int = 1,
             frames_per_buffer: int = 512):
    pyaudio = _pyaudio()
    self.format = pyaudio.paInt16
    self.input_device_index = device
    self.record_times = record_times
    self.rate = rate
    self.channels = channels
    self.frames_per_buffer = frames_per_buffer
    self.p = pyaudio.PyAudio()
    self.stream = None
    self.frames: Optional[np.ndarray] = None  # interleaved int16 samples

start_record

start_record() -> ndarray

Record for record_times seconds; returns the waveform, with one column per channel (squeezed for mono).

Source code in kudio/core/stream.py

def start_record(self) -> np.ndarray:
    """Record for ``record_times`` seconds; returns the waveform, with one
    column per channel (squeezed for mono)."""
    self.stream = self.p.open(
        format=self.format,
        channels=self.channels,
        input_device_index=self.input_device_index,
        rate=self.rate,
        input=True,
        frames_per_buffer=self.frames_per_buffer)

    log.info("Recording %d second(s)...", self.record_times)
    chunks = []
    n_reads = int(self.rate / self.frames_per_buffer * self.record_times)
    for _ in range(n_reads):
        rec = self.stream.read(self.frames_per_buffer)
        chunks.append(np.frombuffer(rec, dtype=np.int16))
    self.frames = np.concatenate(chunks) if chunks else np.empty(0, np.int16)
    per_channel = [self.frames[c::self.channels] for c in range(self.channels)]
    return np.array(per_channel).T.squeeze()

save

save(
    file_dir: PathLike,
    rename: bool = False,
    resample: Optional[int] = None,
) -> bool

Write the last recording to file_dir (16-bit PCM).

Source code in kudio/core/stream.py

def save(self, file_dir: PathLike, rename: bool = False,
         resample: Optional[int] = None) -> bool:
    """Write the last recording to *file_dir* (16-bit PCM)."""
    if self.frames is None:
        log.error("Recorder: nothing recorded yet")
        return False
    file_dir = check_file(file_dir, rename=rename)
    try:
        data = self.frames
        rate = self.rate
        if resample is not None and resample != self.rate:
            import scipy.signal
            # from the frames actually captured: the take is a whole
            # number of buffers, so it is never quite record_times long,
            # and stretching it to that length moved every pitch
            frames = len(data) // self.channels
            target = int(round(frames * resample / self.rate))
            data = scipy.signal.resample(
                data.reshape(frames, self.channels), target, axis=0).reshape(-1)
            data = np.clip(np.round(data), np.iinfo(np.int16).min,
                           np.iinfo(np.int16).max).astype(np.int16)
            rate = resample
        with wave.open(file_dir, 'wb') as wf:
            wf.setnchannels(self.channels)
            wf.setsampwidth(2)  # int16
            wf.setframerate(rate)
            wf.writeframes(data.astype(np.int16).tobytes())
        log.info("Recorder: saved -> %s", file_dir)
        return True
    except Exception as e:
        log.error("Recorder: save failed: %s", e)
        return False

RemoteStreamReader

RemoteStreamReader(buffer: Optional[AudioBuffer])

Bases: Thread

Receive an audio stream over TCP and feed it into an AudioBuffer (optionally playing it back locally).

Methods:

Name	Description
sock_send	Send a text command; returns decoded waveform when the reply is

Source code in kudio/core/stream.py

def __init__(self, buffer: Optional[AudioBuffer]):
    super().__init__(daemon=True)
    self.audio_buffer = buffer
    self.frame_size_: Optional[int] = None
    self.receive_buffer_size_: Optional[int] = None

    self.p = None
    self.isReceiving = threading.Event()
    self.isPlaying = threading.Event()

    self.play_stream = None
    self.stream_dtype: Optional[int] = None
    self.data_type = None

    self.sk: Optional[socket.socket] = None
    self.break_count = 10

sock_send

sock_send(
    input_data: str, recv_buffer_size: int
) -> Optional[ndarray]

Send a text command; returns decoded waveform when the reply is binary audio data.

Source code in kudio/core/stream.py

def sock_send(self, input_data: str, recv_buffer_size: int
              ) -> Optional[np.ndarray]:
    """Send a text command; returns decoded waveform when the reply is
    binary audio data."""
    if input_data == '':
        log.error("Stream: input data error")
        return None
    self.sk.send(input_data.encode("utf-8"))
    data = self.sk.recv(recv_buffer_size)
    try:
        text = data.decode('utf-8')
    except UnicodeDecodeError:
        text = None
    # printable text is a reply; anything else is audio. A silent reply,
    # all zero bytes, decodes as UTF-8 too, and used to be taken for text
    if text is not None and text.strip() and \
            all(c.isprintable() or c.isspace() for c in text):
        log.info("Stream: %s", text)
        return None
    while len(data) < recv_buffer_size:
        data += self.sk.recv(recv_buffer_size - len(data))
    waves = np.frombuffer(data, dtype=np.int16)
    log.info("Stream: wave shape %s", waves.shape)
    return waves

LocalStreamReader

LocalStreamReader(
    device: Optional[int] = None,
    rate: Optional[int] = None,
    channels: int = 1,
    frame_size: int = 2048,
    wav_format: int = PA_INT16,
    buffer: Optional[AudioBuffer] = None,
    on_frame: Optional[Callable[[ndarray], None]] = None,
)

Stream from a local microphone into an AudioBuffer, with optional live monitoring through the default output device.

>>> lsr = LocalStreamReader(channels=1, frame_size=2048, wav_format=PA_INT16)
>>> if lsr.get_status():
...     lsr.play_audio()     # optional monitoring
...     lsr.stream_start()

Parameters:

Name	Type	Description	Default
device	Optional[int]	PyAudio input index. Omitted, the last device that opens as an input (kudio.CheckDevice.input_device).	None
rate	Optional[int]	sample rate, a request. Omitted or refused, the first of 16000, 22050 and 44100 the device accepts, else its own default; get_setup says what was opened.	None
channels	int	channel count, default 1	1
frame_size	int	frames per buffer, default 2048	2048
wav_format	int	PA_FLOAT32 (1) or PA_INT16 (8)	PA_INT16
buffer	Optional[AudioBuffer]	target AudioBuffer (optional)	None
on_frame	Optional[Callable[[ndarray], None]]	called with each captured chunk. An AudioBuffer drops frames once it is full, which is correct for a monitor and wrong for a recording; a callback lets the caller decide. It runs on PortAudio's thread, so it must return promptly — append, do not process.	None

Methods:

Name	Description
play_audio	Open a playback stream that echoes the microphone input.

Source code in kudio/core/stream.py

def __init__(self,
             device: Optional[int] = None,
             rate: Optional[int] = None,
             channels: int = 1,
             frame_size: int = 2048,
             wav_format: int = PA_INT16,
             buffer: Optional[AudioBuffer] = None,
             on_frame: Optional[Callable[[np.ndarray], None]] = None):
    """
    :param device: **PyAudio** input index. Omitted, the last device
        that opens as an input (:meth:`kudio.CheckDevice.input_device`).
    :param rate: sample rate, a request. Omitted or refused, the first of
        16000, 22050 and 44100 the device accepts, else its own default;
        :meth:`get_setup` says what was opened.
    :param channels: channel count, default 1
    :param frame_size: frames per buffer, default 2048
    :param wav_format: ``PA_FLOAT32`` (1) or ``PA_INT16`` (8)
    :param buffer: target AudioBuffer (optional)
    :param on_frame: called with each captured chunk. An
        :class:`AudioBuffer` **drops** frames once it is full, which is
        correct for a monitor and wrong for a recording; a callback lets
        the caller decide. It runs on PortAudio's thread, so it must
        return promptly — append, do not process.
    """
    self.audio_buffer = buffer
    self.on_frame = on_frame
    self.device_: Optional[int] = None
    self.rate_: Optional[int] = None
    self.channels_: Optional[int] = None
    self.frame_size_: Optional[int] = None
    self.receive_buffer_size_: Optional[int] = None

    self.isInitial = threading.Event()
    self.isBroadcasting = threading.Event()
    self.isListening = threading.Event()
    self.play_stream = None
    self.rec_stream = None
    self.p = None

    self.set_stream(device, rate, channels, frame_size, wav_format)

play_audio

play_audio() -> None

Open a playback stream that echoes the microphone input.

Source code in kudio/core/stream.py

def play_audio(self) -> None:
    """Open a playback stream that echoes the microphone input."""
    self._require_device("play_audio")
    self.play_stream = self.p.open(
        format=self.format,
        channels=self.channels_,
        rate=self.rate_,
        output=True,
        frames_per_buffer=self.frame_size_)
    if self.play_stream.is_active():
        self.isBroadcasting.set()
        log.info("Stream: broadcasting started")
    else:
        log.error("Stream: broadcasting failed")

StreamRecorder

StreamRecorder(
    device: Optional[int] = None,
    sr: Optional[int] = None,
    channels: int = 1,
    frame_size: int = 1024,
    max_seconds: Optional[float] = 3600.0,
    monitor: bool = False,
)

Record for as long as you like, and watch the level while it happens.

kudio.record is one blocking call: you commit to a duration up front and nothing is observable until it returns. That is fine for a script and useless for a person at a microphone, who needs to see the level before committing and stop when the sentence ends.

>>> rec = kudio.StreamRecorder(sr=16000)
>>> rec.start()
>>> rec.level_db()
-23.4
>>> y = rec.stop()                     # everything captured, float32

Capture runs on PortAudio's own thread; level, tail and elapsed are safe to poll from a UI timer. Nothing is dropped — unlike AudioBuffer, which is a monitor and discards when full.

Requires kudio[audio] (PyAudio).

Parameters:

Name	Type	Description	Default
device	Optional[int]	PyAudio input index (kudio.CheckDevice), not a sounddevice one — the two number devices independently.	None
sr	Optional[int]	sample rate. Omitted, the first of 16000, 22050 and 44100 the device accepts, and its own default only when none of them opens.	None
max_seconds	Optional[float]	hard cap on the take, so a recorder left running cannot fill memory. None removes the cap.	3600.0
monitor	bool	echo the input to the default output while recording.	False

Methods:

Name	Description
start	Open the stream and begin capturing. Idempotent.
stop	Close the stream and return everything captured.
take	Everything captured so far, without stopping.
level	Peak of the most recent chunk, linear 0..1.
level_db	The same reading in dBFS; -inf for silence.
elapsed	Seconds captured so far.
tail	The most recent seconds of audio, for a rolling display.
drain	Audio captured since the last call, and never twice.
reset	Throw the take away and start counting again, stream untouched.

Attributes:

Name	Type	Description
sr	int	The rate actually negotiated with the device.
capped	bool	True once max_seconds was reached and capture stopped growing.

Source code in kudio/core/stream.py

def __init__(self, device: Optional[int] = None, sr: Optional[int] = None,
             channels: int = 1, frame_size: int = 1024,
             max_seconds: Optional[float] = 3600.0,
             monitor: bool = False):
    """
    :param device: **PyAudio** input index (:class:`kudio.CheckDevice`),
        not a sounddevice one — the two number devices independently.
    :param sr: sample rate. Omitted, the first of 16000, 22050 and 44100
        the device accepts, and its own default only when none of them
        opens.
    :param max_seconds: hard cap on the take, so a recorder left running
        cannot fill memory. ``None`` removes the cap.
    :param monitor: echo the input to the default output while recording.
    """
    self._chunks: List[np.ndarray] = []
    self._lock = threading.Lock()
    self._level = 0.0
    self._frames = 0
    self._drained = 0                     # chunks already handed to drain()
    self._capped = False
    self._channels = max(1, int(channels))
    self._max_seconds = max_seconds
    self._monitor = monitor

    self._stream_args = (device, sr, self._channels, frame_size, PA_FLOAT32)
    self._reader = LocalStreamReader(
        device=device, rate=sr, channels=self._channels,
        frame_size=frame_size, wav_format=PA_FLOAT32,
        on_frame=self._on_frame)
    if not self._reader.get_status():
        from kudio.exceptions import DeviceError
        raise DeviceError("no usable input device for StreamRecorder")
    self._sr = int(self._reader.rate_ or sr or 16000)
    self._channels = int(self._reader.channels_ or self._channels)

sr property

sr: int

The rate actually negotiated with the device.

capped property

capped: bool

True once max_seconds was reached and capture stopped growing.

start

start() -> 'StreamRecorder'

Open the stream and begin capturing. Idempotent.

Source code in kudio/core/stream.py

def start(self) -> "StreamRecorder":
    """Open the stream and begin capturing. Idempotent."""
    if self.recording:
        return self
    if not self._reader.get_status():
        # stop() closes PortAudio; opening it again is what lets one
        # recorder take a second take instead of failing on a dead handle
        if not self._reader.set_stream(*self._stream_args):
            raise DeviceError("no input device could be opened")
        sr = int(self._reader.rate_ or self._sr)
        channels = int(self._reader.channels_ or self._channels)
        with self._lock:
            carried = bool(self._chunks)
        if carried and (sr, channels) != (self._sr, self._channels):
            # the take would hold two rates or two layouts end to end
            self._reader.terminate()
            raise DeviceError(
                f"the device reopened at {sr} Hz, {channels} channel(s), "
                f"and the take so far is {self._sr} Hz, {self._channels}; "
                f"stop() and reset() before starting again")
        self._sr, self._channels = sr, channels
    if self._monitor:
        self._reader.play_audio()
    self._reader.stream_start()
    return self

stop

stop() -> ndarray

Close the stream and return everything captured.

Mono comes back shaped (n,) and multi-channel (n, channels) — the layout kudio.file_load and kudio.save_wave use. Calling this twice returns the same take rather than an empty array.

Source code in kudio/core/stream.py

def stop(self) -> np.ndarray:
    """Close the stream and return everything captured.

    Mono comes back shaped ``(n,)`` and multi-channel ``(n, channels)`` —
    the layout :func:`kudio.file_load` and :func:`kudio.save_wave` use.
    Calling this twice returns the same take rather than an empty array.
    """
    if self.recording:
        self._reader.terminate()
    return self.take()

take

take() -> ndarray

Everything captured so far, without stopping.

Source code in kudio/core/stream.py

def take(self) -> np.ndarray:
    """Everything captured so far, without stopping."""
    with self._lock:
        chunks = list(self._chunks)
    if not chunks:
        return np.zeros(0, dtype=np.float32)
    flat = np.concatenate(chunks).astype(np.float32, copy=False)
    if self._channels > 1:
        usable = (flat.size // self._channels) * self._channels
        return flat[:usable].reshape(-1, self._channels)
    return flat

level

level() -> float

Peak of the most recent chunk, linear 0..1.

Source code in kudio/core/stream.py

def level(self) -> float:
    """Peak of the most recent chunk, linear ``0..1``."""
    with self._lock:
        return self._level

level_db

level_db() -> float

The same reading in dBFS; -inf for silence.

Source code in kudio/core/stream.py

def level_db(self) -> float:
    """The same reading in dBFS; ``-inf`` for silence."""
    peak = self.level()
    return 20.0 * float(np.log10(peak)) if peak > 0 else float('-inf')

elapsed

elapsed() -> float

Seconds captured so far.

Source code in kudio/core/stream.py

def elapsed(self) -> float:
    """Seconds captured so far."""
    with self._lock:
        return self._frames / self._sr if self._sr else 0.0

tail

tail(seconds: float) -> ndarray

The most recent seconds of audio, for a rolling display.

Cheap enough to call from a repeating timer: it copies only the window asked for, not the whole take.

Source code in kudio/core/stream.py

def tail(self, seconds: float) -> np.ndarray:
    """The most recent *seconds* of audio, for a rolling display.

    Cheap enough to call from a repeating timer: it copies only the window
    asked for, not the whole take.
    """
    want = int(round(seconds * self._sr)) * self._channels
    with self._lock:
        chunks, total = list(self._chunks), self._frames * self._channels
    if not chunks or want <= 0:
        return np.zeros(0, dtype=np.float32)
    collected: List[np.ndarray] = []
    got = 0
    for chunk in reversed(chunks):
        collected.append(chunk)
        got += chunk.size
        if got >= want:
            break
    flat = np.concatenate(list(reversed(collected)))[-min(want, total):]
    flat = flat.astype(np.float32, copy=False)
    if self._channels > 1:
        usable = (flat.size // self._channels) * self._channels
        return flat[:usable].reshape(-1, self._channels)
    return flat

drain

drain() -> ndarray

Audio captured since the last call, and never twice.

tail returns a sliding window, which overlaps between calls — right for a display, wrong for anything stateful. Feeding overlapping audio to a streaming denoiser processes the same samples repeatedly and corrupts every recursive estimate it holds.

>>> while rec.recording:
...     monitor(enhancer.process(rec.drain()))

Source code in kudio/core/stream.py

def drain(self) -> np.ndarray:
    """Audio captured **since the last call**, and never twice.

    :meth:`tail` returns a sliding window, which overlaps between calls —
    right for a display, wrong for anything stateful. Feeding overlapping
    audio to a streaming denoiser processes the same samples repeatedly and
    corrupts every recursive estimate it holds.

    >>> while rec.recording:                            # doctest: +SKIP
    ...     monitor(enhancer.process(rec.drain()))
    """
    with self._lock:
        new, self._drained = self._chunks[self._drained:], len(self._chunks)
    if not new:
        return np.zeros(0, dtype=np.float32)
    flat = np.concatenate(new).astype(np.float32, copy=False)
    if self._channels > 1:
        usable = (flat.size // self._channels) * self._channels
        return flat[:usable].reshape(-1, self._channels)
    return flat

reset

reset() -> None

Throw the take away and start counting again, stream untouched.

Source code in kudio/core/stream.py

def reset(self) -> None:
    """Throw the take away and start counting again, stream untouched."""
    with self._lock:
        self._chunks.clear()
        self._frames = 0
        self._drained = 0
        self._capped = False
        self._level = 0.0

record

record(
    seconds: float,
    sr: int = 16000,
    channels: int = 1,
    device: Optional[int] = None,
) -> ndarray

Blocking record for seconds via sounddevice (modern backend).

Returns float32 samples, shape (n,) for mono else (n, channels). Requires kudio[audio].

device is a sounddevice index — get one from kudio.list_devices, not from kudio.CheckDevice, which numbers devices through PyAudio for the streaming classes.

Source code in kudio/core/stream.py

def record(seconds: float, sr: int = 16000, channels: int = 1,
           device: Optional[int] = None) -> np.ndarray:
    """Blocking record for *seconds* via sounddevice (modern backend).

    Returns float32 samples, shape ``(n,)`` for mono else ``(n, channels)``.
    Requires ``kudio[audio]``.

    *device* is a **sounddevice** index — get one from
    :func:`kudio.list_devices`, not from :class:`kudio.CheckDevice`, which
    numbers devices through PyAudio for the streaming classes.
    """
    from kudio.util.check import _sounddevice
    sd = _sounddevice()
    frames = int(seconds * sr)
    data = sd.rec(frames, samplerate=sr, channels=channels,
                  dtype='float32', device=device)
    sd.wait()
    return data.squeeze()

wave_decode

wave_decode(in_data: bytes, channels: int) -> ndarray

Raw int16 bytes -> array of shape (chunk_length, channels).

Source code in kudio/core/stream.py

def wave_decode(in_data: bytes, channels: int) -> np.ndarray:
    """Raw int16 bytes -> array of shape ``(chunk_length, channels)``."""
    result = np.frombuffer(in_data, dtype=np.int16)
    chunk_length = len(result) // channels
    return np.reshape(result, (chunk_length, channels))

wave_encode

wave_encode(signal: ndarray) -> bytes

Multi-channel array -> interleaved int16 bytes.

Source code in kudio/core/stream.py

def wave_encode(signal: np.ndarray) -> bytes:
    """Multi-channel array -> interleaved int16 bytes."""
    return signal.flatten().astype(np.int16).tobytes()

play_audio

play_audio(
    wav: Union[str, Path, Wave_read, ndarray],
    rate: Optional[int] = None,
    channels: int = 1,
) -> Optional[str]

Play audio from a file path, an opened wave.Wave_read, or an ndarray.

ndarray input must be int16 or float32; rate is required in that case, and a (frames, channels) array plays on as many channels. A path is read with soundfile, so any format kudio.file_load reads plays. Returns an error message string when the audio cannot be played, else None.

Raises:

Type	Description
DependencyError	when neither sounddevice nor PyAudio is there.

Source code in kudio/core/stream.py

def play_audio(wav: Union[str, Path, wave.Wave_read, np.ndarray],
               rate: Optional[int] = None, channels: int = 1) -> Optional[str]:
    """Play audio from a file path, an opened ``wave.Wave_read``, or an ndarray.

    ndarray input must be int16 or float32; *rate* is required in that case,
    and a ``(frames, channels)`` array plays on as many channels. A path is
    read with soundfile, so any format :func:`kudio.file_load` reads plays.
    Returns an error message string when the audio cannot be played, else
    ``None``.

    :raises DependencyError: when neither sounddevice nor PyAudio is there.
    """
    if isinstance(wav, (str, Path)):
        if not Path(wav).is_file():
            return "input file can't be played"
        try:
            data, file_rate = sf.read(str(wav), dtype='float32', always_2d=False)
        except Exception as e:                                # noqa: BLE001
            log.error("play_audio: %s", e)
            return "input file can't be played"
        return play_audio(data, file_rate)

    if isinstance(wav, np.ndarray):
        if wav.dtype not in (np.int16, np.float32):
            log.error("play_audio: unsupported dtype %s", wav.dtype)
            return f"unsupported dtype: {wav.dtype}"
        try:
            import sounddevice as sd
            sd.play(wav, rate)
            return None
        except Exception:
            pass  # fall back to pyaudio
        pyaudio = _pyaudio()
        pya = pyaudio.PyAudio()
        try:
            fmt = pyaudio.paInt16 if wav.dtype == np.int16 else pyaudio.paFloat32
            # the array's own channel count, as sounddevice uses it: a stereo
            # array opened as one channel played at half speed
            n_channels = wav.shape[1] if wav.ndim == 2 else channels
            stream = pya.open(format=fmt, channels=n_channels, rate=rate,
                              output=True)
            stream.write(wav.tobytes())
            stream.stop_stream()
            stream.close()
            return None
        except Exception as e:
            log.error("play_audio failed: %s", e)
            return "can't be played by pyaudio"
        finally:
            pya.terminate()

    if isinstance(wav, wave.Wave_read):
        pyaudio = _pyaudio()
        pya = pyaudio.PyAudio()
        try:
            # `wave` only reads integer PCM, but PyAudio maps a 4-byte width
            # to float32, which played a PCM_32 file as noise
            width = wav.getsampwidth()
            fmt = pyaudio.paInt32 if width == 4 else pya.get_format_from_width(width)
            stream = pya.open(format=fmt,
                              channels=wav.getnchannels(),
                              rate=wav.getframerate(),
                              output=True)
            data = wav.readframes(1024)
            while data != b'':
                stream.write(data)
                data = wav.readframes(1024)
            stream.stop_stream()
            stream.close()
            return None
        except Exception as e:
            log.error("play_audio failed: %s", e)
            return "can't be played"
        finally:
            pya.terminate()

    return "unsupported input"

kudio.core.buffer

Thread-safe frame buffers shared between audio producers and consumers.

Classes:

Name	Description
AudioBuffer	Bounded FIFO of audio frames shared between producer/consumer threads.
FetchBuffer	Collect a fixed number of frames from an AudioBuffer in a

AudioBuffer

AudioBuffer(size: int = 2)

Bounded FIFO of audio frames shared between producer/consumer threads.

A stream reader calls add for every incoming frame; consumers either poll frames with get (e.g. for real-time plotting) or collect a fixed number of consecutive frames with get_fixed_size_data (e.g. timed recording). All methods are thread-safe.

Methods:

Name	Description
add	Append one frame.
get	Pop the oldest frame, blocking until one is available.
get_fixed_size_data	Block until queue_size frames have arrived, then return them
clear	Drop all buffered frames. Returns True if anything was dropped.

Source code in kudio/core/buffer.py

def __init__(self, size: int = 2):
    if size < 1:
        raise ValueError(f"size must be >= 1, got {size}")
    self.buffer_size = int(size)
    self._frames: deque = deque()
    self._cond = threading.Condition()
    # fixed-size capture state (timed recording)
    self._capture: Optional[List[np.ndarray]] = None
    self._capture_left = 0
    self._capture_done = threading.Event()

add

add(data: ndarray, drop_full: bool = True) -> None

Append one frame.

With drop_full=True (default) the oldest frame is dropped when the buffer is full; otherwise the call blocks until a slot is free.

Source code in kudio/core/buffer.py

def add(self, data: np.ndarray, drop_full: bool = True) -> None:
    """Append one frame.

    With ``drop_full=True`` (default) the oldest frame is dropped when the
    buffer is full; otherwise the call blocks until a slot is free.
    """
    with self._cond:
        # feed an active fixed-size capture first so timed recordings
        # do not lose frames to a slow consumer
        if self._capture is not None and self._capture_left > 0:
            self._capture.append(data)
            self._capture_left -= 1
            if self._capture_left == 0:
                self._capture_done.set()
        if drop_full:
            if len(self._frames) >= self.buffer_size:
                self._frames.popleft()
        else:
            while len(self._frames) >= self.buffer_size:
                self._cond.wait()
        self._frames.append(data)
        self._cond.notify_all()

get

get(timeout: Optional[float] = None) -> ndarray

Pop the oldest frame, blocking until one is available.

Raises TimeoutError if timeout (seconds) elapses first.

Source code in kudio/core/buffer.py

def get(self, timeout: Optional[float] = None) -> np.ndarray:
    """Pop the oldest frame, blocking until one is available.

    Raises ``TimeoutError`` if *timeout* (seconds) elapses first.
    """
    with self._cond:
        while not self._frames:
            if not self._cond.wait(timeout=timeout):
                raise TimeoutError("AudioBuffer.get timed out")
        data = self._frames.popleft()
        self._cond.notify_all()
        return data

get_fixed_size_data

get_fixed_size_data(
    queue_size: int, timeout: Optional[float] = None
) -> ndarray

Block until queue_size frames have arrived, then return them concatenated with np.hstack.

Raises ValueError when queue_size is below 1, because that wait would never end, and TypeError when it is not a whole number. A timeout that expires raises TimeoutError and ends the capture, so nothing is left waiting on the buffer to take the next one's frames. One capture runs at a time: a second one started while the first waits raises RuntimeError, since the two would share one set of frames.

Source code in kudio/core/buffer.py

def get_fixed_size_data(self, queue_size: int,
                        timeout: Optional[float] = None) -> np.ndarray:
    """Block until *queue_size* frames have arrived, then return them
    concatenated with ``np.hstack``.

    Raises ``ValueError`` when *queue_size* is below 1, because that wait
    would never end, and ``TypeError`` when it is not a whole number. A
    *timeout* that expires raises ``TimeoutError`` and ends the capture,
    so nothing is left waiting on the buffer to take the next one's
    frames. One capture runs at a time: a second one started while the
    first waits raises ``RuntimeError``, since the two would share one
    set of frames.
    """
    count = _checked_frame_count(queue_size, "AudioBuffer.get_fixed_size_data")
    with self._cond:
        if self._capture is not None:
            raise RuntimeError("AudioBuffer: a capture is already running on "
                               "this buffer; wait for it, or capture from "
                               "another AudioBuffer")
        self._capture = []
        self._capture_left = count
        self._capture_done.clear()
    log.info("Buffer: capturing %d frames", count)
    try:
        finished = self._capture_done.wait(timeout)
    finally:
        # also when the wait is interrupted (Ctrl-C in a notebook), or the
        # buffer would refuse every capture after this one
        with self._cond:
            frames, self._capture = self._capture, None
            self._capture_left = 0
    if not finished:
        raise TimeoutError(
            "AudioBuffer.get_fixed_size_data timed out after {0} s with {1} "
            "of {2} frames".format(timeout, len(frames or ()), count))
    return np.hstack(frames)

clear

clear() -> bool

Drop all buffered frames. Returns True if anything was dropped.

Source code in kudio/core/buffer.py

def clear(self) -> bool:
    """Drop all buffered frames. Returns True if anything was dropped."""
    with self._cond:
        had_data = bool(self._frames)
        self._frames.clear()
        self._cond.notify_all()
    if had_data:
        log.info("Buffer: cleared")
    return had_data

FetchBuffer

FetchBuffer(buffer: AudioBuffer)

Bases: Thread

Collect a fixed number of frames from an AudioBuffer in a background thread.

>>> data = FetchBuffer(buffer).get_data(desired_queue_size=10)

One instance runs one capture, because a thread cannot be started twice.

Methods:

Name	Description
get_data	Run the capture and return the frames.

Source code in kudio/core/buffer.py

def __init__(self, buffer: AudioBuffer):
    super().__init__(daemon=True)
    if not isinstance(buffer, AudioBuffer):
        raise TypeError("buffer must be an AudioBuffer")
    self.buffer = buffer
    self.queue_size: Optional[int] = None
    self.data: Optional[np.ndarray] = None
    self._timeout: Optional[float] = None
    self._error: Optional[BaseException] = None

get_data

get_data(
    desired_queue_size: int, timeout: Optional[float] = None
) -> ndarray

Run the capture and return the frames.

Always returns an array; a timeout that expires raises TimeoutError instead, so the caller can tell a stream that stopped delivering from a capture that came back short.

Source code in kudio/core/buffer.py

def get_data(self, desired_queue_size: int, timeout: Optional[float] = None) -> np.ndarray:
    """Run the capture and return the frames.

    Always returns an array; a *timeout* that expires raises
    ``TimeoutError`` instead, so the caller can tell a stream that stopped
    delivering from a capture that came back short.
    """
    if self.queue_size is not None:
        raise RuntimeError(
            "FetchBuffer.get_data runs once per instance and this one has "
            "already run. Build a new FetchBuffer for the next capture."
        )
    self.queue_size = _checked_frame_count(
        desired_queue_size, "FetchBuffer.get_data"
    )
    self._timeout = timeout
    self.start()
    # the capture ends itself at the timeout, so this join returns; the
    # thread used to be left waiting on the buffer and took the next
    # capture's frames
    data = self.join()
    if isinstance(self._error, TimeoutError):
        raise TimeoutError(
            "FetchBuffer.get_data timed out after {0} s waiting for {1} "
            "frames. Check that the stream is running, or allow more "
            "time.".format(timeout, self.queue_size))
    if data is None:
        raise RuntimeError(
            "FetchBuffer.get_data collected nothing for {0} frames. The "
            "capture thread ended early; its traceback is on "
            "stderr.".format(self.queue_size)
        )
    return data

kudio.util.check

Audio device discovery and validation (requires PyAudio).

Classes:

Name	Description
CheckDevice	Query/validate PortAudio devices.

Functions:

Name	Description
list_devices	Devices as sounddevice sees them.
is_available	Smoke-test the terminal colors and the audio input stack.
check_device	List system devices and probe an input; True on success.

CheckDevice

CheckDevice(frame_size: int = 1024)

Query/validate PortAudio devices.

>>> with CheckDevice() as cd:
...     devices = cd.system_devices()
...     device, rate, channels = cd.input_device()

Methods:

Name	Description
system_devices	Return [{'Device', 'index', 'fs', 'out', 'in'}, ...] for host API 0.
check_available_rate	Return the standard sample rates the device accepts, or False.
input_device	Resolve and validate an input device; returns (device, rate, channels).

Source code in kudio/util/check.py

def __init__(self, frame_size: int = 1024):
    self.p = _pyaudio().PyAudio()
    self.frame_size = frame_size
    self.device: Optional[int] = None
    self.rate: Optional[int] = None
    self.channels: Optional[int] = None

system_devices

system_devices() -> List[dict]

Return [{'Device', 'index', 'fs', 'out', 'in'}, ...] for host API 0.

Source code in kudio/util/check.py

def system_devices(self) -> List[dict]:
    """Return `[{'Device', 'index', 'fs', 'out', 'in'}, ...]` for host API 0."""
    device_list = []
    info = self.p.get_host_api_info_by_index(0)
    for i in range(info.get('deviceCount')):
        dev = self.p.get_device_info_by_host_api_device_index(0, i)
        device_list.append({
            'Device': dev.get('name'),
            'index': dev.get('index'),
            'fs': dev.get('defaultSampleRate'),
            'out': dev.get('maxOutputChannels'),
            'in': dev.get('maxInputChannels'),
        })
        log.info("Device %s: in=%s out=%s (%s)",
                 dev.get('index'), dev.get('maxInputChannels'),
                 dev.get('maxOutputChannels'), dev.get('name'))
    return device_list

check_available_rate

check_available_rate(device: int) -> Union[List[int], bool]

Return the standard sample rates the device accepts, or False.

Source code in kudio/util/check.py

def check_available_rate(self, device: int) -> Union[List[int], bool]:
    """Return the standard sample rates the device accepts, or False."""
    if not self._is_input_device(device):
        log.warning("Device unusable: %s", self._describe(device))
        return False
    return [rt for rt in self.standard_rate if self._is_input_device(device, rt)]

input_device

input_device(
    device: Optional[int] = None,
    rate: Optional[int] = None,
    channels: Optional[int] = None,
) -> Tuple[int, int, int]

Resolve and validate an input device; returns (device, rate, channels).

Missing arguments are auto-detected (last input-capable device, first working rate of 16000/22050/44100, mono).

Raises:

Type	Description
DeviceError	when there is no input, or the device, rate or channel count will not open.

Source code in kudio/util/check.py

def input_device(self, device: Optional[int] = None,
                 rate: Optional[int] = None,
                 channels: Optional[int] = None
                 ) -> Tuple[int, int, int]:
    """Resolve and validate an input device; returns (device, rate, channels).

    Missing arguments are auto-detected (last input-capable device, first
    working rate of 16000/22050/44100, mono).

    :raises DeviceError: when there is no input, or the device, rate or
        channel count will not open.
    """
    if device is None:
        mics = [d for d in range(self.p.get_device_count())
                if self._is_input_device(d)]
        if not mics:
            raise DeviceError("No microphone found on this system")
        log.info("Found %d microphone(s)", len(mics))
        device = mics[-1]
    if not self._is_input_device(device):
        raise DeviceError(f"Device unusable: {self._describe(device)}")

    # sample rate
    if not (rate is not None and self._is_input_device(device, rate)):
        rate = next((r for r in (16000, 22050, 44100)
                     if self._is_input_device(device, r)), None)
        if rate is None:
            rate = int(self.get_device_info(device)["defaultSampleRate"])
    if not self._is_input_device(device, rate):
        raise DeviceError(
            f"Bad sample rate for device: {self._describe(device)}")
    log.info("Device %s, sample rate %s", device, rate)

    # channels
    if channels is not None:
        if self.get_device_info(device)["maxInputChannels"] < channels:
            log.warning("Requested channels exceed device maximum: %s",
                        self._describe(device))
            channels = 1
    else:
        channels = 1
    if not self._is_input_device(device, rate, channels):
        raise DeviceError(
            f"Bad channel setting ({channels}): {self._describe(device)}")

    self.device, self.rate, self.channels = device, rate, channels
    self.show_current_device_info()
    return self.device, self.rate, self.channels

list_devices

list_devices(kind: Optional[str] = None) -> List[dict]

Devices as sounddevice sees them.

This is the index space kudio.record accepts. It is not the same numbering as CheckDevice, which enumerates through PyAudio for the streaming/recorder classes — PortAudio's two Python bindings index devices independently, so an index from one must never be passed to the other.

Parameters:

Name	Type	Description	Default
kind	Optional[str]	'input', 'output', or None for every device.	None

Returns:

Type	Description
List[dict]	[{'index', 'name', 'max_input_channels', 'max_output_channels', 'default_samplerate'}, ...] >>> [d['name'] for d in kudio.list_devices('input')]

Source code in kudio/util/check.py

def list_devices(kind: Optional[str] = None) -> List[dict]:
    """Devices as **sounddevice** sees them.

    This is the index space :func:`kudio.record` accepts. It is *not* the same
    numbering as :class:`CheckDevice`, which enumerates through PyAudio for the
    streaming/recorder classes — PortAudio's two Python bindings index devices
    independently, so an index from one must never be passed to the other.

    :param kind: ``'input'``, ``'output'``, or ``None`` for every device.
    :returns: ``[{'index', 'name', 'max_input_channels',
        'max_output_channels', 'default_samplerate'}, ...]``

    >>> [d['name'] for d in kudio.list_devices('input')]        # doctest: +SKIP
    """
    if kind not in (None, 'input', 'output'):
        raise ValueError(f"kind must be 'input', 'output' or None, got {kind!r}")
    sd = _sounddevice()

    devices = []
    for index, info_ in enumerate(sd.query_devices()):
        inputs = int(info_.get('max_input_channels', 0))
        outputs = int(info_.get('max_output_channels', 0))
        if kind == 'input' and inputs < 1:
            continue
        if kind == 'output' and outputs < 1:
            continue
        devices.append({
            'index': index,
            'name': info_.get('name', ''),
            'max_input_channels': inputs,
            'max_output_channels': outputs,
            'default_samplerate': float(info_.get('default_samplerate', 0.0)),
        })
    return devices

is_available

is_available() -> bool

Smoke-test the terminal colors and the audio input stack.

Source code in kudio/util/check.py

def is_available() -> bool:
    """Smoke-test the terminal colors and the audio input stack."""
    from kudio.util.colors import color_test
    return color_test('[kudio] Color test') and check_device()

check_device

check_device() -> bool

List system devices and probe an input; True on success.

The input probed is the one CheckDevice.input_device picks with no arguments: the last input-capable device PyAudio lists, at the first of 16000, 22050 and 44100 Hz it opens at.

Source code in kudio/util/check.py

@timer
def check_device() -> bool:
    """List system devices and probe an input; True on success.

    The input probed is the one :meth:`CheckDevice.input_device` picks with no
    arguments: the last input-capable device PyAudio lists, at the first of
    16000, 22050 and 44100 Hz it opens at.
    """
    try:
        with CheckDevice() as cd:
            cd.system_devices()
            cd.input_device()
        log.info("Audio test passed")
        return True
    except Exception as e:
        log.error("Audio test failed: %s", e)
        return False