humain-voice 0.16.0a5

Python SDK for SauTech Services

Sautech Python SDK

The official Python SDK for SauTech voice services. It supports:

• Fast Transcription (file-based, async and sync)
• Realtime STT (streaming audio over Socket.IO)
• Text-to-Speech
• Batch Transcription (HTTP polling)

Installation

pip install humain-voice

Requires Python >= 3.10.

Configuration

Set the following environment variables provided by SauTech:

• API_URL: Base URL of the ASR service (e.g. https://api.example.com)
• API_PATH: Socket.IO path for the service you are calling. Required, with no default — each service has its own path (e.g. /realtime/socket.io for realtime, /fast-transcription/socket.io for fast transcription, /tts/socket.io for TTS). It must be passed to every client.
• API_KEY: Your API key

You can export these in your shell (the path below is the realtime example):

export API_URL="https://api.example.com"
export API_PATH="/realtime/socket.io"
export API_KEY="<your_api_key>"

Fast Transcription (sync)

Context manager

import os
from pathlib import Path
from sautech import stt

audio_path = Path("samples/king_fahad_test.wav")

with stt.FastTranscriptionClient(
    api_url=os.getenv("API_URL"),
    api_key=os.getenv("API_KEY"),
    api_path=os.getenv("API_PATH"),
) as client:
    with audio_path.open("rb") as f:
        result = client.transcribe_sync(
            f,  # file-like object or bytes via f.read()
            stt.Language.ArEn,
            stt.ASRModel.BayanArEnV1,
        )
        print(result)

Manual (no context manager)

import os
from pathlib import Path
from sautech import stt

audio_path = Path("samples/king_fahad_test.wav")

client = stt.FastTranscriptionClient(
    api_url=os.getenv("API_URL"),
    api_key=os.getenv("API_KEY"),
    api_path=os.getenv("API_PATH"),
)

with audio_path.open("rb") as f:
    result1 = client.transcribe_sync(
        f,
        stt.Language.ArEn,
        stt.ASRModel.BayanArEnV1,
    )
    print(result1)

with audio_path.open("rb") as f:
    result2 = client.transcribe_sync(
        f.read(),
        stt.Language.ArEn,
        stt.ASRModel.BayanArEnV1,
    )
    print(result2)

client.close_sync()

Fast Transcription (async)

Context manager

import os
import asyncio
from pathlib import Path
from sautech import stt

async def main():
    audio_bytes = Path("samples/king_fahad_test.wav").read_bytes()
    async with stt.FastTranscriptionClient(
        api_url=os.getenv("API_URL"),
        api_key=os.getenv("API_KEY"),
        api_path=os.getenv("API_PATH"),
    ) as client:
        result = await client.transcribe(
            audio_bytes,
            stt.Language.ArEn,
            stt.ASRModel.BayanArEnV1,
        )
        print(result)

asyncio.run(main())

Manual (no context manager)

import os
import asyncio
from pathlib import Path
from sautech import stt

async def main():
    audio_bytes = Path("samples/king_fahad_test.wav").read_bytes()

    client = stt.FastTranscriptionClient(
        api_url=os.getenv("API_URL"),
        api_key=os.getenv("API_KEY"),
        api_path=os.getenv("API_PATH"),
    )

    result1 = await client.transcribe(
        audio_bytes,
        stt.Language.ArEn,
        stt.ASRModel.BayanArEnV1,
    )
    print(result1)

    result2 = await client.transcribe(
        audio_bytes,
        stt.Language.ArEn,
        stt.ASRModel.BayanArEnV1,
    )
    print(result2)

    await client.close()

asyncio.run(main())

You can also pass on_response, on_file_upload, and on_error callbacks to receive intermediate updates and handle lifecycle events during processing.

Realtime Streaming

import asyncio
import os
import wave
from sautech import stt

async def run():
    client = stt.RealtimeClient(
        api_url=os.getenv("API_URL"),
        api_key=os.getenv("API_KEY"),
        api_path=os.getenv("API_PATH"),
    )

    stream = await client.start_stream(
        language=stt.Language.ArEn,
        on_connect=lambda: print("connected"),
        on_disconnect=lambda: print("disconnected"),
        on_response=lambda t: print("response:", t),
        on_error=lambda e: print("error:", e),
    )

    with wave.open("samples/king_fahad_test.wav", "rb") as f:
        audio_bytes = f.readframes(f.getnframes())

    chunk_duration_s = 0.1
    sample_rate = 16000
    bytes_per_sample = 2  # 16-bit PCM
    chunk_size = int(sample_rate * bytes_per_sample * chunk_duration_s)

    for start in range(0, len(audio_bytes), chunk_size):
        end = min(start + chunk_size, len(audio_bytes))
        await stream.send(audio_bytes[start:end])
        await asyncio.sleep(chunk_duration_s)

    await stream.close(timeout_seconds=1)

asyncio.run(run())

Parallel streams (single client)

import asyncio
import os
import wave
from sautech import stt

async def stream_audio(stream, audio_bytes: bytes):
    chunk_duration_s = 0.1
    sample_rate = 16000
    bytes_per_sample = 2
    chunk_size = int(sample_rate * bytes_per_sample * chunk_duration_s)

    for start in range(0, len(audio_bytes), chunk_size):
        end = min(start + chunk_size, len(audio_bytes))
        await stream.send(audio_bytes[start:end])
        await asyncio.sleep(chunk_duration_s)

    await stream.close(timeout_seconds=1)

async def run():
    client = stt.RealtimeClient(
        api_url=os.getenv("API_URL"),
        api_key=os.getenv("API_KEY"),
        api_path=os.getenv("API_PATH"),
    )

    stream_a = await client.start_stream(
        language=stt.Language.ArEn,
        on_response=lambda t: print("stream A:", t),
    )
    stream_b = await client.start_stream(
        language=stt.Language.ArEn,
        on_response=lambda t: print("stream B:", t),
    )

    with wave.open("samples/king_fahad_test.wav", "rb") as f:
        audio_bytes = f.readframes(f.getnframes())

    await asyncio.gather(
        stream_audio(stream_a, audio_bytes),
        stream_audio(stream_b, audio_bytes),
    )

asyncio.run(run())

Live Diarization

A standalone RealtimeDiarizationClient identifies speakers in real time over the same realtime socket, accumulating a full per-speaker timeline as audio arrives.

import asyncio
import os
import wave
from sautech.stt import RealtimeDiarizationClient
from sautech.stt.constants import DIARIZATION_RECOMMENDED_CHUNK_BYTES

async def run():
    client = RealtimeDiarizationClient(
        api_url=os.getenv("API_URL"),
        api_key=os.getenv("API_KEY"),
    )

    stream = await client.start_stream(
        on_error=lambda err: print("stream error:", err),
    )

    async def feed():
        with wave.open("samples/king_fahad_test.wav") as w:
            pcm = w.readframes(w.getnframes())
        for off in range(0, len(pcm), DIARIZATION_RECOMMENDED_CHUNK_BYTES):
            await stream.send(pcm[off : off + DIARIZATION_RECOMMENDED_CHUNK_BYTES])
            await asyncio.sleep(0.48)  # 480 ms real-time pace
        timeline = await stream.close()
        print("final timeline:", timeline)

    feeder = asyncio.create_task(feed())
    async for update in stream:
        print(
            f"segments={len(update.segments)} "
            f"newly_finalized={len(update.newly_finalized)} "
            f"actives={len(update.active_segments)} final={update.is_final}"
        )
    await feeder

asyncio.run(run())

Notes:

• Stream raw PCM16LE mono 16 kHz. The recommended chunk size is 15360 bytes (480 ms = one model inference step). Other sizes are accepted but produce a less steady result cadence.
• final_segments arrive as per-response deltas; the SDK accumulates them. update.segments is the full best-known timeline; update.newly_finalized is the per-response delta.
• Active segments may be revised on any update and may overlap (simultaneous speakers).
• stream.close() sends the terminator frame, waits up to 5 s, and returns the best-known timeline instead of throwing on timeout.

Error handling

The SDK exposes a structured error contract aligned with the platform's ErrorResponse. The platform fields id, message, code, retryable, and timestamp are all optional — server paths exist that omit any of them. The error callback always fires; whether an in-flight context is terminated depends on the code's ownership (see ADR-0003).

Error code constants

from sautech.errors import (
    ASR_TRANSCRIPTION_FAILED,
    ASR_MODEL_UNAVAILABLE,
    RATE_LIMIT_EXCEEDED,
    TTS_VOICE_LIST_FAILED,
    SERVER_INTERNAL,
    is_asr_code,
    is_tts_code,
    is_request_scoped_code,
    is_realtime_owned,
    is_tts_owned,
)

Unknown future codes pass through as strings. Legacy aliases such as RATE_LIMITED, VALIDATION_FAILED, and INTERNAL_ERROR remain exported for older deployments.

Socket.IO errors (TTS / realtime STT / fast STT)

The on_error callback receives an ErrorResponse model. Branch on code to decide your policy:

from sautech.errors import ASR_TRANSCRIPTION_FAILED, ASR_MODEL_UNAVAILABLE

def on_error(err):
    # err is always non-None; any field can be None.
    print(f"code={err.code} retryable={err.retryable} msg={err.message}")
    if err.code == ASR_MODEL_UNAVAILABLE:
        # Tell the user the model is down; safe to retry later.
        ...
    elif err.code == ASR_TRANSCRIPTION_FAILED:
        # Final terminal error for this stream — no retry.
        ...

The realtime adapter only terminates an ASR stream context for ASR-shaped codes. A TTS_VOICE_LIST_FAILED arriving on the realtime socket fires the global on_error but does not kill the stream.

Batch transcription HTTP errors

BatchTranscribeError (and its subclasses) carry a structured payload plus per-field accessors:

from sautech.stt.batchtranscription import (
    BatchTranscribeClient,
    BatchTranscribeError,
    BatchTranscribeRateLimitError,
)

try:
    result = await client.submit(audio, "ar")
except BatchTranscribeRateLimitError as err:
    # Caller decides retry policy; SDK never retries internally.
    print(f"rate-limited; retry_after={err.retry_after}s capacity={err.capacity}")
except BatchTranscribeError as err:
    print(f"status={err.status_code}")
    print(f"code={err.code} retryable={err.retryable}")
    print(f"detail={err.detail} job_id={err.job_id}")
    print(f"raw_body={err.raw_body!r}")  # always preserved

User-facing message preference: detail → message → error → raw text. err.message already follows that order.

Note. The max_retries constructor argument is deprecated and is a no-op. The SDK never retries — callers decide policy based on err.retryable, err.code, and err.capacity.

Types

Common enums are available under sautech.stt, for example Language and ASRModel.

Examples

See complete examples in python/examples/ft_client.py and python/examples/rt_client.py. The batch transcription example (python/examples/batch_transcribe_client.py) shows structured-error handling end-to-end. python/examples/diarization_client.py demonstrates live diarization streaming (feeder + async-iterator pattern).