Realtime transcription

# websocat is the practical "curl for WebSockets" -- there is no
# raw curl incantation that speaks this framing. Convert the file
# once, then stream the control frames and the raw audio together;
# -B keeps each write on its own frame instead of coalescing them.
ffmpeg -v error -i voice.m4a -ac 1 -ar 16000 -f f32le voice.raw

{ printf '%s\n' '{"type":"config","model":"whisper-large-v3","language":"persian"}'
  cat voice.raw
  printf '%s\n' '{"type":"done"}'
} | websocat -B 1000000 \
    --header "Sec-WebSocket-Protocol: bearer.$LARSA_API_KEY" \
    wss://api.console.larsa.larsima.com/api/realtime/transcribe

# {"type":"ready","model":"whisper-large-v3","language":"persian"}
# {"type":"final","text":"قرارداد اجاره باید به صورت کتبی تنظیم شود.","duration":3.9}
# {"type":"done"}

import asyncio
import json
import os
import subprocess

from websockets.asyncio.client import connect   # pip install websockets

KEY = os.environ["LARSA_API_KEY"]
URL = "wss://api.console.larsa.larsima.com/api/realtime/transcribe"


def pcm(path):
    """Any container ffmpeg reads -> raw float32 mono at 16 kHz."""
    return subprocess.run(
        ["ffmpeg", "-v", "error", "-i", path, "-ac", "1", "-ar", "16000",
         "-f", "f32le", "pipe:1"], capture_output=True, check=True).stdout


async def main():
    async with connect(URL, subprotocols=[f"bearer.{KEY}"], max_size=None) as ws:
        await ws.send(json.dumps({"type": "config",
                                   "model": "whisper-large-v3",
                                   "language": "persian"}))
        raw, step = pcm("voice.m4a"), 16000 * 4 // 2   # half a second per frame
        for i in range(0, len(raw), step):
            await ws.send(raw[i:i + step])
            await asyncio.sleep(0.05)                  # pace it like live speech
        await ws.send(json.dumps({"type": "done"}))

        async for msg in ws:
            event = json.loads(msg)
            print(event)
            if event.get("type") == "done":
                break


asyncio.run(main())

import { spawnSync } from "node:child_process";
import WebSocket from "ws";   // npm install ws

const KEY = process.env.LARSA_API_KEY;
const URL = "wss://api.console.larsa.larsima.com/api/realtime/transcribe";

function pcm(path) {
  return spawnSync("ffmpeg", ["-v", "error", "-i", path, "-ac", "1",
    "-ar", "16000", "-f", "f32le", "pipe:1"], { maxBuffer: 1 << 28 }).stdout;
}

const ws = new WebSocket(URL, ["bearer." + KEY]);

ws.on("open", () => {
  ws.send(JSON.stringify({ type: "config", model: "whisper-large-v3", language: "persian" }));

  const raw = pcm("voice.m4a");
  const step = (16000 * 4) / 2;   // half a second per frame
  let i = 0;
  const timer = setInterval(() => {
    if (i >= raw.length) {
      clearInterval(timer);
      ws.send(JSON.stringify({ type: "done" }));
      return;
    }
    ws.send(raw.subarray(i, i + step));
    i += step;
  }, 50);
});

ws.on("message", (data) => {
  const event = JSON.parse(data.toString());
  console.log(event);
  if (event.type === "done") ws.close();
});

import { spawnSync } from "node:child_process";
import WebSocket from "ws";   // npm install ws @types/ws

const KEY = process.env.LARSA_API_KEY as string;
const URL = "wss://api.console.larsa.larsima.com/api/realtime/transcribe";

function pcm(path: string): Buffer {
  return spawnSync("ffmpeg", ["-v", "error", "-i", path, "-ac", "1",
    "-ar", "16000", "-f", "f32le", "pipe:1"], { maxBuffer: 1 << 28 }).stdout;
}

const ws = new WebSocket(URL, ["bearer." + KEY]);

ws.on("open", () => {
  ws.send(JSON.stringify({ type: "config", model: "whisper-large-v3", language: "persian" }));

  const raw = pcm("voice.m4a");
  const step = (16000 * 4) / 2;
  let i = 0;
  const timer = setInterval(() => {
    if (i >= raw.length) {
      clearInterval(timer);
      ws.send(JSON.stringify({ type: "done" }));
      return;
    }
    ws.send(raw.subarray(i, i + step));
    i += step;
  }, 50);
});

