"""L8: real baseline SpeechManager/provider, real hello capabilities, LAN device."""
import asyncio,base64,hashlib,json,os,pathlib,shlex,sys,time
from types import SimpleNamespace
from contextlib import asynccontextmanager
sys.path.insert(0,str(pathlib.Path.cwd()/'server'))
from device_gateway import speech
root=pathlib.Path('.cache/speech-diagnostics')
for line in (pathlib.Path.home()/'.config/esp32-device-gateway/gateway.env').read_text().splitlines():
 if line and not line.lstrip().startswith('#') and '=' in line:
  k,v=line.split('=',1);values=shlex.split(v,comments=True);os.environ[k.strip()]=values[0] if values else ''
async def command(obj):
 r,w=await asyncio.open_connection('127.0.0.1',18767);w.write((json.dumps(obj)+'\n').encode());await w.drain();v=json.loads(await r.readline());w.close();await w.wait_closed();return v
async def main():
 status=await command({'command':'status'});assert status['online'] and 'speech_playback' in status['capabilities']
 device=SimpleNamespace(device_id=status['device_id'],connection_id=status['connection_id'],capabilities=set(status['capabilities']),online=True)
 token=json.loads(pathlib.Path('.cache/diagnostic-auth.json').read_text())['token']
 r,w=await asyncio.open_connection('127.0.0.1',18767);w.write((json.dumps({'command':'bridge','token':token})+'\n').encode());await w.drain();assert json.loads(await r.readline())['bridge_ready']
 records=[]
 for lang,text in [('zh','你好，这是语音播放链路测试。'),('en','Hello, this is a speech playback test.')]:
  rec={'language':lang,'actual_device_capabilities':status['capabilities'],'device_connection_id':status['connection_id']};frames=[];pcm_holder=[]
  conversion=speech.audio_to_pcm;bounded=speech.bounded_response
  async def traced_conversion(raw,**kw):
   (root/('l8-'+lang+'.source')).write_bytes(raw);rec['source_bytes']=len(raw);t=time.perf_counter();pcm=await conversion(raw,**kw);rec['ffmpeg_ms']=(time.perf_counter()-t)*1000;pcm_holder.append(pcm);(root/('l8-'+lang+'.pcm')).write_bytes(pcm);return pcm
  async def traced_http(client,url,**kw):
   stream=client.stream;t=time.perf_counter()
   @asynccontextmanager
   async def capture(*a,**k):
    async with stream(*a,**k) as response:
     rec.update(http_status=response.status_code,content_type=response.headers.get('content-type',''));yield response
   client.stream=capture
   try:data,ctype=await bounded(client,url,**kw)
   finally:client.stream=stream
   rec.update(http_download_ms=(time.perf_counter()-t)*1000,http_response_bytes=len(data));return data,ctype
  async def send(d,frame):
   assert d.connection_id==device.connection_id
   frames.append(frame)
   if frame['type']=='speech_start':rec.update(speech_id=frame['speech_id'],time_to_speech_start_ms=(time.perf_counter()-began)*1000)
   if frame['type']=='speech_chunk' and frame['sequence']==0:rec['time_to_first_chunk_ms']=(time.perf_counter()-began)*1000
   w.write((json.dumps(frame,separators=(',',':'))+'\n').encode());await w.drain()
  speech.audio_to_pcm=traced_conversion;speech.bounded_response=traced_http
  manager=speech.SpeechManager(speech.CloudflareMeloTTS(os.environ['GATEWAY_CF_ACCOUNT_ID'],os.environ['GATEWAY_CF_API_TOKEN']),None,speech.SpeechPreferences(root/'l8-prefs.json'),lambda _: 'server',send,lambda d:d.connection_id==device.connection_id,lambda _:False)
  began=time.perf_counter()
  try:
   assert manager.capable(device);assert manager.start(device,'screen',text)
   playback=manager.active[device.connection_id];await playback.task
   assert frames[0]['type']=='speech_start' and frames[-1]['type']=='speech_end'
   chunks=[f for f in frames if f['type']=='speech_chunk'];decoded=b''.join(base64.b64decode(f['audio'],validate=True) for f in chunks)
   assert [f['sequence'] for f in chunks]==list(range(len(chunks))) and frames[-1]['chunks']==len(chunks) and decoded==pcm_holder[0]
   rec.update(result='PASS',pcm_bytes=len(decoded),pcm_duration=len(decoded)/88200,chunks=len(chunks),last_chunk_bytes=len(base64.b64decode(chunks[-1]['audio'])),sha256=hashlib.sha256(decoded).hexdigest(),concat='byte-for-byte PASS',manager_elapsed_ms=(time.perf_counter()-began)*1000)
  except Exception as e:rec.update(result='FAIL',error_class=type(e).__name__)
  finally:speech.audio_to_pcm=conversion;speech.bounded_response=bounded
  records.append(rec);(root/'l8-manager.json').write_text(json.dumps(records,indent=2));print(json.dumps(rec),flush=True)
  if rec['result']!='PASS':raise SystemExit(1)
  await asyncio.sleep(1.5)
 w.close();await w.wait_closed()
asyncio.run(main())