Add language selection to your Twilio transcription line

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
import os
import audioop
import threading
import time
from collections import deque
from datetime import datetime
from typing import Deque, Dict, Optional
 
from orbitalsai.streaming import StreamingClient, StreamingConfig, StreamingEventHandlers
from dotenv import load_dotenv
 
load_dotenv()
 
ORBITALSAI_API_KEY = os.getenv("ORBITALSAI_API_KEY")
 
TWILIO_SAMPLE_RATE = 8000
ORBITALS_SAMPLE_RATE = 16000
CHUNK_SAMPLES = 4000
MAX_CONNECT_RETRIES = 3
CONNECT_RETRY_DELAY = 1.0
 
# Shared in-memory store: holds the last 100 transcript entries across all calls.
# Imported by main.py so both modules see the same deque object.
RECENT_TRANSCRIPTS: Deque[Dict[str, str]] = deque(maxlen=100)
 
 
class TwilioOrbitalsHandlers(StreamingEventHandlers):
    """Write partial and final transcripts to stdout and to the shared store."""
 
    def __init__(self, language: str, call_id: str, store: Deque[Dict[str, str]]) -> None:
        self.language = language
        self.call_id = call_id
        self.store = store
 
    def on_open(self, session_info: dict) -> None:
        # Use the language the server confirms, in case it differs from what we requested.
        server_lang = (session_info.get("language") or self.language).lower()
        self.language = server_lang
        print(f"OrbitalsAI session opened: {session_info.get('session_id')}")
        print(f"Language confirmed: {server_lang}")
 
    def _push(self, text: str, kind: str) -> None:
        self.store.append({
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "language": self.language,
            "call_id": self.call_id,
            "text": text,
            "kind": kind,
        })
 
    def on_transcript_partial(self, transcript: str) -> None:
        if not transcript:
            return
        print(f"\rPartial: {transcript}", end="", flush=True)
        self._push(transcript, "partial")
 
    def on_transcript_final(self, transcript: str, metadata: dict) -> None:
        if not transcript:
            return
        print(f"\nFinal: {transcript}")
        print(
            f"  Cost: \${metadata.get('cost', 0):.4f} | "
            f"Duration: {metadata.get('audio_seconds', 0):.1f}s"
        )
        self._push(transcript, "final")
 
    def on_speech_start(self) -> None:
        print("\n[Speech detected]")
 
    def on_speech_end(self) -> None:
        print("\n[Silence detected]")
 
    def on_error(self, error: Exception) -> None:
        print(f"\nOrbitalsAI error: {error}")
 
    def on_close(self, code: Optional[int], reason: Optional[str]) -> None:
        print(f"\nOrbitalsAI session closed (code={code}, reason={reason})")
 
 
