File size: 16,326 Bytes
0efd0f1 | 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 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 | """Two named synthetic users, return turns, background speech and an interruption."""
import argparse,asyncio,json,struct,time,base64
from pathlib import Path
import aiohttp
import numpy as np
from speaker_v4.common import atomic
from .live_test import audio
from .playback import SyntheticPlayback
async def main(extended=False,background_gain=.16,background_shift=0.,max_repeats=0):
if max_repeats not in (0,1):raise ValueError('QA permits either strict first pass or one explicit repeat per turn')
protocol=4 if max_repeats else 3
atomic(Path('/mnt/nvme/robo-speaker-studies/speaker-live-results')/('three-user-live.json' if extended else 'multi-user-live.json'),
{'public_synthetic_only':True,'protocol_version':protocol,'passed':False,'running':True})
bg=audio('background');a=audio('christoph-intro');b=audio('robert-intro');follow=audio('christoph-followup')
c=audio('selma-intro') if extended else None
follow_b=audio('robert-followup') if extended else None;follow_c=audio('selma-followup-clear') if extended else None
events=[];identities=[];counts={};sample=0;turn_done=asyncio.Event();first_chunk=asyncio.Event();t0=time.monotonic();boundaries=[];negative_rejected=None
completed=[];last_reply_name=None;repeats=[]
dac=SyntheticPlayback()
trace=Path('/mnt/nvme/robo-speaker-studies/speaker-live-results')/f'inflight-{int(time.time())}.jsonl'
def journal(item):
with trace.open('a') as output:output.write(json.dumps({'at_s':round(time.monotonic()-t0,3),**item},ensure_ascii=False)+'\n')
journal({'type':'protocol','public_synthetic_only':True,'extended':extended,'background_shift_s':background_shift,
'protocol_version':protocol,'max_repeats_per_turn':max_repeats})
async with aiohttp.ClientSession() as client:
for _ in range(90):
try:
async with client.get('http://127.0.0.1:8796/state') as response:
diar_ready=(await response.json()).get('ready')
async with client.get('http://127.0.0.1:8793/api/state') as response:
if diar_ready and (await response.json()).get('ready'):break
except aiohttp.ClientError:pass
await asyncio.sleep(.5)
else:raise TimeoutError('Robby/diarizer did not become ready; conversation test was not started')
async with client.ws_connect('http://127.0.0.1:8793/ws?synthetic_diagnostics=1') as ws:
t0=time.monotonic()
async def receive():
nonlocal last_reply_name
async for message in ws:
if message.type==aiohttp.WSMsgType.BINARY:
rid=struct.unpack_from('<i',message.data)[0]
if rid:
count=(len(message.data)-4)//2;counts[rid]=counts.get(rid,0)+count
dac.add(rid,count,sample)
elif message.type==aiohttp.WSMsgType.TEXT:
item=json.loads(message.data)
if item.get('type')=='synthetic_audio_debug':
# Only this loopback client's public synthetic
# fixtures; ordinary/browser sessions cannot emit it.
import soundfile as sf
folder=Path('/mnt/nvme/robo-speaker-studies/speaker-live-results/synthetic-debug');folder.mkdir(exist_ok=True)
path=folder/f'{int(t0)}-{item["start"]:.3f}.wav'
sf.write(path,np.frombuffer(base64.b64decode(item.pop('audio')),'<f2').astype(np.float32),16000)
events.append({**item,'file':str(path)})
if item.get('type') in ('endpoint','reply','barge','turn_done','warning','error','ignored','asr_preview','synthetic_asr_views','synthetic_barge_proposal','synthetic_barge_verification'):
journal(item)
elif item.get('type')=='identity' and any(k in item for k in ('decision','verification','continuity')):
journal({k:v for k,v in item.items() if k in ('type','decision','verification','continuity','speakers','users')})
if item.get('type') in ('endpoint','reply','barge','turn_done','warning','error','asr_preview'):events.append({'at_s':round(time.monotonic()-t0,3),**item})
if item.get('type') in ('ignored','warning','error'):
events.append(item);print('Synthetic diagnostic:',json.dumps(item),flush=True)
if item.get('type')=='identity' and item.get('speakers'):identities.append(item['speakers'])
if item.get('type')=='identity' and item.get('verification'):
events.append({'at_s':round(time.monotonic()-t0,3),'type':'verification',**item['verification']})
if item.get('type')=='identity' and item.get('continuity'):
events.append({'at_s':round(time.monotonic()-t0,3),'type':'continuity',**item['continuity']})
if item.get('type')=='synthetic_asr_views':events.append(item)
if item.get('type') in ('synthetic_barge_proposal','synthetic_barge_verification'):events.append({'at_s':round(time.monotonic()-t0,3),**item})
if item.get('type')=='reply':last_reply_name=item.get('speaker_name')
if item.get('type')=='barge':dac.cancel(sample)
if item.get('type')=='turn_done' and not item.get('barged'):
completed.append({'at_s':time.monotonic()-t0,'name':last_reply_name});turn_done.set()
task=asyncio.create_task(receive())
async def feed(wave=None,seconds=None,fixture=None):
nonlocal sample
n=len(wave) if wave is not None else round(seconds*16000)
if wave is not None:boundaries.append({'fixture':fixture or ('christoph-intro' if wave is a else 'robert-intro' if wave is b else 'selma-intro' if wave is c else 'robert-followup' if wave is follow_b else 'selma-followup-clear' if wave is follow_c else 'christoph-followup'),'start_s':sample/16000,'end_s':(sample+n)/16000})
for at in range(0,n,320):
length=min(320,n-at);indices=(np.arange(length)+sample+round(background_shift*16000))%len(bg);x=bg[indices]*background_gain
if wave is not None:x=x+wave[at:at+length]
pcm=np.clip(np.rint(x*32767),-32768,32767).astype('<i2')
if ws.closed:raise RuntimeError('WebSocket closed during synthetic conversation')
for lo,hi,rid,pos in dac.pieces(sample,length):
await ws.send_bytes(struct.pack('<ii',rid,pos)+pcm[lo:hi].tobytes())
if rid:first_chunk.set()
sample+=length
if sample//320%5==0:
for rid,played in dac.acknowledgments(sample).items():await ws.send_str(f'played:{rid}:{played}')
await asyncio.sleep(max(0,t0+sample/16000-time.monotonic()))
async def wait_reply(name):
target=boundaries[-1].copy();repeated=0
for _ in range(260):
if any(e['name']==name and e['at_s']>=target['start_s'] for e in completed):return
# Optional recovery test, NOT a strict first-pass claim.
# Repeat the same public utterance once, only if no turn
# was accepted after nine seconds. Never repeat an already
# accepted turn just because shared TTS is queued/busy.
accepted=any(e.get('type')=='endpoint' and e.get('speaker_name')==name
and e['at_s']>=target['start_s'] for e in events)
if repeated<max_repeats and not accepted and sample/16000>=target['end_s']+9.:
reasons=[e.get('reason',e.get('text','')) for e in events if e.get('type') in ('ignored','warning')
and e.get('at_s',0)>=target['start_s']]
repeats.append({'name':name,'fixture':target['fixture'],'first_start_s':target['start_s'],
'repeat_start_s':sample/16000,'reason':reasons[-1] if reasons else 'No accepted turn within nine seconds'})
journal({'type':'explicit_test_repeat',**repeats[-1]})
repeated+=1;await feed(audio(target['fixture']),fixture=target['fixture'])
await feed(seconds=.32)
atomic(Path('/mnt/nvme/robo-speaker-studies/speaker-live-results')/('multi-timeout-'+str(int(time.time()))+'.json'),{'events':events,'identities':identities})
raise TimeoutError('Reply did not finish while background continued')
try:
await feed(seconds=1.28)
await feed(a);await wait_reply('Christoph');turn_done.clear();first_chunk.clear()
await feed(b);await wait_reply('Robert');turn_done.clear();first_chunk.clear()
if extended:
await feed(c);await wait_reply('Selma');turn_done.clear();first_chunk.clear()
# The original Selma follow-up failed clean ASR (nonverbal
# synthesis instead of its intended words). Retain it as
# a negative: do not submit background TV as Selma's text.
before=sum(e.get('type')=='endpoint' for e in events)
await feed(audio('selma-followup'),fixture='selma-followup-nonverbal');await feed(seconds=3.84)
negative_rejected=sum(e.get('type')=='endpoint' for e in events)==before
if not negative_rejected:
atomic(Path('/mnt/nvme/robo-speaker-studies/speaker-live-results')/('negative-failure-'+str(int(time.time()))+'.json'),
{'public_synthetic_only':True,'events':events,'identities':identities,'fixture_boundaries':boundaries})
assert negative_rejected,'Nonverbal foreground submitted background words'
await feed(follow);await wait_reply('Christoph');turn_done.clear();first_chunk.clear()
await feed(follow_b);await wait_reply('Robert');turn_done.clear();first_chunk.clear()
last=follow_c if extended else follow
await feed(last)
for _ in range(100):
if first_chunk.is_set():break
await feed(seconds=.32)
# The same enrolled person talks over the actual response.
await feed(last)
await feed(seconds=2.56)
natural_barges=[e for e in events if e.get('type')=='barge']
await ws.send_str('interrupt')
named={'Christoph','Robert','Selma'}
all_endpoints=[e for e in events if e.get('type')=='endpoint'];all_replies=[e for e in events if e.get('type')=='reply']
endpoints=[e for e in all_endpoints if e.get('speaker_name') in named];replies=[e for e in all_replies if e.get('speaker_name') in named]
report={'protocol_version':protocol,'strict_content_checks':True,'max_repeats_per_turn':max_repeats,
'explicit_repeats':repeats,'all_first_attempts_accepted':not repeats,
'public_synthetic_only':True,'background_continues':True,'background_gain':background_gain,'background_shift_s':background_shift,'extended':extended,'endpoints':endpoints,'replies':replies,
'provisional_wake_endpoints':[e for e in all_endpoints if e not in endpoints],
'provisional_replies':[e for e in all_replies if e not in replies],'completed_replies':completed,
'barge_events':natural_barges,'last_enrolled_users':identities[-1] if identities else [],
'nonverbal_foreground_rejected':negative_rejected,
'warnings':[e for e in events if e.get('type') in ('warning','error')],
'asr_previews':[e for e in events if e.get('type')=='asr_preview'],
'verification':[e for e in events if e.get('type')=='verification'],
'continuity':[e for e in events if e.get('type')=='continuity'],
'synthetic_audio_debug':[e for e in events if e.get('type')=='synthetic_audio_debug'],
'asr_views':[e for e in events if e.get('type')=='synthetic_asr_views'],'fixture_boundaries':boundaries,
'barge_verification':[e for e in events if e.get('type')=='synthetic_barge_verification'],
'playback':'Real-time synthetic DAC position tags and acknowledgments, including chunk gaps/cancellation; no physical microphone/DAC'}
folder=Path('/mnt/nvme/robo-speaker-studies/speaker-live-results')
atomic(folder/('multi-attempt-'+str(int(time.time()))+'.json'),report)
print('Synthetic multi-user results:',json.dumps({'endpoints':[(e.get('speaker_name'),e.get('speaker_id')) for e in endpoints],
'reply_count':len(replies),'barge_count':len(report['barge_events']),'warnings':report['warnings']}),flush=True)
expected=['Christoph','Robert','Selma','Christoph','Robert','Selma'] if extended else ['Christoph','Robert','Christoph']
assert len(endpoints)>=len(expected),'Did not submit all named users'
assert len(replies)>=len(expected),'An expected user received no reply (a partial zip is not a pass)'
assert [e.get('speaker_name') for e in endpoints[:len(expected)]]==expected,'Names not bound to returning voice'
bound={}
for name,item in zip(expected,endpoints):
if name in bound:assert bound[name]==item['speaker_id'],'Returning speaker ID changed'
else:bound[name]=item['speaker_id']
assert len(set(bound.values()))==len(bound),'Distinct people merged'
required=['Christoph','Robert','Selma','Geschichte','Witz','zusammen'] if extended else ['Christoph','Robert','Geschichte']
for word,item in zip(required,endpoints):assert word.casefold() in item['transcript'].casefold(),'Wrong/background words submitted in place of a target turn'
assert not any(phrase in item['transcript'].casefold() for item in endpoints
for phrase in ('nachrichten','morgen wird','bericht','über das wetter')),'Background-TV words were appended to an accepted user turn'
assert len(report['last_enrolled_users'])==len(set(expected)),'Background speaker enrolled without wake'
assert all(name in reply.get('text','') for name,reply in zip(expected,replies[:len(expected)])),'Reply did not address current user'
assert report['barge_events'],'Enrolled interruption was lost'
last_user=boundaries[-1]
assert any(last_user['start_s']<=e['at_s']<=last_user['end_s']+1 for e in natural_barges),'No actual voice interruption during target fixture; cleanup Stop does not count'
report['passed']=True
atomic(folder/('three-user-live.json' if extended else 'multi-user-live.json'),report)
print(json.dumps({'passed':True,'extended':extended,'users':list(bound),'turns':len(endpoints),'replies':len(replies),'barge_events':len(report['barge_events'])}),flush=True)
finally:
await ws.close();task.cancel();await asyncio.gather(task,return_exceptions=True)
if __name__=='__main__':
p=argparse.ArgumentParser();p.add_argument('--extended',action='store_true');p.add_argument('--background-gain',type=float,default=.16);p.add_argument('--background-shift',type=float,default=0.)
p.add_argument('--max-repeats',type=int,choices=(0,1),default=0,help='Default 0: strict first pass. 1: separately labelled bounded-recovery test.')
args=p.parse_args();asyncio.run(main(args.extended,args.background_gain,args.background_shift,args.max_repeats))
|