Aya / tests /test_support.py
Ransom070's picture
Aya: a Reachy Mini that sings songs written on the 0G Compute Network
5ac2df3 verified
Raw History Blame Contribute Delete
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 ---------------------------------------------------------------------
@pytest.mark.parametrize(
("name", "root", "intervals"),
[
("C", 0, (0, 4, 7)),
("Am", 9, (0, 3, 7)),
("F#m7", 6, (0, 3, 7, 10)),
("Bb", 10, (0, 4, 7)),
("Gsus4", 7, (0, 5, 7)),
("Edim", 4, (0, 3, 6)),
("Dmaj7", 2, (0, 4, 7, 11)),
],
)
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)
@pytest.mark.parametrize(
"request_text",
["Twinkle Twinkle Little Star", "sing me twinkle twinkle little star", "twinkle, twinkle, little star please"],
)
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
@pytest.mark.parametrize("request_text", ["a song about rain", "something in the style of Burna Boy", "", "a star"])
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