class TwilioOrbitalsTranscriber:
    """
    Twilio media → OrbitalsAI streaming transcription with language support.
 
    Per-frame pipeline (unchanged from basic version):
      1. 8-bit mu-law @ 8 kHz  →  PCM16 @ 8 kHz   (audioop.ulaw2lin)
      2. PCM16 @ 8 kHz         →  PCM16 @ 16 kHz   (audioop.ratecv)
      3. Buffer until 0.25 s of audio, then send_audio()
    """
 
    def __init__(self, language: str, call_id: str, store: Deque[Dict[str, str]]) -> None:
        if not ORBITALSAI_API_KEY:
            raise RuntimeError("ORBITALSAI_API_KEY is not set")
 
        self.language = language.lower()
        self.call_id = call_id
 
        self._config = StreamingConfig(
            sample_rate=ORBITALS_SAMPLE_RATE,
            chunk_size=CHUNK_SAMPLES,
            interim_results=True,
            language=self.language,
        )
        self._client = StreamingClient(api_key=ORBITALSAI_API_KEY, config=self._config)
        self._handlers = TwilioOrbitalsHandlers(self.language, self.call_id, store)
        self._buffer = bytearray()
        self._lock = threading.Lock()
        self._min_send_bytes = CHUNK_SAMPLES * 2
        self._ratecv_state = None
 
    def connect(self) -> None:
        last_error: Optional[Exception] = None
 
        for attempt in range(1, MAX_CONNECT_RETRIES + 1):
            try:
                print(f"Connecting to OrbitalsAI (attempt {attempt}, lang={self.language})...")
                self._client.connect(self._handlers)
 
                # This is not optional. StreamingConfig alone does not transmit
                # language or sample_rate over the WebSocket handshake.
                # Skip this and every call transcribes in English.
                self._client.configure(
                    language=self.language,
                    sample_rate=ORBITALS_SAMPLE_RATE,
                )
 
                print("OrbitalsAI connected and configured.")
                return
 
            except Exception as exc:
                last_error = exc
                print(f"Attempt {attempt} failed: {exc}")
                if attempt < MAX_CONNECT_RETRIES:
                    time.sleep(CONNECT_RETRY_DELAY)
                    self._client = StreamingClient(
                        api_key=ORBITALSAI_API_KEY,
                        config=self._config,
                    )
 
        raise RuntimeError(
            f"Failed to connect after {MAX_CONNECT_RETRIES} attempts: {last_error}"
        )
 
    def stream(self, audio_chunk: bytes) -> None:
        try:
            pcm16_8k = audioop.ulaw2lin(audio_chunk, 2)
        except Exception as exc:
            print(f"mu-law decode error: {exc}")
            return
 
        try:
            pcm16_16k, self._ratecv_state = audioop.ratecv(
                pcm16_8k, 2, 1,
                TWILIO_SAMPLE_RATE, ORBITALS_SAMPLE_RATE,
                self._ratecv_state,
            )
        except Exception as exc:
            print(f"Resample error: {exc}")
            return
 
        with self._lock:
            self._buffer.extend(pcm16_16k)
            if len(self._buffer) < self._min_send_bytes:
                return
            to_send = bytes(self._buffer)
            self._buffer.clear()
 
        try:
            self._client.send_audio(to_send)
        except Exception as exc:
            print(f"send_audio error: {exc}")
 
    def close(self) -> None:
        with self._lock:
            remaining = bytes(self._buffer)
            self._buffer.clear()
 
        if remaining:
            try:
                self._client.send_audio(remaining)
            except Exception as exc:
                print(f"Error flushing remaining audio: {exc}")
 
        try:
            self._client.flush()
        except Exception as exc:
            print(f"flush() error: {exc}")
 
        time.sleep(1.0)
        self._client.disconnect()

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
190
191
import base64
import json
 
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse, PlainTextResponse, Response
 
from twilio_transcriber import TwilioOrbitalsTranscriber, RECENT_TRANSCRIPTS
 
PORT = 5000
INCOMING_CALL_ROUTE = "/"
WEBSOCKET_ROUTE = "/realtime"
 
app = FastAPI()
 
 
@app.api_route(INCOMING_CALL_ROUTE, methods=["GET", "POST"])
async def receive_call(request: Request):
    if request.method == "POST":
        host = request.headers.get("host", "localhost")
        twiml = f"""
<Response>
  <Gather input="dtmf" numDigits="1" action="https://{host}/language" method="POST">
    <Say>For English, press 1. For Hausa, press 2. For Igbo, press 3. For Yoruba, press 4.</Say>
  </Gather>
  <Redirect>https://{host}/</Redirect>
</Response>
""".strip()
        return Response(content=twiml, media_type="text/xml")
 
    # Browser GET: serve the transcript dashboard
    html = """
<!doctype html>
<html>
<head>
  <meta charset="utf-8"/>
  <title>OrbitalsAI Live Transcripts</title>
  <style>
    body { font-family: system-ui, sans-serif; margin: 2rem; }
    h1   { margin-bottom: 1rem; }
    .item { margin-bottom: 0.75rem; }
    .meta { color: #666; font-size: 0.85rem; }
  </style>
</head>
<body>
  <h1>Live Call Transcripts</h1>
  <div id="items"></div>
  <script>
    async function load() {
      const res  = await fetch('/transcripts', { cache: 'no-store' });
      const data = await res.json();
 
      // Show the latest entry per call. Partials update in place
      // rather than flooding the page.
      const byCall = new Map();
      data.items.forEach(item => byCall.set(item.call_id || 'unknown', item));
 
      const container = document.getElementById('items');
      container.innerHTML = '';
      Array.from(byCall.values())
        .sort((a, b) => a.timestamp.localeCompare(b.timestamp))
        .forEach(item => {
          const div  = document.createElement('div');
          div.className = 'item';
          const meta = document.createElement('div');
          meta.className = 'meta';
          meta.textContent = `[${item.timestamp}] lang=${item.language} kind=${item.kind} call=${item.call_id}`;
          const text = document.createElement('div');
          text.textContent = item.text;
          div.appendChild(meta);
          div.appendChild(text);
          container.appendChild(div);
        });
    }
 
    load();
    setInterval(load, 3000);
  </script>
</body>
</html>
""".strip()
    return HTMLResponse(html)
 
 
