Noise Suppression WebSocket examples

# pip install websockets
import asyncio
import contextlib
import json
import wave

import websockets

API_KEY = "YOUR_API_KEY"
INPUT_FILE = "noisy.wav"
OUTPUT_FILE = "denoised.wav"
URL = "wss://api.telnyx.com/v2/noise_suppression"


# Temporary failures: server restart, server error, server busy.
RETRY_CLOSE_CODES = {1001, 1011, 1013}


class Retryable(Exception):
    """A temporary failure (see RETRY_CLOSE_CODES, or HTTP 503). Try again."""


async def denoise_once(pcm, sample_rate):
    chunk_bytes = sample_rate // 50 * 2  # 20 ms
    url = f"{URL}?format=l16&sample_rate={sample_rate}&channels=1"
    headers = {"Authorization": f"Bearer {API_KEY}"}
    output = bytearray()

    try:
        async with websockets.connect(url, additional_headers=headers) as ws:
            async def send():
                for i in range(0, len(pcm), chunk_bytes):
                    await ws.send(pcm[i:i + chunk_bytes])
                await ws.send(json.dumps({"type": "end_of_stream"}))

            sender = asyncio.create_task(send())
            try:
                async for message in ws:
                    if isinstance(message, bytes):
                        output.extend(message)  # empty frames add nothing
                    else:
                        print("Error:", json.loads(message))
            except websockets.ConnectionClosedError:
                pass  # the close code is checked below
            finally:
                sender.cancel()
                with contextlib.suppress(asyncio.CancelledError, websockets.ConnectionClosed):
                    await sender
            if ws.close_code in RETRY_CLOSE_CODES:
                raise Retryable(f"closed with {ws.close_code} {ws.close_reason}")
            if ws.close_code != 1000:
                raise RuntimeError(f"closed with {ws.close_code} {ws.close_reason}")
    except websockets.InvalidStatus as e:
        if e.response.status_code == 503:  # server is draining
            raise Retryable("service unavailable") from e
        raise
    return bytes(output)


async def denoise():
    with wave.open(INPUT_FILE, "rb") as wav:
        if wav.getsampwidth() != 2 or wav.getnchannels() != 1:
            raise ValueError("Input must be mono 16-bit PCM")
        sample_rate = wav.getframerate()
        pcm = wav.readframes(wav.getnframes())

    for attempt in range(5):
        try:
            output = await denoise_once(pcm, sample_rate)
            break
        except Retryable as e:
            if attempt == 4:
                raise RuntimeError("service unavailable after 5 attempts") from e
            print(f"{e}, retrying")
            await asyncio.sleep(2 ** attempt)

    with wave.open(OUTPUT_FILE, "wb") as out:
        out.setnchannels(1)
        out.setsampwidth(2)
        out.setframerate(sample_rate)
        out.writeframes(output)
    print(f"Wrote {len(output)} bytes to {OUTPUT_FILE}")


asyncio.run(denoise())

// npm install ws
const fs = require("fs");
const WebSocket = require("ws");

const API_KEY = "YOUR_API_KEY";
const INPUT_FILE = "noisy.wav";
const OUTPUT_FILE = "denoised.wav";
const URL = "wss://api.telnyx.com/v2/noise_suppression";

// Find the "fmt " and "data" chunks of a PCM WAV file.
function readWav(path) {
  const buf = fs.readFileSync(path);
  let fmt;
  let data;
  for (let pos = 12; pos + 8 <= buf.length; ) {
    const id = buf.toString("ascii", pos, pos + 4);
    const size = buf.readUInt32LE(pos + 4);
    if (id === "fmt ") fmt = buf.subarray(pos + 8, pos + 8 + size);
    if (id === "data") data = buf.subarray(pos + 8, pos + 8 + size);
    pos += 8 + size + (size % 2);
  }
  if (fmt.readUInt16LE(14) !== 16 || fmt.readUInt16LE(2) !== 1) {
    throw new Error("Input must be mono 16-bit PCM");
  }
  return { sampleRate: fmt.readUInt32LE(4), pcm: data };
}

function writeWav(path, pcm, sampleRate) {
  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);
  header.writeUInt16LE(1, 22); // mono
  header.writeUInt32LE(sampleRate, 24);
  header.writeUInt32LE(sampleRate * 2, 28);
  header.writeUInt16LE(2, 32);
  header.writeUInt16LE(16, 34);
  header.write("data", 36);
  header.writeUInt32LE(pcm.length, 40);
  fs.writeFileSync(path, Buffer.concat([header, pcm]));
}

// Temporary failures: server restart, server error, server busy.
const RETRY_CLOSE_CODES = [1001, 1011, 1013];

// Resolves with the denoised audio. Rejects with err.retryable = true on a
// temporary failure (RETRY_CLOSE_CODES or HTTP 503).
function denoiseOnce(pcm, sampleRate) {
  const chunkBytes = Math.floor(sampleRate / 50) * 2; // 20 ms
  return new Promise((resolve, reject) => {
    const ws = new WebSocket(
      `${URL}?format=l16&sample_rate=${sampleRate}&channels=1`,
      { headers: { Authorization: `Bearer ${API_KEY}` } },
    );
    const output = [];

    ws.on("open", () => {
      for (let i = 0; i < pcm.length; i += chunkBytes) {
        ws.send(pcm.subarray(i, i + chunkBytes));
      }
      ws.send(JSON.stringify({ type: "end_of_stream" }));
    });

    ws.on("message", (data, isBinary) => {
      if (isBinary) output.push(data); // empty frames add nothing
      else console.error("Error:", JSON.parse(data));
    });

    ws.on("unexpected-response", (req, res) => {
      const err = new Error(`HTTP ${res.statusCode}`);
      err.retryable = res.statusCode === 503; // server is draining
      reject(err);
      req.destroy();
    });

    // 1000: all audio returned. RETRY_CLOSE_CODES: try again.
    ws.on("close", (code, reason) => {
      if (code === 1000) return resolve(Buffer.concat(output));
      const err = new Error(`Closed with ${code} ${reason}`);
      err.retryable = RETRY_CLOSE_CODES.includes(code);
      reject(err);
    });

    ws.on("error", reject);
  });
}

async function main() {
  const { sampleRate, pcm } = readWav(INPUT_FILE);
  for (let attempt = 0; ; attempt += 1) {
    try {
      const denoised = await denoiseOnce(pcm, sampleRate);
      writeWav(OUTPUT_FILE, denoised, sampleRate);
      console.log(`Wrote ${denoised.length} bytes to ${OUTPUT_FILE}`);
      return;
    } catch (err) {
      if (!err.retryable || attempt >= 4) throw err;
      console.error(`${err.message}, retrying`);
      await new Promise((r) => setTimeout(r, 2 ** attempt * 1000));
    }
  }
}

main().catch((err) => {
  console.error("Error:", err.message);
  process.exitCode = 1;
});