72 lines
3.1 KiB
Python
72 lines
3.1 KiB
Python
"""Real Piper playback test against isolated room-state fixtures. Uses free loopback ports."""
|
|
import array
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
import wave
|
|
import teddy_speech as speech
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
STATE = {"run_id":"speech-smoke", "teddy":{"online":True,"muted":False,"takeover":False,"current":None}}
|
|
class Fixture(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
data=json.dumps({"state":STATE}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Length",str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
def log_message(self,*_): pass
|
|
def wait(status,port,seconds=60):
|
|
until=time.monotonic()+seconds
|
|
while time.monotonic()<until:
|
|
try:
|
|
with urllib.request.urlopen(f"http://127.0.0.1:{port}/health",timeout=1) as response:
|
|
if json.load(response)["status"]==status: return
|
|
except OSError: pass
|
|
time.sleep(.2)
|
|
raise AssertionError("Timed out waiting for audio: "+status)
|
|
def run():
|
|
import socket
|
|
fixture=ThreadingHTTPServer(("127.0.0.1",0),Fixture)
|
|
threading.Thread(target=fixture.serve_forever,daemon=True).start()
|
|
with socket.socket() as sock:
|
|
sock.bind(("127.0.0.1",0))
|
|
status_port=sock.getsockname()[1]
|
|
with (speech.DATA/"smoke.log").open("w",encoding="utf-8") as log:
|
|
process=subprocess.Popen([str(speech.python_path()),str(ROOT/"teddy_speech.py"),"follow",
|
|
"--backend","piper","--room-port",str(fixture.server_port),"--status-port",str(status_port)],cwd=ROOT,
|
|
stdout=log,stderr=log,creationflags=subprocess.CREATE_NO_WINDOW if os.name=="nt" else 0)
|
|
try:
|
|
wait("idle",status_port)
|
|
cue={"id":"welcome","started":0,"transcript":"Welcome to Glitch, the Dark Machine. I am Teddy. Let us bring these arcade machines back to life."}
|
|
STATE["teddy"]["current"]=cue
|
|
wait("playing",status_port)
|
|
print("Piper is playing the test welcome cue.",flush=True)
|
|
time.sleep(2)
|
|
STATE["teddy"]["muted"]=True
|
|
wait("idle",status_port)
|
|
STATE["teddy"]["muted"]=False
|
|
STATE["teddy"]["current"]={**cue,"id":"replay"}
|
|
wait("playing",status_port)
|
|
STATE["teddy"]["current"]=None
|
|
wait("idle",status_port)
|
|
wavs=list((speech.DATA/"cache").glob("*.wav"))
|
|
with wave.open(str(max(wavs,key=lambda p:p.stat().st_mtime))) as wav:
|
|
samples=array.array("h",wav.readframes(wav.getnframes()))
|
|
assert wav.getnframes()>22050 and max(abs(x) for x in samples)>100
|
|
print(f"Verified non-silent PCM WAV: {wav.getframerate()} Hz; {wav.getnframes()/wav.getframerate():.1f}s")
|
|
print("PASS: room-state fixture -> real Piper -> local playback; mute/replay/stop")
|
|
finally:
|
|
process.terminate()
|
|
process.wait(timeout=5)
|
|
fixture.shutdown()
|
|
fixture.server_close()
|
|
if __name__=="__main__":
|
|
run()
|
|
|