Python

python -m pip install humain-voice==0.18.0

export API_URL="https://api.voice.humain.com"
export API_KEY="YOUR_API_KEY"
export API_VERSION="v1"

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path

from humain_voice import stt
from humain_voice.stt.batchtranscription import BatchDiarization


def required_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


async def main() -> None:
    input_path = Path(sys.argv[1] if len(sys.argv) > 1 else "meeting.wav")
    output_path = Path(sys.argv[2] if len(sys.argv) > 2 else "meeting.vtt")

    async with stt.BatchTranscribeClient(
        api_url=required_env("API_URL"),
        api_key=required_env("API_KEY"),
        api_version=os.environ.get("API_VERSION", "v1"),
    ) as client:
        result = await client.transcribe(
            input_path,
            lang=stt.Language.ArEn,
            asr=stt.BatchTranscriptionModel.BayanArEn,
            diarization=BatchDiarization.On,
            save_result=True,
            poll_interval=2.0,
            timeout_seconds=300.0,
            on_progress=lambda response: print("status:", response.status.value),
        )

    print(result.results.transcript if result.results else "")
    output_path.write_text(
        stt.Subtitles.from_response(result).to_vtt(),
        encoding="utf-8",
    )


if __name__ == "__main__":
    asyncio.run(main())

python batch_transcription.py meeting.wav meeting.vtt

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path

from humain_voice import stt


def required_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


def handle_response(response: stt.FtTranscribeResponse) -> None:
    kind = "final" if response.is_final else "partial"
    print(f"{kind}:", response.transcription)


def handle_upload(response: stt.FileUploadedResponse) -> None:
    print("uploaded:", response.id)


def handle_error(error: stt.ErrorResponse | None) -> None:
    if error is not None:
        print("server error:", error.code, error.message)


async def main() -> None:
    input_path = Path(sys.argv[1] if len(sys.argv) > 1 else "short-call.wav")
    output_path = Path(sys.argv[2] if len(sys.argv) > 2 else "short-call.srt")

    async with stt.FastTranscriptionClient(
        api_url=required_env("API_URL"),
        api_path=required_env("API_PATH"),
        api_key=required_env("API_KEY"),
    ) as client:
        result = await client.transcribe(
            input_path.read_bytes(),
            stt.Language.Ar,
            stt.FastTranscriptionModel.BayanAr,
            on_response=handle_response,
            on_file_upload=handle_upload,
            on_error=handle_error,
            timeout_seconds=60.0,
        )

    if result is None:
        raise RuntimeError("Fast transcription ended without a final result")
    output_path.write_text(
        stt.Subtitles.from_response(result).to_srt(),
        encoding="utf-8",
    )


if __name__ == "__main__":
    asyncio.run(main())

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path

from humain_voice import stt


CHUNK_BYTES = 3_200  # 100 ms of PCM16LE, 16 kHz, mono audio.


def required_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


async def main() -> None:
    input_path = Path(sys.argv[1] if len(sys.argv) > 1 else "speech.pcm")
    output_path = Path(sys.argv[2] if len(sys.argv) > 2 else "speech.vtt")
    finalized_words: list[stt.WordSegment] = []
    server_error: stt.ErrorResponse | None = None
    protocol_final_observed = False

    def handle_response(response: stt.RtTranscribeResponse) -> None:
        nonlocal protocol_final_observed
        if response.is_final:
            kind = "final"
        elif response.is_speech_final:
            kind = "speech-final"
        else:
            kind = "partial"
        print(f"{kind}:", response.transcription)
        if response.is_final:
            protocol_final_observed = True
        if response.is_final or response.is_speech_final:
            # The current public Realtime contract does not guarantee increasing
            # seq values, so collect final words in arrival order instead of
            # asking RealtimeSubtitles to deduplicate by id:seq.
            finalized_words.extend(response.words)

    def handle_error(error: stt.ErrorResponse | None) -> None:
        # The released SDK can invoke a stream handler more than once for one
        # routed error, so keep this callback idempotent.
        nonlocal server_error
        server_error = error

    client = stt.RealtimeClient(
        api_url=required_env("API_URL"),
        api_path=required_env("API_PATH"),
        api_key=required_env("API_KEY"),
    )
    async with client:
        stream = await client.start_stream(
            language=stt.Language.ArEn,
            on_response=handle_response,
            on_error=handle_error,
        )
        pcm = input_path.read_bytes()
        for offset in range(0, len(pcm), CHUNK_BYTES):
            await stream.send(pcm[offset : offset + CHUNK_BYTES])
            await asyncio.sleep(0.1)


        # close() sends the last frame and waits for protocol is_final, a routed
        # error, or this timeout. It returns rather than raising on timeout.
        await stream.close(timeout_seconds=5.0)

    if server_error is not None:
        raise RuntimeError(server_error.message or server_error.code or "Realtime stream failed")
    if not protocol_final_observed:
        raise RuntimeError("Realtime stream ended before protocol is_final")
    output_path.write_text(
        stt.Subtitles.from_words(finalized_words).to_vtt(),
        encoding="utf-8",
    )


if __name__ == "__main__":
    asyncio.run(main())

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path

