humain-voice 0.17.0

HUMAIN Voice Python SDK

The official Python SDK for HUMAIN Voice services. It supports:

• Fast Transcription (file-based, async and sync)
• Realtime STT (streaming audio over Socket.IO)
• Batch Transcription (HTTP polling)
• Live Diarization
• ASR subtitle helpers for SRT and WebVTT
• Text-to-Speech

Installation

pip install humain-voice

Requires Python >= 3.10.

Import path

Use humain_voice for new code:

from humain_voice import stt

The legacy import path still works for now, but emits a deprecation warning.

Configuration

Set the following environment variables provided by HUMAIN Voice:

• API_URL: Base URL of the ASR service (e.g. https://api.example.com)
• API_PATH: Socket.IO path. Optional — defaults to /socket.io, the single path that serves every subsystem (realtime, fast transcription, TTS, diarization) on consolidated platform endpoints. Endpoints that have not consolidated yet require an explicit path — in particular, sautech.humain.com serves /realtime/socket.io, not the unprefixed default. Also set it to override for a self-hosted or proxied deployment.
• API_KEY: Your API key

You can export these in your shell:

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

Fast Transcription (sync)

Context manager

import os
from pathlib import Path
from humain_voice 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.BayanArEn,
        )
        print(result)

Manual (no context manager)

import os
from pathlib import Path
from humain_voice 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.BayanArEn,
    )
    print(result1)

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

client.close_sync()

Fast Transcription (async)

Context manager

import os
import asyncio
from pathlib import Path
from humain_voice 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.BayanArEn,
        )
        print(result)

asyncio.run(main())

Manual (no context manager)

import os
import asyncio
from pathlib import Path
from humain_voice 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.BayanArEn,
    )
    print(result1)

    result2 = await client.transcribe(
        audio_bytes,
        stt.Language.ArEn,
        stt.ASRModel.BayanArEn,
    )
    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 humain_voice 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 humain_voice 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 humain_voice.stt import RealtimeDiarizationClient
from humain_voice.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.

Text-to-Speech (TTS)

List the available voices, then synthesize with one of them:

import os
from humain_voice.tts import TTSClient
from humain_voice.tts.constants import TtsModel

with TTSClient(
    api_url=os.getenv("API_URL"),
    api_key=os.getenv("API_KEY"),
) as client:
    voices = client.list_voices_sync(timeout_seconds=5)
    if not voices:
        raise RuntimeError("No voices available")
    print("voices:", [(v["id"], v["label"]) for v in voices])

    audio = client.synthesize_sync(
        "Hello from HUMAIN Voice TTS",
        voice_id=voices[0]["id"],
        model=TtsModel.Nebula,
        timeout_seconds=30,
    )
    with open("out.pcm", "wb") as f:
        f.write(audio)

Async variants (list_voices, synthesize, synthesize_stream) are available on the same client; synthesize_stream yields audio chunks as they arrive.

Subtitles (SRT / VTT)

Fast, realtime, and batch ASR responses include timed word offsets. The subtitle helpers group those words into readable cues and render SRT or WebVTT without adding speaker labels.

Fast and batch responses expose subtitles() directly:

result = await ft_client.transcribe(...)
srt = result.subtitles().to_srt()
vtt = result.subtitles().to_vtt()

batch_result = await batch_client.transcribe(...)
srt = batch_result.subtitles().to_srt()

For one-shot conversion, use the module-level helpers:

from humain_voice import stt

srt = stt.to_srt(result)
vtt = stt.to_vtt(batch_result)

Realtime streams can collect stable final responses automatically:

captions = stt.RealtimeSubtitles()
stream = await rt_client.start_stream(
    language=stt.Language.ArEn,
    subtitles=captions,
)

# send audio, then close the stream
await stream.close()
vtt = captions.to_vtt()

Lower-level helpers are available when you already have words or cues:

cues = stt.words_to_cues(result.words)
srt = stt.cues_to_srt(cues)
vtt = stt.cues_to_vtt(cues)

Speaker labels are never rendered. When batch offsets include speaker metadata, speaker changes are used only as cue boundaries.

Defaults follow common caption readability heuristics:

• up to two lines per cue
• 42 characters per line
• splits on sentence endings, long pauses, long cues, and speaker changes
• forgiving cleanup for common ASR timing issues

Use strict=True when you want validation errors instead of normalization:

captions = result.subtitles(strict=True)
vtt = captions.to_vtt(strict=True)

Strict validation applies to both SRT and VTT output.

Object render methods are intentionally named as conversions: to_srt() and to_vtt(). The lower-level helpers keep input-specific names such as cues_to_srt() and words_to_cues().

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 humain_voice.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 humain_voice.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 humain_voice.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 humain_voice.stt, for example Language and ASRModel.

Examples

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