Dubbing API

1
import os, time, mimetypes
2
from pathlib import Path
3
import httpx
4
from sarvamai import SarvamAI
5
from sarvamai.environment import SarvamAIEnvironment
6
7
8
def call(fn, attempts=5):
9
    for i in range(attempts):
10
        try:
11
            return fn()
12
        except httpx.TransportError:
13
            if i == attempts - 1:
14
                raise
15
            time.sleep(2**i)
16
17
18
client = SarvamAI(
19
    api_subscription_key=os.environ["SARVAM_API_KEY"],
20
    environment=SarvamAIEnvironment(
21
        base="https://api.sarvam.ai",
22
        creative="https://api.sarvam.ai/dubbing",
23
        production="wss://api.sarvam.ai",
24
    ),
25
)
26
27
media = Path("sample.wav")
28
source_language_code = "en-IN"
29
target_language_codes = ["hi-IN"]
30
export_options = ["audio"]
31
32
created = call(
33
    lambda: client.dubbing.create(
34
        source_language_code=source_language_code,
35
        target_language_codes=target_language_codes,
36
        export_options=export_options,
37
        voice_cloning=True,
38
        num_speakers=1,
39
        editor_flow=False,
40
        job_name=media.name,
41
    )
42
)
43
job_id = created.data.job_id
44
45
46
def upload():
47
    with media.open("rb") as fh:
48
        response = httpx.put(
49
            created.data.upload_url,
50
            content=fh,
51
            headers={
52
                "Content-Type": mimetypes.guess_type(media.name)[0] or "application/octet-stream",
53
                "x-ms-blob-type": "BlockBlob",
54
            },
55
            timeout=1800.0,
56
        )
57
    response.raise_for_status()
58
59
60
call(upload)
61
call(lambda: client.dubbing.start(job_id=job_id))
62
63
deadline = time.monotonic() + 3600
64
while time.monotonic() < deadline:
65
    job = call(lambda: client.dubbing.get_live_status(job_id=job_id)).data
66
    if job.status in {"completed", "failed", "partial_failure", "deleted"}:
67
        break
68
    time.sleep(10)
69
else:
70
    raise TimeoutError(job_id)
71
72
if job.status != "completed":
73
    raise RuntimeError(job.error_message or job.status)
74
75
wanted = {(t, f) for t in target_language_codes for f in export_options}
76
deadline = time.monotonic() + 3600
77
while time.monotonic() < deadline:
78
    exports = call(lambda: client.dubbing.get_export_status(job_id=job_id, limit=100)).data.exports or []
79
    settled = {(e.target_language, e.export_type) for e in exports if e.status in {"completed", "failed"}}
80
    if wanted <= settled:
81
        break
82
    time.sleep(10)
83
else:
84
    raise TimeoutError(job_id)
85
86
downloads = {
87
    (e.target_language, e.export_type): e.download_url
88
    for e in exports
89
    if e.status == "completed"
90
}