gogo-telefon/scripts/poc/latency_smoke.py

"""M0 PoC #3 — end-to-end latency smoke test (§14 Phase 0).

Measures mouth-to-ear response time through the REAL pipeline: connects to the
running AudioSocket voice server exactly like Asterisk would, plays a recorded
utterance (or a tone with --fake providers), and measures the gap between
end-of-utterance and the first agent audio frame of the reply.

Target (§15): < 1.5 s, worst case < 2.5 s. GATE: > 2.5 s → rethink providers.

Setup:
    1. backend + voice server running (see docs/VOICE_TESTING.md)
    2. a CallRegistration for the demo tenant is created automatically here

Run:  python scripts/poc/latency_smoke.py [utterance.wav] [--host 127.0.0.1] [--port 9092]
"""

from __future__ import annotations

import asyncio
import sys
import time
import wave
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from sqlalchemy import select  # noqa: E402

from gogo.db import get_sessionmaker  # noqa: E402
from gogo.models import CallRegistration, Tenant  # noqa: E402
from gogo.voice.audio import FRAME_BYTES, FRAME_MS, resample, silence, tone  # noqa: E402

KIND_TERMINATE, KIND_UUID, KIND_AUDIO = 0x00, 0x01, 0x10


def load_utterance(path: str | None) -> bytes:
    if path:
        with wave.open(path, "rb") as w:
            pcm = w.readframes(w.getnframes())
            if w.getframerate() != 8000:
                pcm = resample(pcm, w.getframerate(), 8000)
            return pcm
    print("(no wav given — using a 1s tone; STT will hear nothing useful, "
          "use a real recording for a true number)")
    return tone(440, 1000)


async def register_call() -> str:
    async with get_sessionmaker()() as session:
        tenant = (
            await session.execute(select(Tenant).where(Tenant.slug == "salon-merima"))
        ).scalar_one_or_none()
        if tenant is None:
            print("run `python -m gogo.seed` first")
            sys.exit(1)
        reg = CallRegistration(tenant_id=tenant.id, caller_msisdn="+38765123456")
        session.add(reg)
        await session.commit()
        return str(reg.id)


async def main() -> None:
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    host = sys.argv[sys.argv.index("--host") + 1] if "--host" in sys.argv else "127.0.0.1"
    port = int(sys.argv[sys.argv.index("--port") + 1]) if "--port" in sys.argv else 9092

    import uuid as uuid_mod

    reg_id = uuid_mod.UUID(await register_call())
    utterance = load_utterance(args[0] if args else None) + silence(800)

    reader, writer = await asyncio.open_connection(host, port)
    writer.write(bytes([KIND_UUID, 0, 16]) + reg_id.bytes)
    await writer.drain()

    async def read_msg():
        header = await reader.readexactly(3)
        length = int.from_bytes(header[1:3], "big")
        payload = await reader.readexactly(length) if length else b""
        return header[0], payload

    # 1) greeting: time from handshake to first audio frame
    t_start = time.monotonic()
    kind, _ = await read_msg()
    greeting_first_audio = time.monotonic() - t_start
    print(f"greeting first audio after {greeting_first_audio * 1000:.0f} ms")

    # drain the greeting while it plays (send silence like a real caller)
    async def drain_audio(quiet_ms_needed: int = 600) -> None:
        last_audio = time.monotonic()
        while (time.monotonic() - last_audio) * 1000 < quiet_ms_needed:
            try:
                kind, _ = await asyncio.wait_for(read_msg(), timeout=0.05)
                if kind == KIND_AUDIO:
                    last_audio = time.monotonic()
            except TimeoutError:
                pass
            writer.write(bytes([KIND_AUDIO]) + FRAME_BYTES.to_bytes(2, "big") + silence(FRAME_MS))
            await writer.drain()
            await asyncio.sleep(FRAME_MS / 1000)

    await drain_audio()

    # 2) speak the utterance paced like a phone, then measure to first reply frame
    for i in range(0, len(utterance), FRAME_BYTES):
        frame = utterance[i : i + FRAME_BYTES].ljust(FRAME_BYTES, b"\x00")
        writer.write(bytes([KIND_AUDIO]) + len(frame).to_bytes(2, "big") + frame)
        await writer.drain()
        await asyncio.sleep(FRAME_MS / 1000)
    t_end_of_speech = time.monotonic()

    while True:
        kind, payload = await read_msg()
        if kind == KIND_AUDIO and payload.strip(b"\x00"):
            break
        if kind == KIND_TERMINATE:
            print("server hung up before replying")
            return
    response_latency = time.monotonic() - t_end_of_speech
    print(f"\nmouth-to-ear response latency: {response_latency * 1000:.0f} ms")
    verdict = "OK ✓" if response_latency < 1.5 else ("worst-case ⚠" if response_latency < 2.5 else "FAIL ✗ — rethink providers (§14)")
    print(f"target <1500 ms, worst case <2500 ms → {verdict}")
    writer.close()


if __name__ == "__main__":
    asyncio.run(main())