Transcribe a local file
# pip install requests
import os, sys, time, requests
API = "https://api.aircaps.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['AIRCAPS_API_KEY']}"}
def call(method, path, **kw):
r = requests.request(method, API + path, headers=HEADERS, timeout=60, **kw)
if r.status_code >= 400:
raise SystemExit(r.json()["error"]["message"])
return r.json()
upload = call("POST", "/files/upload-url")
with open(sys.argv[1], "rb") as f:
requests.put(upload["upload_url"], data=f, timeout=600).raise_for_status()
job = call("POST", "/transcripts", json={"model": "a5s-async-v1", "file_id": upload["id"], "language_code": "en"})
while job["status"] in ("queued", "processing"):
time.sleep(3)
job = call("GET", f"/transcripts/{job['id']}")
if job["status"] == "error":
raise SystemExit(job["error"]["message"])
for u in job["utterances"]:
print(f"[{u['start']:8.1f}s] {u['speaker']}: {u['text']}")
// node transcribe.mjs meeting.mp3 (Node 18+)
import { readFile } from "node:fs/promises";
const API = "https://api.aircaps.com/v1";
const headers = { Authorization: `Bearer ${process.env.AIRCAPS_API_KEY}` };
async function call(method, path, body) {
const r = await fetch(API + path, {
method,
headers: body ? { ...headers, "Content-Type": "application/json" } : headers,
body: body ? JSON.stringify(body) : undefined,
});
const data = await r.json();
if (!r.ok) throw new Error(data.error.message);
return data;
}
const upload = await call("POST", "/files/upload-url");
await fetch(upload.upload_url, { method: "PUT", body: await readFile(process.argv[2]) });
let job = await call("POST", "/transcripts", { model: "a5s-async-v1", file_id: upload.id, language_code: "en" });
while (job.status === "queued" || job.status === "processing") {
await new Promise((r) => setTimeout(r, 3000));
job = await call("GET", `/transcripts/${job.id}`);
}
for (const u of job.utterances ?? []) console.log(`[${u.start.toFixed(1)}s] ${u.speaker}: ${u.text}`);
# 1. upload URL
curl -X POST https://api.aircaps.com/v1/files/upload-url -H "Authorization: Bearer $AIRCAPS_API_KEY"
# 2. upload the bytes to the returned upload_url
curl -X PUT --upload-file meeting.mp3 "<upload_url>"
# 3. transcribe with the returned id
curl https://api.aircaps.com/v1/transcripts -H "Authorization: Bearer $AIRCAPS_API_KEY" \
-H "Content-Type: application/json" -d '{"model": "a5s-async-v1", "file_id": "file_...", "language_code": "en"}'
Stream a WAV file
# pip install websockets==15.0.1
# AIRCAPS_API_KEY=aircaps_sk_... python stream.py audio.wav (mono 16 kHz PCM16 WAV)
import asyncio, json, os, sys, time, wave
import websockets
async def main():
headers = {"Authorization": f"Bearer {os.environ['AIRCAPS_API_KEY']}"}
async with websockets.connect("wss://api.aircaps.com/v1/realtime", additional_headers=headers) as ws:
await ws.send(json.dumps({"type": "start", "providers": ["a5sv2"], "encoding": "pcm_s16le",
"sample_rate": 16000, "channels": 1, "language": "en-US"}))
while (event := json.loads(await ws.recv()))["type"] != "ready":
if event["type"] == "error":
raise RuntimeError(event["message"])
async def receive():
async for message in ws:
event = json.loads(message)
if event["type"] == "transcript":
print(event["text"], flush=True)
receiver = asyncio.create_task(receive())
with wave.open(sys.argv[1], "rb") as audio:
start, sent = time.monotonic(), 0
while chunk := audio.readframes(320): # 20 ms
await ws.send(chunk)
sent += len(chunk) // 2
await asyncio.sleep(max(0, start + sent / 16000 - time.monotonic()))
await ws.send(json.dumps({"type": "finish"}))
await receiver
asyncio.run(main())
ffmpeg -i input.mp3 -ac 1 -ar 16000 -sample_fmt s16 audio.wav.
Verify a webhook
from flask import Flask, request
from standardwebhooks.webhooks import Webhook
app = Flask(__name__)
wh = Webhook(os.environ["AIRCAPS_WEBHOOK_SECRET"])
@app.post("/hooks/aircaps")
def hook():
event = wh.verify(request.get_data(), dict(request.headers))
print(event["event"], event["id"])
return "", 204