from humain_voice import stt
from humain_voice.stt.constants import DIARIZATION_RECOMMENDED_CHUNK_BYTES


def required_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


async def main() -> None:
    input_path = Path(sys.argv[1] if len(sys.argv) > 1 else "meeting.pcm")
    output_path = Path(sys.argv[2] if len(sys.argv) > 2 else "meeting.rttm")
    client = stt.RealtimeDiarizationClient(
        api_url=required_env("API_URL"),
        api_path=required_env("API_PATH"),
        api_key=required_env("API_KEY"),
    )

    async with client:
        final_observed = False

        def on_update(update: stt.DiarizationUpdate) -> None:
            nonlocal final_observed
            final_observed = final_observed or update.is_final
            for segment in update.newly_finalized:
                print(segment.speaker, segment.start_time, segment.end_time)

        stream = await client.start_stream(
            on_update=on_update,
            on_error=lambda error: print("server error:", error),
        )
        pcm = input_path.read_bytes()
        if not pcm or len(pcm) % 2:
            raise ValueError("Input must be nonempty PCM16 with an even byte length")
        for offset in range(0, len(pcm), DIARIZATION_RECOMMENDED_CHUNK_BYTES):
            await stream.send(
                pcm[offset : offset + DIARIZATION_RECOMMENDED_CHUNK_BYTES]
            )
            await asyncio.sleep(0.48)

        # close() returns the best-known reconciled timeline after five seconds,
        # even when no is_final update arrived. A callback avoids leaving an async
        # iterator waiting forever on that timeout path.
        timeline = await stream.close(timeout_seconds=5.0)

    destination = output_path if final_observed else Path(f"{output_path}.partial")
    destination.write_text(stt.to_rttm(timeline, uri="meeting"), encoding="utf-8")
    if not final_observed:
        print(f"Final result not observed; wrote incomplete output to {destination}")


if __name__ == "__main__":
    asyncio.run(main())

from __future__ import annotations

import asyncio
import os
import sys
import wave
from pathlib import Path

from humain_voice import stt, tts


def required_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


def write_pcm16_wav(path: Path, pcm: bytes, sample_rate: int) -> None:
    with wave.open(str(path), "wb") as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(sample_rate)
        wav_file.writeframes(pcm)


def handle_error(error: stt.ErrorResponse | None) -> None:
    if error is not None:
        print("server error:", error.code, error.message)


async def main() -> None:
    output_path = Path(sys.argv[1] if len(sys.argv) > 1 else "speech.wav")
    async with tts.TTSClient(
        api_url=required_env("API_URL"),
        api_path=required_env("API_PATH"),
        api_key=required_env("API_KEY"),
    ) as client:
        voices = await client.list_voices(timeout_seconds=5.0)
        if not voices:
            raise RuntimeError("No TTS voices are available")
        voice = next((item for item in voices if item.get("profile")), voices[0])
        if profile := voice.get("profile"):
            print(
                "profile:",
                voice["label"],
                profile["speaker"]["dialect"],
                profile["languages"],
            )

        model = tts.TtsModel.Nebula
        pcm = await client.synthesize(
            "Hello from HUMAIN Voice",
            voice_id=voice["id"],
            model=model,
            # This is an inactivity timeout applied while awaiting each chunk.
            timeout_seconds=30.0,
            on_error=handle_error,
        )

    write_pcm16_wav(output_path, pcm, tts.get_sample_rate(model))


if __name__ == "__main__":
    asyncio.run(main())

from __future__ import annotations

import asyncio
import os
import random
import sys

from humain_voice import stt
from humain_voice.stt.batchtranscription import TranscriptionResponse


def required_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


async def get_result_with_retry(
    client: stt.BatchTranscribeClient,
    job_id: str,
    attempts: int = 5,
) -> TranscriptionResponse:
    for attempt in range(1, attempts + 1):
        try:
            # save_result prevents a terminal read from clearing the stored
            # result before a retry. It does not define a retention duration.
            return await client.get_result(
                job_id, stt.Language.ArEn, save_result=True
            )
        except stt.BatchTranscribeError as error:
            rate_limited = isinstance(error, stt.BatchTranscribeRateLimitError)
            retryable = rate_limited or error.retryable is True
            print(
                {
                    "status_code": error.status_code,
                    "code": error.code,
                    "capacity": error.capacity,
                }
            )
            if not retryable or attempt == attempts:
                raise

            server_delay = 0
            if isinstance(error, stt.BatchTranscribeRateLimitError):
                server_delay = error.retry_after or 0
            exponential_delay = 0.5 * 2 ** (attempt - 1)
            await asyncio.sleep(max(server_delay, exponential_delay) + random.random() * 0.25)

    raise RuntimeError("Retry loop exhausted")


async def main() -> None:
    if len(sys.argv) < 2:
        raise RuntimeError("Pass a batch job ID as the first argument")

    async with stt.BatchTranscribeClient(
        api_url=required_env("API_URL"),
        api_key=required_env("API_KEY"),
    ) as client:
        result = await get_result_with_retry(client, sys.argv[1])
        print(result.status.value, result.results.transcript if result.results else "")


if __name__ == "__main__":
    asyncio.run(main())