"""E2E v4: authenticated websockets client -> streams jfk.wav exactly as the tablet does.

Note: reads the local kiosk-pass.txt (plaintext, box-only, never published) to log in.
A re-creator should point _PW at their own kiosk password file or run the login against
kiosk_auth.setup_password output."""
import asyncio, json, struct, wave, os, time, http.client
import websockets

WAV = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cache", "jfk.wav")
URI = "ws://127.0.0.1:8443/"
FR = 2048
_PW = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "kiosk-pass.txt")).read().strip()

def _login_cookie():
    c = http.client.HTTPConnection("127.0.0.1", 8443, timeout=10)
    c.request("POST", "/auth", body=json.dumps({"password": _PW}),
              headers={"Content-Type": "application/json"})
    r = c.getresponse(); r.read()
    for k, v in r.getheaders():
        if k.lower() == "set-cookie":
            return v.split(";")[0]
    raise RuntimeError(f"login failed: HTTP {r.status}")

async def main():
    cookie = _login_cookie()
    with wave.open(WAV, "rb") as w:
        sr = w.getframerate(); n = w.getnframes(); raw = w.readframes(n)
    assert sr == 16000
    pcm = [max(-32767, min(32767, v * 16)) for v in struct.unpack(f"<{n}h", raw)]  # jfk is quiet; boost 16x
    t0 = time.time()
    events = []

    async with websockets.connect(URI, max_size=2**22,
                                  additional_headers={"Cookie": cookie}) as ws:
        await ws.send(json.dumps({"type": "hello", "rate": 16000}))
        await ws.send(json.dumps({"type": "ask", "on": True}))   # Phase 2: answer the utterance

        async def pump():
            async for msg in ws:
                if isinstance(msg, str):
                    try:
                        m = json.loads(msg)
                        events.append((time.time() - t0, m))
                    except Exception:
                        pass
                else:
                    events.append((time.time() - t0, {"type": "binary", "n": len(msg)}))
        task = asyncio.create_task(pump())

        for i in range(0, len(pcm), FR):
            chunk = pcm[i:i+FR]
            if len(chunk) < FR: chunk = chunk + [0] * (FR - len(chunk))
            await ws.send(struct.pack(f"<{FR}h", *chunk))
            await asyncio.sleep(FR / sr / 1.6)

        sil = (0,) * FR
        for _ in range(140):                       # 140 * 2048 = ~18s silence cap
            await ws.send(struct.pack(f"<{FR}h", *sil))
            await asyncio.sleep(0.02)
        t_end = time.time() + 150                  # agent-mode: TTFT ~13s + tool passes + TTS
        while time.time() < t_end:
            if any(e[1].get("type") == "tts_end" for e in events):
                break
            await asyncio.sleep(0.3)
        task.cancel()

    print(f"=== {len(events)} events over {time.time()-t0:.1f}s ===")
    bins = sum(e[1].get("n", 0) for e in events if e[1].get("type") == "binary")
    for t, e in events:
        k = e.get("type")
        if k == "binary": continue
        if k == "partial":
            print(f"  +{t*1000:5.0f}ms  partial  {e.get('text')!r}")
        elif k == "final":
            print(f"  +{t*1000:5.0f}ms  FINAL({e.get('ms')}ms)  {e.get('text')!r}")
        elif k == "state":
            print(f"  +{t*1000:5.0f}ms  state -> {e.get('s')}")
        elif k in ("tts_start", "tts_end", "tts_stop"):
            print(f"  +{t*1000:5.0f}ms  {k}")
        else:
            s = e.get("text") or e.get("s") or ""
            print(f"  +{t*1000:5.0f}ms  {k} {str(s)[:80]}")
    print(f"  total TTS audio bytes received: {bins}")

asyncio.run(main())