Fix multiple issues with tcod.sdl.audio

Adds much needed testing of the tcod.sdl.audio module.

Convert some asserts into real errors.

Handle exceptions raised in audio callback as unraisable.

Author: HexDecimal

.vscode/settings.json

@@ -412,6 +412,7 @@
         "typestr",
         "undoc",
         "Unifont",
+        "unraisable",
         "unraisablehook",
         "unraiseable",
         "upscaling",

CHANGELOG.md

@@ -4,6 +4,10 @@ Changes relevant to the users of python-tcod are documented here.
 This project adheres to [Semantic Versioning](https://semver.org/) since version `2.0.0`.
 
 ## [Unreleased]
+### Fixed
+- `AudioDevice.stopped` was inverted.
+- Fixed the audio mixer stop and fadeout methods.
+- Exceptions raised in the audio mixer callback no longer cause a messy crash, they now go to `sys.unraisablehook`.
 
 ## [16.0.0] - 2023-05-27
 ### Added

tcod/sdl/audio.py

@@ -47,6 +47,7 @@
 import sys
 import threading
 import time
+from dataclasses import dataclass
 from types import TracebackType
 from typing import Any, Callable, Hashable, Iterator
 
@@ -65,7 +66,9 @@ def _get_format(format: DTypeLike) -> int:
     assert dt.fields is None
     bitsize = dt.itemsize * 8
     assert 0 < bitsize <= lib.SDL_AUDIO_MASK_BITSIZE
-    assert dt.str[1] in "uif"
+    if not dt.str[1] in "uif":
+        msg = f"Unexpected dtype: {dt}"
+        raise TypeError(msg)
     is_signed = dt.str[1] != "u"
     is_float = dt.str[1] == "f"
     byteorder = dt.byteorder
@@ -81,7 +84,21 @@ def _get_format(format: DTypeLike) -> int:
 
 
 def _dtype_from_format(format: int) -> np.dtype[Any]:
-    """Return a dtype from a SDL_AudioFormat."""
+    """Return a dtype from a SDL_AudioFormat.
+
+    >>> _dtype_from_format(tcod.lib.AUDIO_F32LSB)
+    dtype('float32')
+    >>> _dtype_from_format(tcod.lib.AUDIO_F32MSB)
+    dtype('>f4')
+    >>> _dtype_from_format(tcod.lib.AUDIO_S16LSB)
+    dtype('int16')
+    >>> _dtype_from_format(tcod.lib.AUDIO_S16MSB)
+    dtype('>i2')
+    >>> _dtype_from_format(tcod.lib.AUDIO_U16LSB)
+    dtype('uint16')
+    >>> _dtype_from_format(tcod.lib.AUDIO_U16MSB)
+    dtype('>u2')
+    """
     bitsize = format & lib.SDL_AUDIO_MASK_BITSIZE
     assert bitsize % 8 == 0
     byte_size = bitsize // 8
@@ -203,6 +220,8 @@ def __init__(
 
     def __repr__(self) -> str:
         """Return a representation of this device."""
+        if self.stopped:
+            return f"<{self.__class__.__name__}() stopped=True>"
         items = [
             f"{self.__class__.__name__}(device_id={self.device_id})",
             f"frequency={self.frequency}",
@@ -211,7 +230,9 @@ def __repr__(self) -> str:
             f"channels={self.channels}",
             f"buffer_samples={self.buffer_samples}",
             f"buffer_bytes={self.buffer_bytes}",
+            f"paused={self.paused}",
         ]
+
         if self.silence:
             items.append(f"silence={self.silence}")
         if self._handle is not None:
@@ -241,7 +262,9 @@ def _sample_size(self) -> int:
     @property
     def stopped(self) -> bool:
         """Is True if the device has failed or was closed."""
-        return bool(lib.SDL_GetAudioDeviceStatus(self.device_id) != lib.SDL_AUDIO_STOPPED)
+        if not hasattr(self, "device_id"):
+            return True
+        return bool(lib.SDL_GetAudioDeviceStatus(self.device_id) == lib.SDL_AUDIO_STOPPED)
 
     @property
     def paused(self) -> bool:
@@ -404,7 +427,9 @@ def play(
     def _verify_audio_sample(self, sample: ArrayLike) -> NDArray[Any]:
         """Verify an audio sample is valid and return it as a Numpy array."""
         array: NDArray[Any] = np.asarray(sample)
-        assert array.dtype == self.mixer.device.format
+        if array.dtype != self.mixer.device.format:
+            msg = f"Audio sample must be dtype={self.mixer.device.format}, input was dtype={array.dtype}"
+            raise TypeError(msg)
         if len(array.shape) == 1:
             array = array[:, np.newaxis]
         return array
@@ -434,7 +459,7 @@ def fadeout(self, time: float) -> None:
             time_samples = round(time * self.mixer.device.frequency) + 1
             buffer: NDArray[np.float32] = np.zeros((time_samples, self.mixer.device.channels), np.float32)
             self._on_mix(buffer)
-            buffer *= np.linspace(1.0, 0.0, time_samples + 1, endpoint=False)[1:]
+            buffer *= np.linspace(1.0, 0.0, time_samples + 1, endpoint=False)[1:, np.newaxis]
             self.sound_queue[:] = [buffer]
 
     def stop(self) -> None:
@@ -536,13 +561,41 @@ class _AudioCallbackUserdata:
     device: AudioDevice
 
 
+@dataclass
+class _UnraisableHookArgs:
+    exc_type: type[BaseException]
+    exc_value: BaseException | None
+    exc_traceback: TracebackType | None
+    err_msg: str | None
+    object: object
+
+
+class _ProtectedContext:
+    def __init__(self, obj: object = None) -> None:
+        self.obj = obj
+
+    def __enter__(self) -> None:
+        pass
+
+    def __exit__(
+        self, exc_type: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None
+    ) -> bool:
+        if exc_type is None:
+            return False
+        if sys.version_info < (3, 8):
+            return False
+        sys.unraisablehook(_UnraisableHookArgs(exc_type, value, traceback, None, self.obj))  # type: ignore[arg-type]
+        return True
+
+
 @ffi.def_extern()  # type: ignore
-def _sdl_audio_callback(userdata: Any, stream: Any, length: int) -> None:
+def _sdl_audio_callback(userdata: Any, stream: Any, length: int) -> None:  # noqa: ANN401
     """Handle audio device callbacks."""
     data: _AudioCallbackUserdata = ffi.from_handle(userdata)
     device = data.device
     buffer = np.frombuffer(ffi.buffer(stream, length), dtype=device.format).reshape(-1, device.channels)
-    device._callback(device, buffer)
+    with _ProtectedContext(device):
+        device._callback(device, buffer)
 
 
 def _get_devices(capture: bool) -> Iterator[str]:

tests/test_sdl.py

@@ -1,11 +1,9 @@
 """Test SDL specific features."""
-import contextlib
 import sys
 
 import numpy as np
 import pytest
 
-import tcod.sdl.audio
 import tcod.sdl.render
 import tcod.sdl.sys
 import tcod.sdl.video
@@ -83,15 +81,3 @@ def test_sdl_render_bad_types() -> None:
         tcod.sdl.render.Renderer(tcod.ffi.cast("SDL_Renderer*", tcod.ffi.NULL))
     with pytest.raises(TypeError):
         tcod.sdl.render.Renderer(tcod.ffi.new("SDL_Rect*"))
-
-
-def test_sdl_audio_device() -> None:
-    with contextlib.closing(tcod.sdl.audio.open(frequency=44100, format=np.float32, channels=2, paused=True)) as device:
-        assert device.convert(np.zeros(4, dtype=np.float32), 22050).shape[0] == 8  # noqa: PLR2004
-        assert device.convert(np.zeros((4, 4), dtype=np.float32)).shape == (4, 2)
-        assert device.convert(np.zeros(4, dtype=np.int8)).shape[0] == 4  # noqa: PLR2004
-        device.paused = False
-        device.paused = True
-        assert device.queued_samples == 0
-        with contextlib.closing(tcod.sdl.audio.BasicMixer(device)) as mixer:
-            assert mixer

tests/test_sdl_audio.py

@@ -0,0 +1,121 @@
+"""Test tcod.sdl.audio module."""
+import contextlib
+import sys
+import time
+from typing import Any
+
+import numpy as np
+import pytest
+from numpy.typing import NDArray
+
+import tcod.sdl.audio
+
+# ruff: noqa: D103
+
+
+def test_devices() -> None:
+    list(tcod.sdl.audio.get_devices())
+    list(tcod.sdl.audio.get_capture_devices())
+
+
+def test_audio_device() -> None:
+    with tcod.sdl.audio.open(frequency=44100, format=np.float32, channels=2, paused=True) as device:
+        assert not device.stopped
+        assert device.convert(np.zeros(4, dtype=np.float32), 22050).shape[0] == 8  # noqa: PLR2004
+        assert device.convert(np.zeros((4, 4), dtype=np.float32)).shape == (4, 2)
+        assert device.convert(np.zeros(4, dtype=np.int8)).shape[0] == 4  # noqa: PLR2004
+        assert device.paused is True
+        device.paused = False
+        assert device.paused is False
+        device.paused = True
+        assert device.queued_samples == 0
+        with pytest.raises(TypeError):
+            device.callback  # noqa: B018
+        with pytest.raises(TypeError):
+            device.callback = lambda _device, _stream: None
+        with contextlib.closing(tcod.sdl.audio.BasicMixer(device)) as mixer:
+            assert mixer.daemon
+            assert mixer.play(np.zeros(4, np.float32)).busy
+            mixer.play(np.zeros(0, np.float32))
+            mixer.play(np.full(1, 0.01, np.float32), on_end=lambda _: None)
+            mixer.play(np.full(1, 0.01, np.float32), loops=2, on_end=lambda _: None)
+            mixer.play(np.full(4, 0.01, np.float32), loops=2).stop()
+            mixer.play(np.full(100000, 0.01, np.float32))
+            with pytest.raises(TypeError, match=r".*must be dtype=float32.*was dtype=int32"):
+                mixer.play(np.zeros(1, np.int32))
+            time.sleep(0.001)
+            mixer.stop()
+
+
+def test_audio_capture() -> None:
+    with tcod.sdl.audio.open(capture=True) as device:
+        assert not device.stopped
+        assert isinstance(device.dequeue_audio(), np.ndarray)
+
+
+def test_audio_device_repr() -> None:
+    with tcod.sdl.audio.open(format=np.uint16, paused=True, callback=True) as device:
+        assert "silence=" in repr(device)
+        assert "callback=" in repr(device)
+    assert "stopped=" in repr(device)
+
+
+def test_convert_bad_shape() -> None:
+    with pytest.raises(TypeError):
+        tcod.sdl.audio.convert_audio(
+            np.zeros((1, 1, 1), np.float32), 8000, out_rate=8000, out_format=np.float32, out_channels=1
+        )
+
+
+def test_convert_bad_type() -> None:
+    with pytest.raises(TypeError, match=r".*bool"):
+        tcod.sdl.audio.convert_audio(np.zeros(8, bool), 8000, out_rate=8000, out_format=np.float32, out_channels=1)
+    with pytest.raises(RuntimeError, match=r"Invalid source format"):
+        tcod.sdl.audio.convert_audio(np.zeros(8, np.int64), 8000, out_rate=8000, out_format=np.float32, out_channels=1)
+
+
+def test_convert_float64() -> None:
+    np.testing.assert_array_equal(
+        tcod.sdl.audio.convert_audio(
+            np.ones(8, np.float64), 8000, out_rate=8000, out_format=np.float32, out_channels=1
+        ),
+        np.ones((8, 1), np.float32),
+    )
+
+
+def test_audio_callback() -> None:
+    class CheckCalled:
+        was_called: bool = False
+
+        def __call__(self, device: tcod.sdl.audio.AudioDevice, stream: NDArray[Any]) -> None:
+            self.was_called = True
+            assert isinstance(device, tcod.sdl.audio.AudioDevice)
+            assert isinstance(stream, np.ndarray)
+            assert len(stream.shape) == 2  # noqa: PLR2004
+
+    check_called = CheckCalled()
+    with tcod.sdl.audio.open(callback=check_called, paused=False) as device:
+        device.callback = device.callback
+        while not check_called.was_called:
+            time.sleep(0.001)
+
+
+@pytest.mark.skipif(sys.version_info < (3, 8), reason="Needs sys.unraisablehook support")
+@pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
+def test_audio_callback_unraisable() -> None:
+    """Test unraisable error in audio callback.
+
+    This can't be checked with pytest very well, so at least make sure this doesn't crash.
+    """
+
+    class CheckCalled:
+        was_called: bool = False
+
+        def __call__(self, device: tcod.sdl.audio.AudioDevice, stream: NDArray[Any]) -> None:
+            self.was_called = True
+            raise Exception("Test unraisable error")  # noqa
+
+    check_called = CheckCalled()
+    with tcod.sdl.audio.open(callback=check_called, paused=False):
+        while not check_called.was_called:
+            time.sleep(0.001)