Twilio Integration

import asyncio
import base64
import json
import logging
import os

from fastapi import FastAPI, WebSocket
from fastapi.responses import Response

from kugelaudio import AudioChunk, KugelAudio, KugelAudioError

PUBLIC_HOST = os.environ["PUBLIC_HOST"]  # e.g. abc123.ngrok.app, without https://
VOICE_ID = 1071

logger = logging.getLogger("twilio-kugelaudio")
app = FastAPI()
kugel = KugelAudio(api_key=os.environ["KUGELAUDIO_API_KEY"])


@app.post("/twiml")
async def twiml() -> Response:
    xml = (
        '<?xml version="1.0" encoding="UTF-8"?>'
        f'<Response><Connect><Stream url="wss://{PUBLIC_HOST}/media" /></Connect></Response>'
    )
    return Response(content=xml, media_type="text/xml")


class Call:
    """One phone call: speaks queued text into the Twilio stream."""

    def __init__(self, ws: WebSocket, stream_sid: str):
        self.ws = ws
        self.stream_sid = stream_sid
        self.texts: asyncio.Queue[str] = asyncio.Queue()
        self.pending_marks: set[str] = set()
        self.replies = 0
        self.speaker = asyncio.create_task(self._speak())

    def say(self, text: str) -> None:
        """Queue text. Replies play in order, one after the other."""
        self.texts.put_nowait(text)

    async def interrupt(self) -> None:
        """Barge-in: stop generating, drop queued text, empty Twilio's buffer."""
        self.speaker.cancel()
        while not self.texts.empty():
            self.texts.get_nowait()
        self.pending_marks.clear()  # Twilio echoes cleared marks; ignore them
        await self._send({"event": "clear"})
        self.speaker = asyncio.create_task(self._speak())

    def on_mark(self, name: str) -> None:
        if name in self.pending_marks:
            self.pending_marks.discard(name)
            logger.info("%s finished playing", name)

    async def close(self) -> None:
        self.speaker.cancel()

    async def _speak(self) -> None:
        while True:
            text = await self.texts.get()
            try:
                async for item in kugel.tts.stream_async(
                    text=text,
                    model_id="kugel-3.5",
                    voice_id=VOICE_ID,
                    language="en",
                    output_format="ulaw_8000",  # what Twilio plays: µ-law, 8000 Hz, mono
                    reuse_connection=False,  # one connection per reply, so a cancel closes it
                ):
                    if isinstance(item, AudioChunk):
                        payload = base64.b64encode(item.audio).decode("ascii")
                        await self._send({"event": "media", "media": {"payload": payload}})
            except KugelAudioError:
                # Log and keep the call up, so the caller still hears the next reply.
                logger.exception("KugelAudio TTS failed for %r", text)
                continue
            self.replies += 1
            name = f"reply-{self.replies}"
            self.pending_marks.add(name)
            await self._send({"event": "mark", "mark": {"name": name}})

    async def _send(self, message: dict) -> None:
        await self.ws.send_text(json.dumps({**message, "streamSid": self.stream_sid}))


@app.websocket("/media")
async def media(ws: WebSocket) -> None:
    await ws.accept()
    call: Call | None = None
    try:
        async for message in ws.iter_text():
            data = json.loads(message)
            event = data["event"]
            if event == "start":
                call = Call(ws, data["start"]["streamSid"])
                call.say(
                    "Hello! You are hearing a KugelAudio voice on a phone line. "
                    "Press any key on your phone to interrupt me."
                )
            elif event == "media":
                pass  # the caller's audio: base64 µ-law at 8000 Hz, for your speech recognition
            elif event == "dtmf" and call:
                await call.interrupt()
                call.say(f"You pressed {data['dtmf']['digit']}.")
            elif event == "mark" and call:
                call.on_mark(data["mark"]["name"])
            elif event == "stop":
                break
    finally:
        if call:
            await call.close()