Spaces:
Running
Running
File size: 2,087 Bytes
8a652b8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | 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()
|