tests/test_console_audio.py

from __future__ import annotations

import asyncio
from types import SimpleNamespace
from unittest.mock import MagicMock

import numpy as np
import pytest

from peachy.console import LocalStream


class _Handler:
    def __init__(self, output: tuple[int, np.ndarray]) -> None:
        self.output = output
        self.output_queue: asyncio.Queue[object] = asyncio.Queue()
        self.last_activity_time = 0.0

    def set_activity_observer(self, _observer: object) -> None:
        pass

    async def emit(self) -> tuple[int, np.ndarray]:
        return self.output


@pytest.mark.asyncio
@pytest.mark.parametrize(("provider_rate", "source_samples"), [(16_000, 160), (24_000, 240)])
async def test_play_loop_resamples_provider_audio_to_robot_rate(
    provider_rate: int,
    source_samples: int,
) -> None:
    pushed: list[np.ndarray] = []
    media = SimpleNamespace(
        audio=None,
        get_output_audio_samplerate=lambda: 48_000,
        push_audio_sample=lambda frame: pushed.append(np.asarray(frame)),
    )
    stream = LocalStream(
        _Handler((provider_rate, np.ones((1, source_samples), dtype=np.int16))),
        SimpleNamespace(media=media),
    )

    media.push_audio_sample = lambda frame: (
        pushed.append(np.asarray(frame)),
        stream._stop_event.set(),
    )
    await stream.play_loop()

    assert len(pushed) == 1
    assert pushed[0].dtype == np.float32
    assert pushed[0].shape == (480,)


@pytest.mark.asyncio
async def test_play_loop_uses_the_rate_from_each_tuple() -> None:
    media = SimpleNamespace(
        audio=None,
        get_output_audio_samplerate=MagicMock(return_value=24_000),
        push_audio_sample=MagicMock(),
    )
    stream = LocalStream(
        _Handler((48_000, np.ones(480, dtype=np.int16))),
        SimpleNamespace(media=media),
    )
    media.push_audio_sample.side_effect = lambda _frame: stream._stop_event.set()

    await stream.play_loop()

    pushed = media.push_audio_sample.call_args.args[0]
    assert pushed.shape == (240,)
    media.get_output_audio_samplerate.assert_called_once_with()