File size: 1,913 Bytes
51419fe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | import os
import requests
from typing import Optional
from . import APIProvider, PermanentError, register
@register("smallestai")
class SmallestAIProvider(APIProvider):
def transcribe(
self,
model_variant: str,
audio_file_path: Optional[str],
sample: dict,
use_url: bool = False,
language: str = "en",
) -> str:
api_key = os.getenv("SMALLESTAI_API_KEY")
if not api_key or api_key == "your_api_key":
raise ValueError(
"SMALLESTAI_API_KEY environment variable not set, get your key at https://console.smallest.ai"
)
endpoint = "https://api.smallest.ai/waves/v1/pulse/get_text"
if use_url:
audio_url = sample["row"]["audio"][0]["src"]
response = requests.post(
endpoint,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json={"url": audio_url},
params={"language": language},
timeout=300,
)
else:
with open(audio_file_path, "rb") as audio_file:
audio_data = audio_file.read()
response = requests.post(
endpoint,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/octet-stream",
},
data=audio_data,
params={"language": language},
timeout=300,
)
if response.status_code != 429 and 400 <= response.status_code < 500:
raise PermanentError(
f"Smallest AI API returned {response.status_code}: {response.text}"
)
response.raise_for_status()
return response.json().get("transcription", "") or "."
|