Metadata-Version: 2.4
Name: humain-voice
Version: 0.17.1
Summary: Python SDK for HUMAIN Voice services
Author-email: HUMAIN <info@humain.com>, Yousseif Elshahawy <yelshahwy@humain.com>
License-Expression: Apache-2.0
Classifier: Programming Language :: Python :: 3.10
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: aiohttp>=3.12.15
Requires-Dist: anyio>=4.0.0
Requires-Dist: loguru>=0.7.3
Requires-Dist: pydantic>=2.11.7
Requires-Dist: python-socketio[client]>=5.13.0
Provides-Extra: test
Requires-Dist: pytest>=8.0.0; extra == "test"
Requires-Dist: pytest-asyncio>=0.23.0; extra == "test"

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

```bash
pip install humain-voice
```

Requires Python >= 3.10.

### Import path

Use `humain_voice` for new code:

```python
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:

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

### Fast Transcription (sync)

#### Context manager

```python
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)

```python
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

```python
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)

```python
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

```python
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)

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

```python
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:

```python
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:

```python
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:

```python
from humain_voice import stt

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

Realtime streams can collect stable final responses automatically:

```python
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:

```python
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:

```python
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`. Its fields are all **optional**, including `id`, `message`, `code`, `retryable`, `timestamp`, `retry_after_seconds`, `data`, `reason`, and `retry_scope`. The error callback always fires; whether an in-flight context is terminated depends on the **code's ownership** (see [ADR-0003](../../docs/adr/0003-structured-error-contract.md)).

#### Error code constants

```python
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:

```python
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:

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