@app.post("/language")
async def select_language(request: Request):
    """
    Twilio posts the digit the caller pressed and the CallSid.
    Map the digit to a language code and start the media stream,
    passing both as TwiML <Parameter> values so the WebSocket
    handler can read them from the 'start' event.
    """
    form = await request.form()
    digits   = (form.get("Digits") or "").strip()
    call_sid = form.get("CallSid", "unknown")
 
    language_map = {
        "1": ("english", "English"),
        "2": ("hausa",   "Hausa"),
        "3": ("igbo",    "Igbo"),
        "4": ("yoruba",  "Yoruba"),
    }
    language_code, language_label = language_map.get(digits, ("english", "English"))
 
    print(f"Caller selected: {language_label} (digit={digits}, call_sid={call_sid})")
 
    host = request.headers.get("host", "localhost")
    twiml = f"""
<Response>
  <Say>You selected {language_label}. Please speak after the tone.</Say>
  <Connect>
    <Stream url="wss://{host}{WEBSOCKET_ROUTE}">
      <Parameter name="lang"    value="{language_code}"/>
      <Parameter name="call_id" value="{call_sid}"/>
    </Stream>
  </Connect>
</Response>
""".strip()
    return Response(content=twiml, media_type="text/xml")
 
 
@app.get("/transcripts")
async def list_transcripts():
    return {"items": list(RECENT_TRANSCRIPTS)}
 
 
@app.websocket(WEBSOCKET_ROUTE)
async def transcription_websocket(ws: WebSocket):
    """
    Twilio sends four event types over this WebSocket:
      connected: handshake complete, no audio yet
      start: stream is about to begin; contains customParameters
      media: one frame of base64-encoded mu-law audio
      stop: call ended
    """
    await ws.accept()
    transcriber: TwilioOrbitalsTranscriber | None = None
 
    try:
        while True:
            message = await ws.receive_text()
            data = json.loads(message)
            event_type = data.get("event")
 
            if event_type == "connected":
                print("Twilio WebSocket connected")
 
            elif event_type == "start":
                # Read language and call_id from the <Parameter> values
                # that /language embedded in the TwiML <Stream>.
                start_info = data.get("start", {})
                custom     = start_info.get("customParameters", {}) or {}
                lang       = custom.get("lang", "english")
                call_id    = (
                    custom.get("call_id")
                    or start_info.get("callSid")
                    or start_info.get("streamSid", "unknown")
                )
 
                try:
                    transcriber = TwilioOrbitalsTranscriber(
                        language=lang,
                        call_id=call_id,
                        store=RECENT_TRANSCRIPTS,
                    )
                    transcriber.connect()
                    print(f"Stream started (lang={lang}, call_id={call_id})")
                except Exception as exc:
                    print(f"Could not start transcriber: {exc}")
                    transcriber = None
 
            elif event_type == "media":
                if not transcriber:
                    continue
                raw = base64.b64decode(data["media"]["payload"])
                transcriber.stream(raw)
 
            elif event_type == "stop":
                print("Stream stopped")
                if transcriber:
                    transcriber.close()
                break
 
    except WebSocketDisconnect:
        print("Twilio WebSocket disconnected")
        if transcriber:
            transcriber.close()
 
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=PORT, reload=False)