Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .github/workflows/python-versions.yml
Original file line number Diff line number Diff line change
Expand Up @@ -35,3 +35,10 @@ jobs:
python -m sounddevice
python -c "import sounddevice as sd; print(sd._libname)"
python -c "import sounddevice as sd; print(sd.get_portaudio_version())"
- name: Install optional NumPy for convenience API tests
if: ${{ !startsWith(matrix.python-version, 'pypy-') }}
run: python -m pip install numpy
- name: Run synthetic ALSA regressions
timeout-minutes: 2
working-directory: git-repo
run: python -m pytest -q tests
3 changes: 3 additions & 0 deletions doc/api/platform-specific-settings.rst
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,13 @@ Platform-specific Settings
.. autosummary::
:nosignatures:

AlsaSettings
AsioSettings
CoreAudioSettings
WasapiSettings

.. autoclass:: AlsaSettings

.. autoclass:: AsioSettings

.. autoclass:: CoreAudioSettings
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ numpy = ["numpy"]

[dependency-groups]
dev = [{ include-group = "test" }, { include-group = "doc" }]
test = []
test = ["pytest"]
doc = [
"insipid-sphinx-theme",
"sphinx-last-updated-by-git",
Expand Down
9 changes: 9 additions & 0 deletions sounddevice_build.py
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,15 @@

typedef unsigned long PaWinWaveFormatChannelMask;

/* pa_linux_alsa.h */
typedef struct PaAlsaStreamInfo {
unsigned long size;
PaHostApiTypeId hostApiType;
unsigned long version;
const char *deviceString;
} PaAlsaStreamInfo;
void PaAlsa_InitializeStreamInfo(PaAlsaStreamInfo *info);

/* pa_asio.h */

#define paAsioUseChannelSelectors 0x01
Expand Down
117 changes: 97 additions & 20 deletions src/sounddevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -173,7 +173,8 @@ def play(data, samplerate=None, mapping=None, blocking=False, loop=False,

"""
ctx = _CallbackContext(loop=loop)
ctx.frames = ctx.check_data(data, mapping, kwargs.get('device'))
ctx.frames = ctx.check_data(
data, mapping, kwargs.get('device'), kwargs.get('extra_settings'))

def callback(outdata, frames, time, status):
assert len(outdata) == frames
Expand Down Expand Up @@ -359,7 +360,8 @@ def playrec(data, samplerate=None, channels=None, dtype=None,

"""
ctx = _CallbackContext()
output_frames = ctx.check_data(data, output_mapping, kwargs.get('device'))
output_frames = ctx.check_data(
data, output_mapping, kwargs.get('device'), kwargs.get('extra_settings'))
if dtype is None:
dtype = ctx.data.dtype # ignore module defaults
out, input_frames = ctx.check_out(out, output_frames, channels, dtype,
Expand Down Expand Up @@ -690,7 +692,7 @@ def check_input_settings(device=None, channels=None, dtype=None,
"""
parameters, dtype, samplesize, samplerate = _get_stream_parameters(
'input', device=device, channels=channels, dtype=dtype, latency=None,
extra_settings=extra_settings, samplerate=samplerate)
extra_settings=extra_settings, samplerate=samplerate, for_check=True)
_check(_lib.Pa_IsFormatSupported(parameters, _ffi.NULL, samplerate))


Expand All @@ -704,7 +706,7 @@ def check_output_settings(device=None, channels=None, dtype=None,
"""
parameters, dtype, samplesize, samplerate = _get_stream_parameters(
'output', device=device, channels=channels, dtype=dtype, latency=None,
extra_settings=extra_settings, samplerate=samplerate)
extra_settings=extra_settings, samplerate=samplerate, for_check=True)
_check(_lib.Pa_IsFormatSupported(_ffi.NULL, parameters, samplerate))


Expand Down Expand Up @@ -808,6 +810,12 @@ def __init__(self, kind, samplerate=None, blocksize=None, device=None,
if prime_output_buffers_using_stream_callback:
stream_flags |= _lib.paPrimeOutputBuffersUsingStreamCallback

# Retain effective settings, including their C buffers, through close.
if extra_settings is None:
extra_settings = tuple(default.extra_settings)
self._extra_settings = (tuple(_split(extra_settings)),
tuple(default.extra_settings))

if kind == 'duplex':
idevice, odevice = _split(device)
ichannels, ochannels = _split(channels)
Expand Down Expand Up @@ -2167,7 +2175,7 @@ class default:

See Also
--------
AsioSettings, CoreAudioSettings, WasapiSettings
AlsaSettings, AsioSettings, CoreAudioSettings, WasapiSettings

"""
samplerate: ... = None
Expand Down Expand Up @@ -2316,6 +2324,44 @@ class CallbackAbort(Exception):
"""


class AlsaSettings:

def __init__(self, device_string):
"""ALSA-specific settings for opening a PCM by address.

Use as ``extra_settings`` with ``device=None`` (or ``-2``).
The address overrides `default.device`. Channels and sample rate
must be supplied explicitly or through defaults. Streams also
require a numeric latency in seconds.

Parameters
----------
device_string : str
ALSA PCM address, such as ``'hw:2,0'`` or a named PCM from the
ALSA configuration. It must be nonempty and contain no NUL.
An ALSA address is not a persistent hardware identifier.

Notes
-----
The stream's device attribute reports PortAudio's special value
``-2``. Passing an explicit device index or name is an error.
For direct output with ``mapping=[1]`` in `play()` or `playrec()`,
use `OutputStream` with explicit channels instead; automatic
stereo detection requires enumerated device information.

"""
if not isinstance(device_string, str):
raise TypeError('device_string must be a string')
if not device_string or '\0' in device_string:
raise ValueError('device_string must be nonempty and contain no NUL')
_check(_lib.Pa_HostApiTypeIdToHostApiIndex(_lib.paALSA),
'ALSA host API is unavailable')
self._device_string = _ffi.new('char[]', device_string.encode('utf-8'))
self._streaminfo: ... = _ffi.new('PaAlsaStreamInfo*')
_lib.PaAlsa_InitializeStreamInfo(self._streaminfo)
self._streaminfo.deviceString = self._device_string


class AsioSettings:

def __init__(self, channel_selectors):
Expand Down Expand Up @@ -2541,7 +2587,7 @@ def __init__(self, loop=False):
self.event = threading.Event()
self.status = CallbackFlags()

def check_data(self, data, mapping, device):
def check_data(self, data, mapping, device, extra_settings=None):
"""Check data and output mapping."""
import numpy as np
data = np.asarray(data)
Expand All @@ -2563,6 +2609,13 @@ def check_data(self, data, mapping, device):
# first two channels, which is unexpected when specifying mapping=[1].
# In this case, we play silence on the second channel, but only if the
# device actually supports a second channel:
effective_settings = (default.extra_settings if extra_settings is None
else extra_settings)
if (mapping_is_explicit and np.array_equal(mapping, [0]) and
isinstance(_select_input_or_output(effective_settings, 'output'),
AlsaSettings)):
raise ValueError('For ALSA direct output mapping=[1], use '
'OutputStream with explicit channels')
if (mapping_is_explicit and np.array_equal(mapping, [0]) and
query_devices(device, 'output')['max_output_channels'] >= 2):
channels = 2
Expand Down Expand Up @@ -2730,12 +2783,22 @@ def _check_dtype(dtype):


def _get_stream_parameters(kind, device, channels, dtype, latency,
extra_settings: ..., samplerate):
extra_settings: ..., samplerate, *, for_check=False):
"""Get parameters for one direction (input or output) of a stream."""
assert kind in ('input', 'output')
if device is None:
device = default.device
device = _get_device_id(device, kind, raise_on_error=True)
if extra_settings is None:
extra_settings = default.extra_settings
extra_settings = _select_input_or_output(extra_settings, kind)
direct_alsa = isinstance(extra_settings, AlsaSettings)
if direct_alsa:
device = _select_input_or_output(device, kind)
if device is not None and device != -2:
raise ValueError('AlsaSettings conflicts with an explicit device')
device = _lib.paUseHostApiSpecificDeviceSpecification
else:
if device is None:
device = default.device
device = _get_device_id(device, kind, raise_on_error=True)
if channels is None:
channels = default.channels
channels = _select_input_or_output(channels, kind)
Expand All @@ -2745,15 +2808,33 @@ def _get_stream_parameters(kind, device, channels, dtype, latency,
if latency is None:
latency = default.latency
latency = _select_input_or_output(latency, kind)
if extra_settings is None:
extra_settings = default.extra_settings
extra_settings = _select_input_or_output(extra_settings, kind)
if samplerate is None:
samplerate = default.samplerate

info = query_devices(device)
if channels is None:
channels = info['max_' + kind + '_channels']
if direct_alsa:
import math
import numbers
if (isinstance(channels, bool) or
not isinstance(channels, numbers.Integral) or channels < 1):
raise ValueError('AlsaSettings requires a positive channel count')
if (isinstance(samplerate, bool) or
not isinstance(samplerate, numbers.Real) or
not math.isfinite(samplerate) or samplerate <= 0):
raise ValueError('AlsaSettings requires a positive finite sample rate')
if for_check:
latency = 0.0 # Pa_IsFormatSupported ignores suggested latency.
elif (isinstance(latency, bool) or
not isinstance(latency, numbers.Real) or
not math.isfinite(latency) or latency < 0):
raise ValueError('AlsaSettings requires finite numeric latency >= 0')
else:
info = query_devices(device)
if channels is None:
channels = info['max_' + kind + '_channels']
if latency in ('low', 'high'):
latency = info['default_' + latency + '_' + kind + '_latency']
if samplerate is None:
samplerate = info['default_samplerate']
try:
# If NumPy is available, get canonical dtype name
dtype = _sys.modules['numpy'].dtype(dtype).name
Expand All @@ -2764,10 +2845,6 @@ def _get_stream_parameters(kind, device, channels, dtype, latency,
except KeyError as e:
raise ValueError('Invalid ' + kind + ' sample format') from e
samplesize = _check(_lib.Pa_GetSampleSize(sampleformat))
if latency in ('low', 'high'):
latency = info['default_' + latency + '_' + kind + '_latency']
if samplerate is None:
samplerate = info['default_samplerate']
parameters: ... = _ffi.new('PaStreamParameters*', (
device, channels, sampleformat, latency,
extra_settings._streaminfo if extra_settings else _ffi.NULL))
Expand Down
30 changes: 30 additions & 0 deletions tests/README.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
Synthetic ALSA tests
===================

These Linux tests exercise sounddevice and the real PortAudio ALSA backend
using null and file PCMs. No microphone or sound card is needed.
Run them in a fresh Python process so the isolated ALSA configuration is set
before PortAudio initialization. The fixture does not include system or user
ALSA configuration and restores the environment when the test session ends.

Install PortAudio and ALSA plugins (Debian/Ubuntu)::

sudo apt-get install libportaudio2
python -m pip install . pytest numpy
python -m pytest -q tests

The temporary directory contains a deterministic input file and ALSA config.
It is removed at the end of the session. Non-Linux runs skip these tests;
missing ALSA support on Linux fails rather than silently skipping coverage.

The tests cover direct input/output/duplex, exact input bytes, close/reopen,
invalid parameters, settings lifetime, default overrides and existing indexed
and named selection. They do not validate hardware identity, physical capture,
hotplug, real-time scheduling or a negotiated hardware buffer size.

The raw-stream tests require only pytest and sounddevice. The convenience API
test uses NumPy and reports a skip when NumPy is unavailable (including the
PyPy CI jobs). CPython CI installs NumPy and exercises that test too.

ALSA_CONFIG_PATH cannot represent spaces or colons in its configuration path.
If the process temporary directory contains either, the fixture uses /tmp.
66 changes: 66 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
"""Configure synthetic ALSA PCMs before sounddevice initializes PortAudio."""

import json
import random
import os
from pathlib import Path
import sys
import tempfile

import pytest


def pytest_configure(config):
if sys.platform != 'linux':
return
if 'sounddevice' in sys.modules:
raise pytest.UsageError('Run ALSA tests in a fresh Python process')
# ALSA_CONFIG_PATH splits on spaces/colons without an escaping syntax.
temp_root = tempfile.gettempdir()
if ':' in temp_root or ' ' in temp_root:
temp_root = '/tmp'
directory = tempfile.TemporaryDirectory(
prefix='sounddevice-alsa-', dir=temp_root)
root = Path(directory.name)
rng = random.Random(0)
data = bytes(rng.getrandbits(8) for _ in range(192000))
(root / 'input.raw').write_bytes(data)
# This standalone configuration has no hardware PCMs or user includes.
alsa_config = root / 'alsa.conf'
alsa_config.write_text(
"""
pcm.null { type null }
# PortAudio 19.6 expects a deferred default/dmix entry during enumeration.
pcm.default { type null }
pcm.test_named {
type null
hint { show on description "Synthetic PCM" }
}
pcm.portaudio_file_capture {
type file
slave.pcm "null"
infile %s
file %s
format "raw"
}
"""
% (json.dumps(str(root / 'input.raw'), ensure_ascii=False),
json.dumps(str(root / 'output.raw'), ensure_ascii=False))
)
previous = os.environ.get('ALSA_CONFIG_PATH')
os.environ['ALSA_CONFIG_PATH'] = str(alsa_config)

def cleanup():
if previous is None:
os.environ.pop('ALSA_CONFIG_PATH', None)
else:
os.environ['ALSA_CONFIG_PATH'] = previous
directory.cleanup()

config.add_cleanup(cleanup)
config._alsa_capture_bytes = data


@pytest.fixture
def alsa_capture_bytes(pytestconfig):
return pytestconfig._alsa_capture_bytes
Loading