humain-voice 0.16.1b2

Python SDK for HUMAIN Voice services

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.

Changelog

Release notes are maintained in the repository root CHANGELOG.md. The PyPI README is generated during publish and includes the full changelog.

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 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 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.

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).

---

Changelog

All notable changes to the HUMAIN Voice SDK (Python humain-voice, JS @humain-voice/sdk, Go module) are documented here. The format follows Keep a Changelog, and the project adheres to Semantic Versioning.

Versions are shared across all three language packages; a release tags python/vX.Y.Z, javascript/vX.Y.Z, and golang/vX.Y.Z together. The Teams release notification posts the section matching the tag's version, so keep the top section here current before tagging.

[Unreleased]

[0.16.1-beta.2] - 2026-06-18

Changed

• Rebranded repository, package, and release documentation to HUMAIN Voice across the root, Python, JavaScript, and Go docs.
• Python, JavaScript, and Go examples, comments, and release metadata now use HUMAIN Voice wording instead of the old product name.
• The deprecated JavaScript compatibility package now emits HUMAIN_VOICE_COMPAT_PACKAGE_DEPRECATED and uses neutral compatibility wording in its warning message.
• The legacy Python import path now emits HumainVoiceCompatibilityWarning with neutral compatibility wording.
• Example package manifests now depend on the registry package names instead of local source paths that expose old package-directory names.
• Archived live-diarization planning docs were condensed so the public docs no longer preserve obsolete package paths.

[0.16.1-beta.1] - 2026-06-18

Added

• The Python SDK can now be imported as humain_voice, matching the humain-voice distribution name.
• The deprecated JavaScript compatibility package now wraps @humain-voice/sdk.
• PyPI and npm publish jobs now generate registry READMEs that append this root changelog, so package registry pages show release notes from the same source as the repository.
• The Go module now includes a checked-in golang/CHANGELOG.md mirror so tagged Go source includes release notes for Go consumers and pkg.go.dev.

Changed

• JavaScript examples now import @humain-voice/sdk.
• Go dependencies were refreshed and the Go CI/container baseline moved to Go 1.25.
• Go README and release notes no longer reference the non-existent pkg/fasttranscription compatibility wrapper.

Deprecated

• The legacy Python import path remains available for compatibility, but now emits a deprecation warning. New code should import from humain_voice.
• The legacy JavaScript compatibility package remains available, but now emits a deprecation warning. New code should import from @humain-voice/sdk.

[0.16.1] - 2026-06-15

Changed

• We now recommend selecting the Arabic/English model with the unversioned BayanArEn constant instead of pinning to BayanArEnV1. The unversioned constant always maps to the current production model, so you automatically benefit from accuracy improvements without a code change. If you have a specific reason to stay on a fixed model version, the BayanArEnV1 and BayanArEnV2 constants remain available — nothing you ship today breaks.

[0.16.0] - 2026-06-11

⚠️ Breaking Changes

• ASR and diarization model identifiers were renamed. The SDK constants used to select a model changed, so any code that selects a model must be updated before upgrading. Functionality is unchanged; these are the same models under new, brand-aligned names. Always select models through the SDK constants below rather than raw strings. Migration from 0.15.x:

  What you used (0.15.x)	New constant
  ASRModel.Ar_1	ASRModel.NidaAr
  ASRModel.Ar_2	ASRModel.BayanAr
  ASRModel.Ar_8k_1	ASRModel.NidaArTelephony
  ASRModel.ArEn	ASRModel.BayanArEn
  ASRModel.ArEn_1	ASRModel.BayanArEnV1
  ASRModel.ArEn_2	ASRModel.BayanArEnV2
  BatchDiarization.M1	BatchDiarization.D1
  BatchDiarization.M2	BatchDiarization.D2

  Go callers use the package-prefixed forms (e.g. ASRModelBayanArEnV1, BatchDiarizationD1). The old constants are no longer accepted — code using them will need to migrate to the new names.

• The batch transcription client no longer retries failed requests automatically. Previous versions retried internally (via tenacity in Python and p-retry in JS). The client now makes exactly one HTTP call per operation and raises immediately on failure. The max_retries / maxRetries constructor option is kept for compatibility but is now a no-op; passing a non-zero value emits a deprecation warning. If you depend on retries, you must now implement your own policy — every raised error exposes retryable, retry_after, and capacity (for HTTP 429) so you can back off correctly.

• Error handling moved to a structured, typed exception hierarchy. Errors are now routed by error code to the client that owns them, and failures raise a specific exception type instead of a generic error. For batch transcription: HTTP 401 raises an auth error, HTTP 429 raises a rate-limit error (carrying retry_after and capacity), and all other failures raise the base BatchTranscribeError. The full server payload is available on the exception (payload / raw_body) for inspection. If you currently catch a broad error type or parse error strings, switch to catching the typed exceptions and reading their attributes.

Added

• Realtime speaker diarization over the streaming transport — receive per-speaker labels on transcripts as audio streams in.
• SRT and VTT subtitle helpers in Python, JS, and Go to turn transcription results directly into ready-to-use subtitle files.

[0.15.1] - 2026-05-13

Added

• Official Go SDK, at full feature parity with the Python and JavaScript packages: realtime speech-to-text, fast transcription, text-to-speech, and batch transcription, all over the same streaming transport. If you work in Go, you no longer need to call the platform API directly.

[0.15.0] - 2026-05-06

⚠️ Breaking Changes

• Client imports, event names, and configuration shapes changed. Internally the STT and TTS clients were rebuilt on a shared connection layer, and that shifted parts of the public surface — some import paths, streaming event names, and configuration option shapes are different. Review your client setup and event handlers against the updated examples when upgrading from 0.14.x.
• disconnect_socketio was removed. Connection teardown is now handled by the client itself; if you imported disconnect_socketio directly, drop the call — closing the client cleans up the connection for you.

Changed

• STT and TTS clients now share a single connection and lifecycle, which makes connection handling more consistent and predictable across both. This is an internal change, but it is the reason for the surface adjustments noted above.

[0.14.5] - 2026-02-03

Added

• Run multiple concurrent streams from a single RealtimeClient. You can now open several independent realtime sessions from one client instance instead of constructing a separate client per stream.

Changed

• Invalid ASR and TTS request fields are now rejected up front. The SDK validates request fields — including voice-reference inputs for TTS — before sending, so mistakes surface immediately as a clear client-side error rather than failing partway through a request. Requests that previously appeared to start before failing server-side may now raise a validation error sooner; fix the flagged field and the request proceeds as normal.