Inverse Text Normalization (ITN)

import asyncio
import websockets
import json
from urllib.parse import urlencode
BASE_WS_URL = "wss://api.smallest.ai/waves/v1/stt/live?model=pulse"
params = {
    "language": "en",
    "encoding": "linear16",
    "sample_rate": "16000",
    "word_timestamps": "true",
    "itn_normalize": "true",
}
WS_URL = f"{BASE_WS_URL}&{urlencode(params)}"
API_KEY = "YOUR_API_KEY"
async def transcribe_file(audio_file: str):
    """Single-shot: stream a fixed audio buffer, then close the session.
    For multi-turn voice agents, use the Multi-turn voice agent example
    further down. It keeps the WebSocket open across user turns.
    """
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
        # Stream the file in 4096-byte chunks
        with open(audio_file, "rb") as f:
            while chunk := f.read(4096):
                await ws.send(chunk)
        # All audio sent. For one-off transcription with no more audio coming,
        # send close_stream. It flushes, runs ITN over the full buffer, emits
        # is_final + is_last, and closes the WebSocket.
        await ws.send(json.dumps({"type": "close_stream"}))
        async for message in ws:
            data = json.loads(message)
            if data.get("is_final"):
                print(f"Final: {data['transcript']}")
                # With ITN: "the total is $25."
                # Without:  "the total is twenty five dollars."
            else:
                print(f"Interim: {data['transcript']}")
            if data.get("is_last"):
                break
asyncio.run(transcribe_file("audio.wav"))

params = {
    "language": "en",
    "encoding": "linear16",
    "sample_rate": "16000",
    "itn_normalize": "true",
    "eou_timeout_ms": "600",        # Match your VAD silence threshold
    "word_timestamps": "true",
}
WS_URL = f"{BASE_WS_URL}&{urlencode(params)}"
async def run_voice_agent():
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
        # Audio producer: stream mic frames as the user speaks across many turns
        async def stream_audio():
            while session_active:
                frame = await mic_queue.get()
                await ws.send(frame)
        # Per-turn finalizer: your VAD calls this when the user pauses
        async def on_user_end_of_turn():
            await ws.send(json.dumps({"type": "finalize"}))
            # Server emits is_final: true with the full ITN-normalized transcript;
            # the WS stays open so the user can speak again.
        # Consumer: receive transcripts (each turn yields one is_final: true frame)
        async def consume():
            async for message in ws:
                data = json.loads(message)
                if data.get("is_final"):
                    print(f"Turn final: {data['transcript']}")
                    # → hand to your LLM, generate the agent reply, speak via TTS,
                    #   and loop back to listening for the next user turn
                if data.get("is_last"):
                    break  # only fires after close_stream below
        # ... wire stream_audio / consume / VAD-driven on_user_end_of_turn together ...
        # End of call (user hung up, agent finished, etc.): close the session
        await ws.send(json.dumps({"type": "close_stream"}))

async def transcribe_once(audio_file: str):
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
        with open(audio_file, "rb") as f:
            while chunk := f.read(4096):
                await ws.send(chunk)
        # All audio sent → close_stream is the only message that emits is_last=true
        await ws.send(json.dumps({"type": "close_stream"}))
        async for message in ws:
            data = json.loads(message)
            if data.get("is_final"):
                print(f"Final: {data['transcript']}")
            if data.get("is_last"):
                break

params = {
    "language": "en",
    "encoding": "linear16",
    "sample_rate": "16000",
    "itn_normalize": "true",
    "redact_pii": "true",           # Redact names, SSN, emails, phone numbers
    "diarize": "true",              # Speaker diarization
    "word_timestamps": "true",
}