WebRTC & Audio Processing

Microphone → MediaStream → AudioContext → OpusEncoder → WebSocket → Backend
                                                                        ↓
                                           STT (Speech-to-Text)
                                           LLM (Language Model)
                                           TTS (Text-to-Speech)
                                                                        ↓
Speakers ← AudioContext ← OpusDecoder ← WebSocket ← Backend

export interface AudioProcessor {
  audioContext: AudioContext;
  opusRecorder: OpusRecorder;
  decoder: DecoderWorker;
  outputWorklet: AudioWorkletNode;
  inputAnalyser: AnalyserNode;
  outputAnalyser: AnalyserNode;
  mediaStreamDestination: MediaStreamAudioDestinationNode;
}

const recorderOptions = {
  mediaTrackConstraints: {
    audio: {
      echoCancellation: true,
      noiseSuppression: false,
      autoGainControl: true,
      channelCount: 1,
    },
    video: false,
  },
  encoderPath: "/encoderWorker.min.js",
  bufferLength: Math.round((960 * audioContext.sampleRate) / 24000),
  encoderFrameSize: 20,
  encoderSampleRate: 24000,
  maxFramesPerPage: 2,
  numberOfChannels: 1,
  recordingGain: 1,
  resampleQuality: 3,
  encoderComplexity: 0,
  encoderApplication: 2049,
  streamPages: true,
};

const opusRecorder = new OpusRecorder(recorderOptions);
opusRecorder.ondataavailable = (data: Uint8Array) => {
  // opus actually always works at 48khz internally
  micDuration = opusRecorder.encodedSamplePosition / 48000;
  onOpusRecorded(data);
};

export const base64EncodeOpus = (opusData: Uint8Array) => {
  let binary = "";
  for (let i = 0; i < opusData.byteLength; i++) {
    binary += String.fromCharCode(opusData[i]);
  }
  return window.btoa(binary);
};

const decoder = new Worker("/decoderWorker.min.js");
let micDuration = 0;

decoder.onmessage = (event: MessageEvent<any>) => {
  if (!event.data) {
    return;
  }
  const frame = event.data[0];
  outputWorklet.port.postMessage({
    frame: frame,
    type: "audio",
    micDuration: micDuration,
  });
};

decoder.postMessage({
  command: "init",
  bufferLength: (960 * audioContext.sampleRate) / 24000,
  decoderSampleRate: 24000,
  outputBufferSampleRate: audioContext.sampleRate,
  resampleQuality: 0,
});

const inputAnalyser = audioContext.createAnalyser();
inputAnalyser.fftSize = 2048;
source.connect(inputAnalyser);

const outputAnalyser = audioContext.createAnalyser();
outputAnalyser.fftSize = 2048;
outputWorklet.connect(outputAnalyser);

opus_reader = sphn.OpusStreamReader(SAMPLE_RATE)
wait_for_first_opus = True

while True:
    message_raw = await websocket.receive_text()
    message: ora.ClientEvent = ClientEventAdapter.validate_json(message_raw)
    
    if isinstance(message, ora.InputAudioBufferAppend):
        opus_bytes = base64.b64decode(message.audio)
        if wait_for_first_opus:
            # Check for first packet bit (opus_bytes[5] & 2)
            if opus_bytes[5] & 2:
                wait_for_first_opus = False
            else:
                continue
        pcm = await asyncio.to_thread(opus_reader.append_bytes, opus_bytes)
        
        if pcm.size:
            await handler.receive((SAMPLE_RATE, pcm[np.newaxis, :]))

opus_writer = sphn.OpusStreamWriter(SAMPLE_RATE)

while True:
    emitted_by_handler = await handler.emit()
    
    if isinstance(emitted_by_handler, tuple):
        _sr, audio = emitted_by_handler
        audio = audio_to_float32(audio)
        opus_bytes = await asyncio.to_thread(opus_writer.append_pcm, audio)
        
        # Due to buffering/chunking, Opus doesn't necessarily output
        # something on every PCM added
        if opus_bytes:
            to_emit = ora.ResponseAudioDelta(
                delta=base64.b64encode(opus_bytes).decode("utf-8"),
            )

# Only release the audio such that it's AUDIO_BUFFER_SEC ahead of real time.
# If the value is too low, it might cause stuttering.
# If it's too high, it's difficult to control the synchronization of the text
# and the audio, because that's controlled by emit() and WebRTC. Note that some
# desynchronization can still occur if the TTS is less than real-time, because
# WebRTC will decide to do some buffering of the audio on the fly.
AUDIO_BUFFER_SEC = FRAME_TIME_SEC * 4

def get_cloudflare_rtc_configuration():
    # see: https://fastrtc.org/deployment/#cloudflare-calls-api
    turn_key_id = os.environ.get("TURN_KEY_ID")
    turn_key_api_token = os.environ.get("TURN_KEY_API_TOKEN")
    ttl = 86400  # 24 hours

    response = requests.post(
        f"https://rtc.live.cloudflare.com/v1/turn/keys/{turn_key_id}/credentials/generate-ice-servers",
        headers={
            "Authorization": f"Bearer {turn_key_api_token}",
            "Content-Type": "application/json",
        },
        json={"ttl": ttl},
    )
    if response.ok:
        return response.json()