Download tests/test_support.py from Ransom070/Aya: direct link, hf CLI and curl.
- Browser
- Download file 12.9 kB
-
https://huggingface.co/spaces/Ransom070/Aya/resolve/main/tests/test_support.py
- Command line
-
hf download hf://spaces/Ransom070/Aya/tests/test_support.py
-
curl -L -o test_support.py https://huggingface.co/spaces/Ransom070/Aya/resolve/main/tests/test_support.py
12.9 kB
| """The band, the repertoire, the library, the listener and the configuration.""" | |
| from __future__ import annotations | |
| import json | |
| import time | |
| from pathlib import Path | |
| import httpx | |
| import numpy as np | |
| import pytest | |
| from aya import repertoire | |
| from aya.band import parse_chord, render_band | |
| from aya.score import Score, parse_score | |
| from aya.config import DEFAULT_DAEMON_URL, Config | |
| from aya.listen import MIN_RECORDING_S, DEFLECTION_THRESHOLD_RAD, AntennaButton, VoiceRequests | |
| from aya.library import Library, score_to_dict | |
| from tests.conftest import FakeMedia, FakeVoice | |
| from aya.performance import render_song | |
| # -- band --------------------------------------------------------------------- | |
| def test_reads_the_chord_names_a_songwriter_writes(name: str, root: int, intervals: tuple[int, ...]) -> None: | |
| assert parse_chord(name) == (root, intervals) | |
| def test_reads_unicode_accidentals() -> None: | |
| assert parse_chord("A♭maj7") == (8, (0, 4, 7, 11)) | |
| def test_an_unknown_quality_falls_back_to_a_major_triad() -> None: | |
| # Wrong far less often than falling back to silence. | |
| assert parse_chord("Cwhatever") == (0, (0, 4, 7)) | |
| def test_a_name_that_is_not_a_chord_is_refused() -> None: | |
| assert parse_chord("hello") is None | |
| assert parse_chord("") is None | |
| def test_the_band_plays_for_the_length_of_the_song(song_score: Score) -> None: | |
| band = render_band(song_score, 22050) | |
| assert band.size >= song_score.duration_seconds * 22050 * 0.9 | |
| assert float(np.max(np.abs(band))) <= 0.51 | |
| def test_the_band_still_keeps_time_for_a_song_with_no_chords(song_score: Score) -> None: | |
| from dataclasses import replace | |
| band = render_band(replace(song_score, chords=()), 22050) | |
| # No harmony, but the beat pulse still marks out the grid for the dance. | |
| assert float(np.max(np.abs(band))) > 0.0 | |
| # -- repertoire --------------------------------------------------------------- | |
| def test_every_bundled_song_is_internally_consistent() -> None: | |
| for song in repertoire.REPERTOIRE: | |
| assert song.notes | |
| assert song.credit == "traditional" | |
| # The harmony must cover the tune, or the song ends unaccompanied. | |
| assert sum(chord.beats for chord in song.chords) >= song.duration_beats | |
| assert song.motion | |
| def test_every_bundled_song_survives_a_round_trip_through_the_parser() -> None: | |
| for song in repertoire.REPERTOIRE: | |
| reparsed = parse_score(score_to_dict(song)) | |
| assert reparsed.title == song.title | |
| assert len(reparsed.notes) == len(song.notes) | |
| def test_recognises_a_song_it_knows(request_text: str) -> None: | |
| assert repertoire.find(request_text) is repertoire.TWINKLE | |
| def test_recognises_a_song_asked_for_loosely() -> None: | |
| assert repertoire.find("can you sing row row row your boat") is repertoire.ROW_YOUR_BOAT | |
| def test_does_not_hijack_a_request_for_something_new(request_text: str) -> None: | |
| # A false positive means ignoring what was actually asked for, which is worse than | |
| # spending twenty seconds writing something. | |
| assert repertoire.find(request_text) is None | |
| def test_there_is_always_something_to_fall_back_on() -> None: | |
| assert repertoire.fallback() in repertoire.REPERTOIRE | |
| # -- library ------------------------------------------------------------------ | |
| def test_a_score_survives_being_stored_and_read_back(tmp_path: Path, song_score: Score) -> None: | |
| library = Library(tmp_path) | |
| key = library.key_for("a test song") | |
| library.save_score(key, song_score) | |
| restored = library.load_score(key) | |
| assert restored is not None | |
| assert restored.title == song_score.title | |
| assert restored.lyrics == song_score.lyrics | |
| assert restored.request == song_score.request | |
| assert len(restored.notes) == len(song_score.notes) | |
| assert [chord.root for chord in restored.chords] == [chord.root for chord in song_score.chords] | |
| assert [cue.gesture for cue in restored.motion] == [cue.gesture for cue in song_score.motion] | |
| def test_audio_survives_being_stored_and_read_back(tmp_path: Path, voice: FakeVoice, song_score: Score) -> None: | |
| library = Library(tmp_path) | |
| key = library.key_for("a test song") | |
| song = render_song(song_score, voice, with_band=False) | |
| library.save_audio(key, song) | |
| restored = library.load_audio(key, song_score) | |
| assert restored is not None | |
| assert restored.sample_rate == song.sample_rate | |
| assert restored.audio.size == song.audio.size | |
| assert np.allclose(restored.audio, song.audio, atol=1e-3) | |
| def test_the_same_request_finds_the_same_song_however_it_was_typed(tmp_path: Path) -> None: | |
| library = Library(tmp_path) | |
| assert library.key_for("A Song About Rain") == library.key_for(" a song about rain ") | |
| assert library.key_for("a song", lyrics="one") != library.key_for("a song", lyrics="two") | |
| def test_asking_for_a_song_that_was_never_stored_returns_nothing(tmp_path: Path, song_score: Score) -> None: | |
| library = Library(tmp_path) | |
| assert library.load_score("missing") is None | |
| assert library.load_audio("missing", song_score) is None | |
| def test_a_damaged_score_is_discarded_rather_than_crashing(tmp_path: Path) -> None: | |
| library = Library(tmp_path) | |
| tmp_path.mkdir(exist_ok=True) | |
| library.score_path("broken").write_text("{not json") | |
| assert library.load_score("broken") is None | |
| def test_a_damaged_index_starts_a_new_one(tmp_path: Path, song_score: Score) -> None: | |
| library = Library(tmp_path) | |
| tmp_path.mkdir(exist_ok=True) | |
| (tmp_path / "index.json").write_text("nonsense") | |
| library.save_score(library.key_for("x"), song_score) | |
| assert len(library.entries()) == 1 | |
| def test_the_catalogue_lists_the_newest_first(tmp_path: Path, song_score: Score) -> None: | |
| from dataclasses import replace | |
| library = Library(tmp_path) | |
| library.save_score(library.key_for("first"), replace(song_score, title="First")) | |
| library.save_score(library.key_for("second"), replace(song_score, title="Second")) | |
| assert [entry["title"] for entry in library.entries()] == ["Second", "First"] | |
| def test_forgetting_a_song_removes_it_everywhere(tmp_path: Path, voice: FakeVoice, song_score: Score) -> None: | |
| library = Library(tmp_path) | |
| key = library.key_for("a test song") | |
| library.save_score(key, song_score) | |
| library.save_audio(key, render_song(song_score, voice, with_band=False)) | |
| assert library.forget(key) is True | |
| assert library.load_score(key) is None | |
| assert not library.audio_path(key).exists() | |
| assert library.entries() == [] | |
| assert library.forget(key) is False | |
| # -- listening ---------------------------------------------------------------- | |
| def antenna_client(positions: list[list[float]]) -> httpx.Client: | |
| """A stub daemon that reports the given antenna positions in turn.""" | |
| readings = list(positions) | |
| def handler(request: httpx.Request) -> httpx.Response: | |
| return httpx.Response(200, json=readings.pop(0) if len(readings) > 1 else readings[0]) | |
| return httpx.Client(transport=httpx.MockTransport(handler)) | |
| def test_the_antenna_button_learns_its_resting_position_first() -> None: | |
| button = AntennaButton("http://robot", antenna_client([[0.1, 0.1]])) | |
| # The first reading defines rest, so it can never itself count as a press. | |
| assert button.poll() is False | |
| def test_bending_an_antenna_presses_the_button() -> None: | |
| push = DEFLECTION_THRESHOLD_RAD + 0.2 | |
| button = AntennaButton("http://robot", antenna_client([[0.0, 0.0], [push, 0.0]])) | |
| button.poll() | |
| button._last_poll = 0.0 | |
| assert button.poll() is True | |
| def test_motor_jitter_does_not_press_the_button() -> None: | |
| button = AntennaButton("http://robot", antenna_client([[0.0, 0.0], [DEFLECTION_THRESHOLD_RAD - 0.05, 0.0]])) | |
| button.poll() | |
| button._last_poll = 0.0 | |
| assert button.poll() is False | |
| def test_an_unreachable_daemon_disables_the_button_without_failing() -> None: | |
| def refuse(request: httpx.Request) -> httpx.Response: | |
| raise httpx.ConnectError("no daemon", request=request) | |
| button = AntennaButton("http://robot", httpx.Client(transport=httpx.MockTransport(refuse))) | |
| assert button.poll() is False | |
| assert button.available is False | |
| def test_a_recorded_request_comes_back_as_pcm(media: FakeMedia) -> None: | |
| push = DEFLECTION_THRESHOLD_RAD + 0.2 | |
| listener = VoiceRequests("http://robot", client=antenna_client([[0.0, 0.0], [push, 0.0], [push, 0.0], [0.0, 0.0]])) | |
| assert listener.update(media) is None # learns rest | |
| listener.button._last_poll = 0.0 | |
| media.microphone = [np.full(media.input_rate, 0.5, dtype=np.float32)] | |
| assert listener.update(media) is None # starts recording | |
| assert listener.recording is True | |
| listener.button._last_poll = 0.0 | |
| media.microphone = [np.full(media.input_rate, 0.5, dtype=np.float32)] | |
| listener.update(media) | |
| listener.button._last_poll = 0.0 | |
| captured = listener.update(media) | |
| assert captured is not None | |
| assert captured.dtype == np.int16 | |
| assert captured.size >= media.input_rate * MIN_RECORDING_S | |
| assert media.recording is False | |
| def test_a_graze_is_not_treated_as_a_request(media: FakeMedia) -> None: | |
| push = DEFLECTION_THRESHOLD_RAD + 0.2 | |
| listener = VoiceRequests("http://robot", client=antenna_client([[0.0, 0.0], [push, 0.0], [0.0, 0.0]])) | |
| listener.update(media) | |
| listener.button._last_poll = 0.0 | |
| media.microphone = [np.zeros(int(media.input_rate * 0.1), dtype=np.float32)] | |
| listener.update(media) | |
| listener.button._last_poll = 0.0 | |
| assert listener.update(media) is None | |
| def test_a_media_backend_without_a_microphone_does_not_break_listening() -> None: | |
| class Deaf: | |
| pass | |
| push = DEFLECTION_THRESHOLD_RAD + 0.2 | |
| listener = VoiceRequests("http://robot", client=antenna_client([[0.0, 0.0], [push, 0.0], [0.0, 0.0]])) | |
| listener.update(Deaf()) | |
| listener.button._last_poll = 0.0 | |
| assert listener.update(Deaf()) is None | |
| def test_a_recording_that_never_ends_is_cut_off(media: FakeMedia, monkeypatch: pytest.MonkeyPatch) -> None: | |
| push = DEFLECTION_THRESHOLD_RAD + 0.2 | |
| listener = VoiceRequests("http://robot", client=antenna_client([[0.0, 0.0], [push, 0.0]])) | |
| listener.update(media) | |
| listener.button._last_poll = 0.0 | |
| media.microphone = [np.full(media.input_rate * 2, 0.5, dtype=np.float32)] | |
| listener.update(media) | |
| # Pretend the antenna has been held far past the limit. | |
| listener._started = time.monotonic() - 999.0 | |
| listener.button._last_poll = 0.0 | |
| assert listener.update(media) is not None | |
| # -- configuration ------------------------------------------------------------ | |
| def test_configuration_works_with_an_empty_environment(monkeypatch: pytest.MonkeyPatch) -> None: | |
| for name in list(__import__("os").environ): | |
| if name.startswith("AYA_"): | |
| monkeypatch.delenv(name) | |
| config = Config.from_environment() | |
| assert config.daemon_url == DEFAULT_DAEMON_URL | |
| assert config.with_band is True | |
| assert config.songwriter.model | |
| def test_configuration_reads_the_environment(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("AYA_DAEMON_URL", "http://reachy-mini.local:8000") | |
| monkeypatch.setenv("AYA_SETTINGS_PORT", "9000") | |
| monkeypatch.setenv("AYA_PIPER_VOICE", "~/voices/en.onnx") | |
| monkeypatch.setenv("AYA_WITH_BAND", "no") | |
| monkeypatch.setenv("AYA_SONGWRITER_MODEL", "some-other-model") | |
| config = Config.from_environment() | |
| assert config.daemon_url == "http://reachy-mini.local:8000" | |
| assert config.settings_port == 9000 | |
| assert config.piper_voice is not None and config.piper_voice.is_absolute() | |
| assert config.with_band is False | |
| assert config.songwriter.model == "some-other-model" | |
| # Overriding only the model must keep the rest of the provider intact. | |
| assert config.songwriter.address.startswith("0x") | |
| def test_an_unreadable_number_falls_back_instead_of_failing(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("AYA_SETTINGS_PORT", "not a port") | |
| assert Config.from_environment().settings_port == 8042 | |
| def test_a_score_serialises_to_the_shape_the_songwriter_produces(song_score: Score) -> None: | |
| payload = score_to_dict(song_score) | |
| assert json.loads(json.dumps(payload)) | |
| assert parse_score(payload).title == song_score.title | |