Synthèse vocale (WebSocket)

import asyncio
import gradium

async def main():
    client = gradium.client.GradiumClient(api_key="your-api-key")
    audio_chunks = []

    async with client.tts_realtime(
        voice_id="YTpq7expH9539ERJ",
        output_format="pcm",
    ) as tts:
        async def sender():
            await tts.send_text("Hello, world.")
            await tts.send_eos()

        async def receiver():
            async for msg in tts:
                if msg["type"] == "audio":
                    audio_chunks.append(msg["audio"])
                elif msg["type"] == "end_of_stream":
                    return

        await asyncio.gather(sender(), receiver())

    with open("output.pcm", "wb") as f:
        f.write(b"".join(audio_chunks))

asyncio.run(main())

import gradium

client = gradium.client.GradiumClient()
result = await client.tts(
    setup={
        "model_name": "default",
        "voice_id": "YTpq7expH9539ERJ",
        "output_format": "wav"
    },
    text="Hello, world!"
)

with open("output.wav", "wb") as f:
    f.write(result.raw_data)

print(f"Sample rate: {result.sample_rate}")
print(f"Request ID: {result.request_id}")

stream = await client.tts_stream(
    setup={
        "model_name": "default",
        "voice_id": "LFZvm12tW_z0xfGo",
        "output_format": "pcm"
    },
    text="This is a longer text that will be streamed."
)

async for audio_chunk in stream.iter_bytes():
    print(f"Received {len(audio_chunk)} bytes")

result = await client.tts(
    setup={
        "model_name": "default",
        "voice_id": "YTpq7expH9539ERJ",
        "output_format": "wav"
    },
    text="Hello with my custom voice!"
)

# WAV format
result = await client.tts(setup={"voice_id": "YTpq7expH9539ERJ", "output_format": "wav"}, text="Hello")

# PCM format: the data is sampled at 48kHz, 16-bit signed integer, mono
result = await client.tts(setup={"voice_id": "YTpq7expH9539ERJ", "output_format": "pcm"}, text="Hello")

# Get numpy array from PCM
pcm_array = result.pcm()
pcm16_array = result.pcm16()

sample_text = "Hello, this is a test from the Gradium Text to Speech system. <flush> We are testing the flush."
test_audio = await client.tts(
    setup={'voice_id': 'YTpq7expH9539ERJ', 'output_format': 'wav'},
    text=sample_text,
)

sample_text = 'Hello, this is a test from the Gradium Text to Speech system. <break time="1.5s" /> We are testing the pause.'

test_audio = await client.tts(
    setup={'voice_id': 'YTpq7expH9539ERJ', 'output_format': 'wav'},
    text=sample_text,
)

result = await client.tts(
    setup={"voice_id": "YTpq7expH9539ERJ", "output_format": "wav"},
    text="Hello, world!"
)

for item in result.text_with_timestamps:
    print(f"{item.text}: {item.start_s:.2f}s - {item.stop_s:.2f}s")

async def text_generator():
    yield "Hello, "
    yield "this is "
    yield "a streaming "
    yield "example."

stream = await client.tts_stream(
    setup={"voice_id": "YTpq7expH9539ERJ", "output_format": "pcm"},
    text=text_generator()
)

async for chunk in stream.iter_bytes():
    pass

import gradium

client = gradium.client.GradiumClient()

result = await client.tts(
    setup={
        "voice_id": "YTpq7expH9539ERJ",
        "output_format": "wav",
        "pronunciation_id": "bb1ckYhNHCcIJjdK",  # Whatever your ID is.
    },
    text="The text you want to generate."
)

with open("output.wav", "wb") as f:
    f.write(result.raw_data)

wscat -c "wss://api.gradium.ai/api/speech/tts" \
  -H "x-api-key: your_api_key"
# After connection, type:
# {"type":"setup","voice_id":"YTpq7expH9539ERJ","model_name":"default","output_format":"wav"}
# {"type":"text","text":"Hello, world!"}
# {"type":"end_of_stream"}

import asyncio
import base64
import json

import websockets


async def synthesise(api_key: str, voice_id: str, text: str) -> bytes:
    setup = {
        "type": "setup",
        "voice_id": voice_id,
        "model_name": "default",
        "output_format": "wav",
    }
    audio_chunks = []

    async with websockets.connect(
        "wss://api.gradium.ai/api/speech/tts",
        additional_headers={"x-api-key": api_key},
    ) as ws:
        await ws.send(json.dumps(setup))
        ready = json.loads(await ws.recv())
        assert ready["type"] == "ready"

        await ws.send(json.dumps({"type": "text", "text": text}))
        await ws.send(json.dumps({"type": "end_of_stream"}))

        while True:
            msg = json.loads(await ws.recv())
            if msg["type"] == "audio":
                audio_chunks.append(base64.b64decode(msg["audio"]))
            elif msg["type"] == "end_of_stream":
                break
            elif msg["type"] == "error":
                raise RuntimeError(msg["message"])

    return b"".join(audio_chunks)


audio = asyncio.run(synthesise("your_api_key", "YTpq7expH9539ERJ", "Hello, world!"))
with open("output.wav", "wb") as f:
    f.write(audio)