Voice Changer

export VENICE_API_KEY="your-api-key"

curl "https://api.venice.ai/api/v1/models?type=music" \
  -H "Authorization: Bearer $VENICE_API_KEY" |
  jq '.data[] | select(.model_spec.voice_changer == true) | {
    id,
    voices: .model_spec.voices,
    default_voice: .model_spec.default_voice,
    supports_custom_voice_id: .model_spec.supports_custom_voice_id,
    accepted_audio_formats: .model_spec.accepted_audio_formats,
    max_source_audio_duration_seconds: .model_spec.max_source_audio_duration_seconds,
    supports_background_noise_removal: .model_spec.supports_background_noise_removal,
    supports_seed: .model_spec.supports_seed,
    durations: .model_spec.pricing.durations
  }'

curl https://api.venice.ai/api/v1/audio/voice-changer/quote \
  -H "Content-Type: application/json" \
  -d '{
    "model": "elevenlabs-voice-changer",
    "duration_seconds": 60
  }'

{
  "quote": 0.35,
  "duration_seconds": 60
}

{
  "duration_seconds": 7.008,
  "model": "elevenlabs-voice-changer",
  "queue_id": "01a164db-8652-7fa9-94d6-25754090bd64",
  "status": "QUEUED"
}

curl https://api.venice.ai/api/v1/audio/voice-changer/retrieve \
  -H "Authorization: Bearer $VENICE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "elevenlabs-voice-changer",
    "queue_id": "0190f2c4-9c1e-7a3b-8f42-2c9d5e7a1b34"
  }' \
  --output response.bin

{
  "status": "PROCESSING",
  "average_execution_time": 10000,
  "execution_duration": 4200
}

import os
import time
from pathlib import Path

import requests

BASE_URL = "https://api.venice.ai/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['VENICE_API_KEY']}"}
MODEL = "elevenlabs-voice-changer"
VOICE = "George"
# Length you expect to send, in whole seconds. The charge uses the length Venice measures.
EXPECTED_DURATION_SECONDS = 60

source = Path("source-recording.mp3")

# Voice names are not validated at queue time, so check before paying for a round trip.
models = requests.get(f"{BASE_URL}/models", params={"type": "music"}, headers=HEADERS)
models.raise_for_status()
spec = next(m["model_spec"] for m in models.json()["data"] if m["id"] == MODEL)
if VOICE not in spec["voices"] and not spec.get("supports_custom_voice_id"):
    raise SystemExit(f"{VOICE!r} is not a voice on {MODEL}: {spec['voices']}")

quote = requests.post(
    f"{BASE_URL}/audio/voice-changer/quote",
    json={"model": MODEL, "duration_seconds": EXPECTED_DURATION_SECONDS},
)
quote.raise_for_status()
print(f"Estimated cost: ${quote.json()['quote']:.2f}")

with source.open("rb") as audio:
    queued = requests.post(
        f"{BASE_URL}/audio/voice-changer/queue",
        headers=HEADERS,
        data={"model": MODEL, "voice": VOICE},
        files={"file": audio},
    )
queued.raise_for_status()
job = queued.json()
print(f"Queued {job['queue_id']} ({job['duration_seconds']:.1f}s billed)")

deadline = time.monotonic() + 12 * 60
while time.monotonic() < deadline:
    result = requests.post(
        f"{BASE_URL}/audio/voice-changer/retrieve",
        headers=HEADERS,
        json={
            "model": job["model"],
            "queue_id": job["queue_id"],
            "delete_media_on_completion": True,
        },
    )

    content_type = result.headers.get("Content-Type", "").split(";")[0].strip()
    if content_type == "audio/mpeg":
        output = Path("converted-audio.mp3")
        output.write_bytes(result.content)
        print(f"Saved {output} ({result.headers.get('x-venice-audio-duration')}s)")
        break

    body = result.json()
    if not result.ok:
        # Terminal failure. credits_refunded tells you whether the charge came back.
        print(f"{result.status_code}: {body.get('error')}")
        print(f"credits_refunded: {body.get('credits_refunded')}")
        raise SystemExit(1)

    if body.get("status") != "PROCESSING":
        print(body)
        raise SystemExit(1)

    # The average is a hint. On long inputs execution_duration overtakes it; fall back to a fixed interval.
    remaining_ms = body["average_execution_time"] - body["execution_duration"]
    time.sleep(5 if remaining_ms <= 0 else min(remaining_ms / 1000, 15))
else:
    raise SystemExit(
        f"Still processing after 12 minutes. Poll retrieve again with queue_id={job['queue_id']}."
    )