Triton ASR Client

Async Python client for sending audio to a Triton ASR service using SSE or WebSocket streaming. It yields real-time transcription results as JSON-like events.

---

Features

• 🚀 Async SSE & WebSocket streaming modes
• 🎧 Streams audio bytes directly
• 🧩 Yields parsed dict events (partial/final text, status, errors)
• 🔧 Simple integration with Triton Inference Server

---

Requirements

• Python 3.10+
• tritonclient[grpc], tritonclient[http], websockets

Install

pip install git+https://git.d.aiengines.ir/bi/asr_triton_client.git@05c8f7a88d3ff41d82591b6beb41bab86d81d421

---

Environment

Set your Triton endpoint (default: localhost:8001):

export TRITON_URL="your.triton.server:8001"

---

Example — SSE Mode

import asyncio
from .service import TritonGrpcClient
import json, os
from pathlib import Path

TRITON_URL = os.getenv("TRITON_URL", "localhost:8001")
client = TritonGrpcClient(triton_url=TRITON_URL)

async def asr_sse_mode(client, audio_path: Path):
    print("=== SSE MODE ===")
    raw = audio_path.read_bytes()
    try:
        async for s in client.event_stream_from_bytes(raw=raw, filename=audio_path.name):
            print("[SSE]", json.dumps(s, ensure_ascii=False))
    except Exception as exc:
        print("SSE error:", exc)

asyncio.run(asr_sse_mode(client, Path("/test_audio_file.mp3")))

---

Example — WebSocket Mode

import asyncio
from .service import TritonGrpcClient
import json, os
from pathlib import Path

TRITON_URL = os.getenv("TRITON_URL", "localhost:8001")
client = TritonGrpcClient(triton_url=TRITON_URL)

async def asr_ws_mode(client, audio_path: Path):
    print("=== WS MODE ===")
    raw = audio_path.read_bytes()
    try:
        async for s in client.event_stream_via_ws(raw=raw, filename=audio_path.name):
            print("[WS]", json.dumps(s, ensure_ascii=False))
    except Exception as exc:
        print("WS error:", exc)

asyncio.run(asr_ws_mode(client, Path("/test_audio_file.mp3")))

---

Example Output

Below is a real example of SSE output while streaming an MP3 file:

[BG] Entering stream_transcript loop
[SSE] OUT: "data: {"time": 1760512211.4691734, "text": "سلام وقت", "is_final": false}\n\n"
[SSE] OUT: "data: {"time": 1760512211.55668, "text": "به‌خیر", "is_final": false}\n\n"
...
[SSE] OUT: "data: {"time": 1760512226.2526345, "text": "سلام وقت‌به‌خیر امروز درباره طراحی جدید صحبت می‌کنیم", "is_final": true}\n\n"

Each event contains:

• time: event timestamp
• text: recognized speech fragment
• is_final: indicates final transcript segment

---

API Overview

Method	Description
TritonGrpcClient(triton_url)	Create a client connected to Triton
event_stream_from_bytes(raw, filename)	SSE-based audio streaming
event_stream_via_ws(raw, filename)	WebSocket-based audio streaming

Each stream yields dict events like:

{
  "time": 1760512211.47,
  "text": "hello world",
  "is_final": True
}

---

Troubleshooting

• Connection refused → Check TRITON_URL and Triton server status
• Bad event data → Verify model/gateway returns valid JSON events
• WS handshake failed → Ensure the server supports WebSocket

---