test_audio_stream.py

#! /usr/bin/env python
#
# Copyright 2023 Spotify AB
#
# Licensed under the GNU Public License, Version 3.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    https://www.gnu.org/licenses/gpl-3.0.html
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import time

import numpy as np
import pytest

import pedalboard

# Very silly: even just creating an AudioStream object that reads from an `iPhone Microphone``
# will cause a locally-present iPhone to emit a sound. Running `pytest` on my laptop makes my
# phone ding.
DEVICE_NAMES_TO_SKIP = {
    "iPhone Microphone",
    "AirPods",
    # PulseAudio devices:
    "Open Sound System",
    "JACK Audio Connection Kit",
    "plugin",
}
INPUT_DEVICE_NAMES = [
    n
    for n in pedalboard.io.AudioStream.input_device_names
    if not any(substring.lower() in n.lower() for substring in DEVICE_NAMES_TO_SKIP)
]

OUTPUT_DEVICE_NAMES = [
    n
    for n in pedalboard.io.AudioStream.output_device_names
    if not any(substring.lower() in n.lower() for substring in DEVICE_NAMES_TO_SKIP)
]

ACCEPTABLE_ERRORS_ON_CI = {"No driver"}


# Note: this test may do nothing on CI, because we don't have mock audio devices available.
# This will run on Linux, macOS and probably Windows as long as at least one audio device is available.
@pytest.mark.parametrize("input_device_name", INPUT_DEVICE_NAMES)
@pytest.mark.parametrize("output_device_name", OUTPUT_DEVICE_NAMES)
def test_create_stream(input_device_name: str, output_device_name: str):
    try:
        stream = pedalboard.io.AudioStream(
            input_device_name, output_device_name, allow_feedback=True
        )
    except Exception as e:
        if any(substr in str(e) for substr in ACCEPTABLE_ERRORS_ON_CI):
            raise pytest.skip(str(e))
        raise

    assert stream is not None
    assert input_device_name in repr(stream)
    assert output_device_name in repr(stream)
    assert not stream.running
    assert isinstance(stream.plugins, pedalboard.Chain)
    with stream:
        assert stream.running

        # Ensure that modifying a running stream does not crash:
        stream.plugins.append(pedalboard.Gain(gain_db=-120))
        for _ in range(0, 10):
            stream.plugins.append(pedalboard.Gain(gain_db=-120))
        for i in reversed(range(len(stream.plugins))):
            del stream.plugins[i]
        assert stream.running
    assert not stream.running


# Note: this test may do nothing on CI, because we don't have mock audio devices available.
# This will run on Linux, macOS and probably Windows as long as at least one audio device is available.
@pytest.mark.skipif(
    (
        pedalboard.io.AudioStream.default_output_device_name == "Null Audio Device"
        or pedalboard.io.AudioStream.default_output_device_name is None
    ),
    reason="Test requires a working audio device.",
)
def test_write_to_stream():
    try:
        stream = pedalboard.io.AudioStream(
            None, pedalboard.io.AudioStream.default_output_device_name
        )
    except Exception as e:
        if any(substr in str(e) for substr in ACCEPTABLE_ERRORS_ON_CI):
            raise pytest.skip(str(e))
        raise

    assert stream is not None
    assert not stream.running
    with stream:
        assert stream.running
        stream.write(np.zeros((2, 1024), dtype=np.float32), stream.sample_rate)
    assert not stream.running


# Note: this test may do nothing on CI, because we don't have mock audio devices available.
# This will run on Linux, macOS and probably Windows as long as at least one audio device is available.
@pytest.mark.skipif(
    (
        pedalboard.io.AudioStream.default_output_device_name == "Null Audio Device"
        or pedalboard.io.AudioStream.default_output_device_name is None
    ),
    reason="Test requires a working audio device.",
)
def test_write_to_stream_without_opening():
    try:
        stream = pedalboard.io.AudioStream(
            None, pedalboard.io.AudioStream.default_output_device_name
        )
    except Exception as e:
        if any(substr in str(e) for substr in ACCEPTABLE_ERRORS_ON_CI):
            raise pytest.skip(str(e))
        raise

    assert stream is not None
    assert not stream.running
    stream.write(np.zeros((2, 1024), dtype=np.float32), stream.sample_rate)
    assert not stream.running
    assert stream.buffered_input_sample_count is None


# Note: this test may do nothing on CI, because we don't have mock audio devices available.
# This will run on Linux, macOS and probably Windows as long as at least one audio device is available.
@pytest.mark.skipif(
    (
        pedalboard.io.AudioStream.default_input_device_name == "Null Audio Device"
        or pedalboard.io.AudioStream.default_input_device_name is None
    ),
    reason="Test requires a working audio device.",
)
def test_read_from_stream():
    try:
        stream = pedalboard.io.AudioStream(pedalboard.io.AudioStream.default_input_device_name)
    except Exception as e:
        if any(substr in str(e) for substr in ACCEPTABLE_ERRORS_ON_CI):
            raise pytest.skip(str(e))
        raise

    assert stream is not None
    assert not stream.running
    result = stream.read(1)
    assert result.shape == (stream.num_input_channels, 1)
    assert not stream.running
    assert stream.buffered_input_sample_count == 0


# Note: this test may do nothing on CI, because we don't have mock audio devices available.
# This will run on Linux, macOS and probably Windows as long as at least one audio device is available.
@pytest.mark.skipif(
    (
        pedalboard.io.AudioStream.default_input_device_name == "Null Audio Device"
        or pedalboard.io.AudioStream.default_input_device_name is None
    ),
    reason="Test requires a working audio device.",
)
def test_read_from_stream_measures_dropped_frames():
    try:
        stream = pedalboard.io.AudioStream(pedalboard.io.AudioStream.default_input_device_name)
    except Exception as e:
        if any(substr in str(e) for substr in ACCEPTABLE_ERRORS_ON_CI):
            raise pytest.skip(str(e))
        raise

    assert stream is not None
    with stream:
        if stream.sample_rate == 0:
            raise pytest.skip("Sample rate of default audio device is 0")
        assert stream.running
        assert stream.dropped_input_frame_count == 0
        # Sleep long enough to overflow the internal buffer and cause dropped frames.
        # Use a generous multiplier to account for varying buffer sizes across platforms.
        time.sleep(max(0.5, 20 * stream.buffer_size / stream.sample_rate))
        assert (
            stream.buffered_input_sample_count is not None
            and stream.buffered_input_sample_count > 0
        )
        dropped_count = stream.dropped_input_frame_count
        if dropped_count == 0:
            raise pytest.skip(
                "No frames were dropped during the test window; "
                "audio device may have a large internal buffer"
            )
    # The input buffer was cleared on __exit__, so the buffer count should be zero:
    assert stream.buffered_input_sample_count == 0

    # ...but we should still know how many frames were dropped.
    # Allow up to one extra buffer's worth of dropped frames, because the audio
    # thread may drop one more buffer between our snapshot and stream shutdown.
    assert stream.dropped_input_frame_count >= dropped_count
    assert stream.dropped_input_frame_count <= dropped_count + stream.buffer_size