"""Measure paid native-key REST and real-time PCM streaming, without retries. Requires httpx and websockets. Supply ORUK_API_KEY or --key-file and a JSONL manifest containing audio_filepath and duration. Input must be public or your own mono 16 kHz PCM16 WAV audio. Raw receipts contain request IDs, never keys. """ import argparse import asyncio import hashlib import io import json import math import os from pathlib import Path import platform import time import uuid import wave import httpx import websockets def distribution(values): values = sorted(values) if not values: return {"n": 0} def quantile(p): at = (len(values) - 1) * p low = int(at) return values[low] + (values[min(low + 1, len(values) - 1)] - values[low]) * (at - low) return {"n": len(values), "p50": quantile(.5), "p95": quantile(.95), "p99": quantile(.99), "min": values[0], "max": values[-1]} def fixture(path): audio = Path(path).read_bytes() with wave.open(io.BytesIO(audio)) as reader: assert (reader.getnchannels(), reader.getsampwidth(), reader.getframerate()) == (1, 2, 16000) pcm = reader.readframes(reader.getnframes()) return {"audio": audio, "pcm": pcm, "duration": len(pcm) / 32000, "sha256": hashlib.sha256(audio).hexdigest()} def summarize(rows, elapsed): successes = [row for row in rows if row["ok"]] seconds = sum(row["audio_seconds"] for row in successes) return {"requests": len(rows), "successes": len(successes), "errors": len(rows) - len(successes), "elapsed_seconds": elapsed, "requests_per_second": len(successes) / elapsed, "successful_audio_seconds": seconds, "audio_seconds_per_second": seconds / elapsed, **{name: distribution([row[name] for row in successes if name in row]) for name in ("client_ms", "asr_ms", "server_ms")}} async def benchmark(args): key = Path(args.key_file).read_text().strip() if args.key_file else os.environ["ORUK_API_KEY"] manifest = Path(args.manifest) candidates = [json.loads(line) for line in manifest.read_text().splitlines() if line.strip()] selected = [] for target in (1.5, 2, 3, 4, 5, 6, 8, 10, 12, 16, 20, 25): chosen = min((r for r in candidates if r not in selected), key=lambda r: abs(r["duration"] - target)) selected.append(chosen) clips = [fixture(r["audio_filepath"]) for r in selected] prefix = "orukeet-bench-" + uuid.uuid4().hex[:12] + "-" report = {"started_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "request_prefix": prefix, "client": {"platform": platform.platform(), "httpx": httpx.__version__, "websockets": websockets.__version__, "http_version": "HTTP/1.1"}, "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "fixtures": [{k: c[k] for k in ("sha256", "duration")} for c in clips], "method": "Closed-loop concurrency; preloaded PCM16 WAV; native Oruk key; warm model; connection reuse; no retries or outlier removal. REST measures request start through complete JSON response. Streaming sends 20 ms frames at real-time cadence; measures final frame send through final text, excluding connection setup and recording. Percentiles use linear interpolation. Warmups retained separately.", "rest": [], "streaming": [], "optional_tasks": [], "warmups": []} output = Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) journal = output.with_suffix(".jsonl").open("w") receipts = [] def record(row): receipts.append(row) journal.write(json.dumps(row) + "\n") journal.flush() def save(): output.write_text(json.dumps({**report, "receipts": receipts}, indent=2) + "\n") async def post(client, base, clip, group, flags=None): ident = prefix + uuid.uuid4().hex row = {"id": ident, "group": group, "fixture_sha256": clip["sha256"], "audio_seconds": clip["duration"], "ok": False} started = time.perf_counter() try: response = await client.post(base + "/v1/audio/transcriptions", headers={"Authorization": "Bearer " + key, "X-Request-ID": ident}, data={"model": "oruk-orukeet", **(flags or {})}, files={"file": ("audio.wav", clip["audio"], "audio/wav")}) row.update(client_ms=(time.perf_counter() - started) * 1000, status=response.status_code) data = response.json() if response.status_code == 200: row.update({k: data[k] for k in ("asr_ms", "server_ms")}) row["cost_microusd"] = data["usage"]["cost_microusd"] rate = .0001 + sum(rate for flag, rate in (("emotion_detection", .008), ("diarize", .004)) if (flags or {}).get(flag) == "true") assert row["cost_microusd"] == math.ceil(clip["duration"] * rate / 60 * 1_000_000 - 1e-8) assert abs(data["duration"] - clip["duration"]) < .001 assert isinstance(data["text"], str) and data["text"].strip() if (flags or {}).get("emotion_detection"): assert len(data["emotion_detection"]["scores"]) == 15 if (flags or {}).get("diarize"): assert data["segments"] and data["speakers"] row["ok"] = True else: row["error_code"] = data.get("error", {}).get("code", "http_error") except Exception as exc: row.update(client_ms=(time.perf_counter() - started) * 1000, error_type=type(exc).__name__) record(row) return row try: for base in args.base: async with httpx.AsyncClient(timeout=90, limits=httpx.Limits(max_connections=8, max_keepalive_connections=8, keepalive_expiry=120)) as client: warmups = await asyncio.gather(*(post(client, base, clips[i], base + "-warmup") for i in range(8))) report["warmups"].append({"base_url": base, "requests": len(warmups), "errors": sum(not x["ok"] for x in warmups)}) for concurrency in (1, 2, 4, 8): group = f"{base}-c{concurrency}" rows = [] cursor = 0 async def worker(): nonlocal cursor while cursor < args.requests: index = cursor cursor += 1 rows.append(await post(client, base, clips[index % len(clips)], group)) started = time.perf_counter() await asyncio.gather(*(worker() for _ in range(concurrency))) result = {"base_url": base, "concurrency": concurrency, **summarize(rows, time.perf_counter() - started)} report["rest"].append(result) save() print(json.dumps(result), flush=True) stream_clips = [clips[i] for i in (0, 1, 2, 4, 6, 8)] started = time.perf_counter() async with websockets.connect(args.websocket, additional_headers={"Authorization": "Bearer " + key}, compression=None, open_timeout=30) as ws: assert json.loads(await ws.recv())["type"] == "ready" report["websocket"] = {"url": args.websocket, "handshake_and_ready_ms": (time.perf_counter() - started) * 1000, "fixture_sha256": [c["sha256"] for c in stream_clips]} rows = [] phase_start = time.perf_counter() for index in range(args.stream_requests): clip = stream_clips[index % len(stream_clips)] ident = prefix + uuid.uuid4().hex row = {"id": ident, "group": "streaming", "fixture_sha256": clip["sha256"], "audio_seconds": clip["duration"], "ok": False} await ws.send(json.dumps({"type": "configure", "request_id": ident, "options": {}})) assert json.loads(await ws.recv())["type"] == "configured" start = time.perf_counter() for offset in range(0, len(clip["pcm"]), 640): frame = clip["pcm"][offset:offset + 640] await asyncio.sleep(max(0, start + (offset + len(frame)) / 32000 - time.perf_counter())) last = time.perf_counter() await ws.send(frame) await ws.send(json.dumps({"type": "commit"})) result = json.loads(await asyncio.wait_for(ws.recv(), timeout=90)) row["client_ms"] = (time.perf_counter() - last) * 1000 if result["type"] == "final": row.update({k: result[k] for k in ("asr_ms", "server_ms")}) row["cost_microusd"] = result["usage"]["cost_microusd"] assert row["cost_microusd"] == math.ceil(clip["duration"] / 60 * 100 - 1e-8) row["ok"] = bool(result["text"].strip()) else: row["error_code"] = result.get("code", result["type"]) record(row) rows.append(row) if (index + 1) % 10 == 0: print(json.dumps({"streaming_completed": index + 1}), flush=True) save() if not row["ok"]: break # Do not silently reconnect or retry a failed session. report["streaming"] = summarize(rows, time.perf_counter() - phase_start) if args.task_audio: task_clip = fixture(args.task_audio) report["task_fixture"] = {k: task_clip[k] for k in ("duration", "sha256")} async with httpx.AsyncClient(timeout=90) as client: for flags in ({"emotion_detection": "true"}, {"diarize": "true"}, {"emotion_detection": "true", "diarize": "true"}): rows = [] started = time.perf_counter() for _ in range(5): rows.append(await post(client, args.base[0], task_clip, "+".join(flags), flags)) report["optional_tasks"].append({"flags": flags, **summarize(rows, time.perf_counter() - started)}) save() report["completed_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) report["ok"] = all(row["ok"] for row in receipts) finally: save() journal.close() print(json.dumps({"ok": report["ok"], "total_requests": len(receipts), "output": str(output)}), flush=True) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--key-file") parser.add_argument("--manifest", required=True) parser.add_argument("--output", required=True) parser.add_argument("--requests", type=int, default=120, help="Measured requests per REST route and concurrency") parser.add_argument("--stream-requests", type=int, default=60) parser.add_argument("--base", action="append", default=None) parser.add_argument("--websocket", default="wss://orukeet-direct.oruk.ai/v1/audio/transcriptions/stream") parser.add_argument("--task-audio") args = parser.parse_args() args.base = args.base or ["https://speech-api.oruk.ai", "https://orukeet-direct.oruk.ai"] if args.requests < 1 or args.stream_requests < 1: parser.error("Request counts must be positive") asyncio.run(benchmark(args))