Grok Bot
138 lines
6.2 KiB
Python
138 lines
6.2 KiB
Python
"""Tests. Everything audio here is a synthetic TEST signal made locally, never pilot audio.
|
|
Run: python3 -m unittest discover -s _pipeline/tests -v (from the repo root)"""
|
|
import http.server
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
|
|
|
import acx_check
|
|
import assemble
|
|
import audiomeasure as am
|
|
import script
|
|
import settings
|
|
import voice
|
|
|
|
CFG = settings.load()
|
|
|
|
|
|
def test_piece(path, hz, seconds, level):
|
|
"""TEST signal: speech-like (harmonics, 4 Hz syllable rhythm, a little noise) at amplitude `level`."""
|
|
expr = "%g*(0.6*sin(2*PI*%g*t)+0.3*sin(4*PI*%g*t)+0.1*sin(6*PI*%g*t))*(0.55+0.45*sin(2*PI*4*t))" % (level, hz, hz, hz)
|
|
am.ffmpeg(["-f", "lavfi", "-i", "aevalsrc='%s':s=24000:d=%g" % (expr, seconds), "-ar", "24000", "-ac", "1", path])
|
|
|
|
|
|
class AudioTests(unittest.TestCase):
|
|
def test_assemble_then_check_passes(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
items = []
|
|
for i, (hz, lvl) in enumerate([(120, 0.02), (150, 0.05), (130, 0.03)]):
|
|
p = os.path.join(d, "t%d.wav" % i)
|
|
test_piece(p, hz, 4, lvl)
|
|
items.append({"wav": p, "gap_after_s": 0.7})
|
|
if i == 0:
|
|
items.append({"cue": "chime", "gap_after_s": 0.25})
|
|
out = os.path.join(d, "audio", "chapter-99.mp3")
|
|
gain = assemble.assemble_items(CFG, items, out)
|
|
m = acx_check.measure(CFG, out)
|
|
print("\nTEST assembled: gain %.1f dB, measured %s" % (gain, {k: (round(v, 2) if isinstance(v, float) else v) for k, v in m.items()}))
|
|
for name, value, passed in acx_check.judge(CFG, out, m):
|
|
self.assertTrue(passed, "%s %s" % (name, value))
|
|
|
|
def test_checker_fails_bad_files(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
raw = os.path.join(d, "raw.wav")
|
|
test_piece(raw, 130, 4, 1.4) # hot test signal, clips at full scale
|
|
bad = os.path.join(d, "bad.mp3") # stereo, 128k, 22.05 kHz, no room tone
|
|
am.ffmpeg(["-i", raw, "-ac", "2", "-ar", "22050", "-c:a", "libmp3lame", "-b:a", "128k", bad])
|
|
failed = {n for n, v, p in acx_check.judge(CFG, bad, acx_check.measure(CFG, bad)) if not p}
|
|
print("\nTEST bad file 1 fails on:", sorted(failed))
|
|
self.assertTrue({"format", "rms dB", "noise floor dB", "head s", "tail s"} <= failed)
|
|
hot = os.path.join(d, "hot.mp3") # correct format, but too loud
|
|
am.ffmpeg(["-i", raw, "-ac", "1", "-ar", "44100", "-c:a", "libmp3lame", "-b:a", "192k", hot])
|
|
m = acx_check.measure(CFG, hot)
|
|
failed = {n for n, v, p in acx_check.judge(CFG, hot, m) if not p}
|
|
print("TEST bad file 2 (peak %.1f dB) fails on:" % m["peak_db"], sorted(failed))
|
|
self.assertIn("peak dB", failed)
|
|
self.assertNotIn("format", failed)
|
|
|
|
|
|
class StubHandler(http.server.BaseHTTPRequestHandler):
|
|
seen = []
|
|
|
|
def do_POST(self):
|
|
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
StubHandler.seen.append((self.path, body))
|
|
if self.path.endswith("/audio/speech"):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = os.path.join(d, "x.wav")
|
|
test_piece(p, 130, 1, 0.1)
|
|
data = open(p, "rb").read()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "audio/wav")
|
|
else:
|
|
data = json.dumps({"choices": [{"message": {"content": "A test summary."}}]}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
|
|
class ClientTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.server = http.server.HTTPServer(("127.0.0.1", 0), StubHandler)
|
|
threading.Thread(target=cls.server.serve_forever, daemon=True).start()
|
|
cls.url = "http://127.0.0.1:%d" % cls.server.server_port
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
|
|
def cfg(self):
|
|
c = json.loads(json.dumps(CFG))
|
|
c["voice"]["endpoint"] = self.url + "/v1/audio/speech"
|
|
c["llm"].update(endpoint=self.url + "/v1/chat/completions", model="m")
|
|
return c
|
|
|
|
def test_voice_request_and_wav(self):
|
|
StubHandler.seen.clear()
|
|
data = voice.speak(self.cfg(), "Hello there.")
|
|
path, body = StubHandler.seen[0]
|
|
self.assertEqual(path, "/v1/audio/speech")
|
|
self.assertEqual(body["input"], "Hello there.")
|
|
self.assertEqual(body["response_format"], "wav")
|
|
self.assertEqual(data[:4], b"RIFF")
|
|
|
|
def test_sentence_split_keeps_abbreviations(self):
|
|
s = voice.sentences('One day Dr. Alexis Carrel sent word. "The world," he said. E. Stanley Jones came.')
|
|
self.assertEqual(len(s), 3)
|
|
|
|
def test_script_wraps_inserted_parts_when_switches_on(self):
|
|
c = self.cfg()
|
|
c["summaries"].update(chapter_preview=True, chapter_summary=True, concept_summary=True)
|
|
ch = {"number": 1, "title": "We Go Seeking", "switches": {}}
|
|
segs = script.chapter_script(c, {"summaries": {}}, ch, ["Para one is here.", "Para two is here.", "Para three is here."])
|
|
kinds = [s["kind"] for s in segs]
|
|
self.assertEqual(kinds[:5], ["header", "cue", "label", "chapter_preview", "cue"])
|
|
self.assertEqual(kinds[-4:], ["cue", "label", "chapter_summary", "cue"])
|
|
self.assertEqual(kinds.count("concept_summary"), 1)
|
|
self.assertEqual([s["text"] for s in segs if s["kind"] == "author"], ["Para one is here.", "Para two is here.", "Para three is here."])
|
|
|
|
def test_switches_default_off_and_chapter_wins(self):
|
|
ch = {"number": 1, "title": "x", "switches": {"chapter_summary": True}}
|
|
self.assertFalse(script.switch(CFG, {"summaries": {}}, {"switches": {}}, "chapter_summary"))
|
|
self.assertTrue(script.switch(CFG, {"summaries": {"chapter_summary": False}}, ch, "chapter_summary"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|