ws.on("message", (data: WebSocket.RawData) => {
  const event = JSON.parse(data.toString());
  console.log(event);
  if (event.type === "done") ws.close();
});

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/exec"
	"time"

	"github.com/gorilla/websocket" // go get github.com/gorilla/websocket
)

func pcm(path string) []byte {
	out, err := exec.Command("ffmpeg", "-v", "error", "-i", path, "-ac", "1",
		"-ar", "16000", "-f", "f32le", "pipe:1").Output()
	if err != nil {
		log.Fatal(err)
	}
	return out
}

func main() {
	key := os.Getenv("LARSA_API_KEY")
	header := http.Header{"Sec-WebSocket-Protocol": {"bearer." + key}}

	conn, _, err := websocket.DefaultDialer.Dial(
		"wss://api.console.larsa.larsima.com/api/realtime/transcribe", header)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()

	cfg, _ := json.Marshal(map[string]string{
		"type": "config", "model": "whisper-large-v3", "language": "persian"})
	conn.WriteMessage(websocket.TextMessage, cfg)

	raw := pcm("voice.m4a")
	step := 16000 * 4 / 2
	for i := 0; i < len(raw); i += step {
		end := i + step
		if end > len(raw) {
			end = len(raw)
		}
		conn.WriteMessage(websocket.BinaryMessage, raw[i:end])
		time.Sleep(50 * time.Millisecond)
	}
	done, _ := json.Marshal(map[string]string{"type": "done"})
	conn.WriteMessage(websocket.TextMessage, done)

	for {
		_, msg, err := conn.ReadMessage()
		if err != nil {
			break
		}
		fmt.Println(string(msg))
		var event map[string]any
		json.Unmarshal(msg, &event)
		if event["type"] == "done" {
			break
		}
	}
}

// Cargo.toml: tokio-tungstenite = { version = "0.24", features = ["native-tls"] }
//             tokio = { version = "1", features = ["full"] }
//             futures-util = "0.3"
use futures_util::{SinkExt, StreamExt};
use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::Message;
use std::{env, process::Command, time::Duration};

