Talk to Prism over WebSocket

import { on } from "node:events";
import { readFileSync, writeFileSync } from "node:fs";
import WebSocket from "ws";

const URL = "wss://prism.api.rime.ai/v1/realtime";
const INSTRUCTIONS =
  "You answer the phone for Juniper Cafe. It's open from 8 a.m. to 4 p.m. " +
  "every day. Answer in one short sentence.";
const FRAME = 1280; // 40 ms of 16 kHz mono PCM16

function sendCallAudio(socket, question) {
  // Send the question at speaking pace, then silence, as a live call would.
  let offset = 0;
  return setInterval(() => {
    const frame = Buffer.alloc(FRAME);
    question.subarray(offset, offset + FRAME).copy(frame);
    const audio = frame.toString("base64");
    socket.send(JSON.stringify({ type: "input_audio_buffer.append", audio }));
    offset += FRAME;
  }, 40);
}

async function readReply(socket, question) {
  const audio = [];
  let sending;
  try {
    for await (const [message] of on(socket, "message", { close: ["close"] })) {
      const event = JSON.parse(message);
      switch (event.type) {
        case "session.updated":
          // Prism refuses audio sent before the session is set up.
          sending = sendCallAudio(socket, question);
          break;
        case "input_audio_buffer.speech_started":
          console.log("Caller started speaking");
          break;
        case "input_audio_buffer.speech_stopped":
          console.log("Caller stopped speaking");
          break;
        case "response.created":
          process.stdout.write("Prism: ");
          break;
        case "response.text.delta":
          process.stdout.write(event.delta);
          break;
        case "response.audio.delta":
          audio.push(Buffer.from(event.delta, "base64"));
          break;
        case "error":
          throw new Error(`${event.error.code}: ${event.error.message}`);
        case "response.done": {
          console.log();
          const { status, status_details } = event.response;
          if (status !== "completed") {
            throw new Error(`Reply ${status}: ${status_details.reason}`);
          }
          return Buffer.concat(audio);
        }
      }
    }
  } finally {
    clearInterval(sending);
  }
  throw new Error("The session closed before the reply ended");
}

// Prism replies in 24 kHz mono PCM16. A 44-byte header makes it a WAV file.
function toWav(pcm) {
  const header = Buffer.alloc(44);
  header.write("RIFF", 0);
  header.writeUInt32LE(36 + pcm.length, 4);
  header.write("WAVEfmt ", 8);
  header.writeUInt32LE(16, 16);
  header.writeUInt16LE(1, 20); // PCM
  header.writeUInt16LE(1, 22); // mono
  header.writeUInt32LE(24000, 24); // samples per second
  header.writeUInt32LE(48000, 28); // bytes per second
  header.writeUInt16LE(2, 32);
  header.writeUInt16LE(16, 34);
  header.write("data", 36);
  header.writeUInt32LE(pcm.length, 40);
  return Buffer.concat([header, pcm]);
}

const question = readFileSync(process.argv[2]);
const socket = new WebSocket(URL, {
  headers: { Authorization: `Bearer ${process.env.RIME_API_KEY}` },
});
socket.on("open", () => {
  socket.send(
    JSON.stringify({
      type: "session.update",
      event_id: "evt_setup", // Comes back in any error about this message
      session: { voice: "aurora", instructions: INSTRUCTIONS },
    }),
  );
});
let timedOut = false;
const deadline = setTimeout(() => {
  timedOut = true;
  socket.terminate();
}, 60_000);
try {
  const reply = await readReply(socket, question);
  writeFileSync("reply.wav", toWav(reply));
  console.log(`Saved ${(reply.length / 48000).toFixed(1)} seconds of audio to reply.wav`);
} catch (error) {
  throw timedOut ? new Error("No reply within 60 seconds") : error;
} finally {
  clearTimeout(deadline);
  socket.close();
}