fn pcm(path: &str) -> Vec<u8> {
    Command::new("ffmpeg")
        .args(["-v", "error", "-i", path, "-ac", "1", "-ar", "16000", "-f", "f32le", "pipe:1"])
        .output().unwrap().stdout
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let key = env::var("LARSA_API_KEY")?;
    let mut request =
        "wss://api.console.larsa.larsima.com/api/realtime/transcribe".into_client_request()?;
    request.headers_mut().insert("Sec-WebSocket-Protocol", format!("bearer.{key}").parse()?);

    let (ws_stream, _) = connect_async(request).await?;
    let (mut write, mut read) = ws_stream.split();

    write.send(Message::Text(
        r#"{"type":"config","model":"whisper-large-v3","language":"persian"}"#.into(),
    )).await?;

    let raw = pcm("voice.m4a");
    let step = 16000 * 4 / 2;
    for chunk in raw.chunks(step) {
        write.send(Message::Binary(chunk.to_vec())).await?;
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    write.send(Message::Text(r#"{"type":"done"}"#.into())).await?;

    while let Some(Ok(msg)) = read.next().await {
        if let Message::Text(text) = msg {
            println!("{text}");
            if text.contains("\"done\"") {
                break;
            }
        }
    }
    Ok(())
}

// java.net.http.WebSocket -- built in since JDK 11, no dependency needed.
import java.io.ByteArrayOutputStream;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.WebSocket;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CountDownLatch;

public class Realtime {
    static byte[] pcm(String path) throws Exception {
        Process p = new ProcessBuilder("ffmpeg", "-v", "error", "-i", path,
                "-ac", "1", "-ar", "16000", "-f", "f32le", "pipe:1").start();
        var out = new ByteArrayOutputStream();
        p.getInputStream().transferTo(out);
        p.waitFor();
        return out.toByteArray();
    }

    public static void main(String[] args) throws Exception {
        String key = System.getenv("LARSA_API_KEY");
        CountDownLatch done = new CountDownLatch(1);

        WebSocket.Listener listener = new WebSocket.Listener() {
            @Override
            public CompletionStage<?> onText(WebSocket ws, CharSequence data, boolean last) {
                System.out.println(data);
                if (data.toString().contains("\"done\"")) done.countDown();
                ws.request(1);
                return null;
            }
        };

        WebSocket ws = HttpClient.newHttpClient().newWebSocketBuilder()
                .subprotocols("bearer." + key)
                .buildAsync(URI.create("wss://api.console.larsa.larsima.com/api/realtime/transcribe"), listener)
                .join();

        ws.sendText("{\"type\":\"config\",\"model\":\"whisper-large-v3\",\"language\":\"persian\"}", true);

        byte[] raw = pcm("voice.m4a");
        int step = 16000 * 4 / 2;
        for (int i = 0; i < raw.length; i += step) {
            int end = Math.min(i + step, raw.length);
            ws.sendBinary(ByteBuffer.wrap(raw, i, end - i), true).join();
            Thread.sleep(50);
        }
        ws.sendText("{\"type\":\"done\"}", true);
        done.await();
    }
}

// System.Net.WebSockets.ClientWebSocket -- built in, no dependency needed.
using System.Diagnostics;
using System.Net.WebSockets;
using System.Text;

var key = Environment.GetEnvironmentVariable("LARSA_API_KEY");
using var ws = new ClientWebSocket();
ws.Options.AddSubProtocol($"bearer.{key}");
await ws.ConnectAsync(new Uri("wss://api.console.larsa.larsima.com/api/realtime/transcribe"), CancellationToken.None);

async Task Send(string text) =>
    await ws.SendAsync(Encoding.UTF8.GetBytes(text), WebSocketMessageType.Text, true, CancellationToken.None);

await Send("""{"type":"config","model":"whisper-large-v3","language":"persian"}""");

byte[] Pcm(string path)
{
    using var ff = Process.Start(new ProcessStartInfo("ffmpeg",
        $"-v error -i {path} -ac 1 -ar 16000 -f f32le pipe:1") { RedirectStandardOutput = true })!;
    using var mem = new MemoryStream();
    ff.StandardOutput.BaseStream.CopyTo(mem);
    ff.WaitForExit();
    return mem.ToArray();
}

var raw = Pcm("voice.m4a");
var step = 16000 * 4 / 2;
for (var i = 0; i < raw.Length; i += step)
{
    var end = Math.Min(i + step, raw.Length);
    await ws.SendAsync(raw[i..end], WebSocketMessageType.Binary, true, CancellationToken.None);
    await Task.Delay(50);
}
await Send("""{"type":"done"}""");

var buf = new byte[8192];
while (true)
{
    var result = await ws.ReceiveAsync(buf, CancellationToken.None);
    var text = Encoding.UTF8.GetString(buf, 0, result.Count);
    Console.WriteLine(text);
    if (text.Contains("\"done\"")) break;
}

<?php
// composer require textalk/websocket
require "vendor/autoload.php";

use WebSocket\Client;

$key = getenv("LARSA_API_KEY");
$client = new Client("wss://api.console.larsa.larsima.com/api/realtime/transcribe", [
    "headers" => ["Sec-WebSocket-Protocol" => "bearer.$key"],
]);

$client->send(json_encode([
    "type" => "config", "model" => "whisper-large-v3", "language" => "persian",
]));

$raw = shell_exec("ffmpeg -v error -i voice.m4a -ac 1 -ar 16000 -f f32le pipe:1");

$step = 16000 * 4 / 2;
for ($i = 0; $i < strlen($raw); $i += $step) {
    $client->send(substr($raw, $i, $step), "binary");
    usleep(50000);
}
$client->send(json_encode(["type" => "done"]));

while (true) {
    $message = $client->receive();
    echo $message . "\n";
    if (str_contains($message, '"done"')) break;
}

# gem install websocket-client-simple
require "websocket-client-simple"
require "json"
require "open3"

key = ENV.fetch("LARSA_API_KEY")
url = "wss://api.console.larsa.larsima.com/api/realtime/transcribe"

ws = WebSocket::Client::Simple.connect(url,
  headers: { "Sec-WebSocket-Protocol" => "bearer.#{key}" })

ws.on :open do
  ws.send({ type: "config", model: "whisper-large-v3", language: "persian" }.to_json)

  raw, _ = Open3.capture2("ffmpeg", "-v", "error", "-i", "voice.m4a",
                           "-ac", "1", "-ar", "16000", "-f", "f32le", "pipe:1")
  step = 16_000 * 4 / 2
  raw.bytes.each_slice(step) do |chunk|
    ws.send(chunk.pack("C*"), type: :binary)
    sleep 0.05
  end
  ws.send({ type: "done" }.to_json)
end

ws.on :message do |msg|
  puts msg.data
  ws.close if msg.data.include?('"done"')
end

sleep 30   # keep the process alive while the callbacks above fire