import requests
import json
import base64
class BhashiniClient:
"""
A client for interacting with Bhashini's ASR, NMT, and TTS services.
Methods:
list_available_languages(task_type): Lists available languages for a given task.
get_supported_voices(source_language): Gets supported genders for TTS in a language.
asr(audio_content, source_language, audio_format='wav', sampling_rate=16000): Performs ASR.
translate(text, source_language, target_language): Translates text from source to target language.
tts(text, source_language, gender='female', sampling_rate=8000): Performs TTS.
"""
PIPELINE_CONFIG_ENDPOINT = "https://meity-auth.ulcacontrib.org/ulca/apis/v0/model/getModelsPipeline"
INFERENCE_ENDPOINT = "https://dhruva-api.bhashini.gov.in/services/inference/pipeline"
PIPELINE_ID = "64392f96daac500b55c543cd"
def __init__(self, user_id, api_key, pipeline_id = PIPELINE_ID):
"""
Initializes the BhashiniClient with user credentials and pipeline ID.
Args:
user_id (str): Your user ID.
api_key (str): Your ULCA API key.
pipeline_id (str): The pipeline ID.
Raises:
Exception: If the pipeline configuration retrieval fails.
"""
self.user_id = user_id
self.api_key = api_key
self.pipeline_id = pipeline_id
self.headers = {
"Content-Type": "application/json",
"userID": self.user_id,
"ulcaApiKey": self.api_key
}
self.config = self._get_pipeline_config()
self.pipeline_data = self._parse_pipeline_config(self.config)
self.inference_api_key = self.pipeline_data['inferenceApiKey']
def _get_pipeline_config(self):
"""
Retrieves the pipeline configuration.
Returns:
dict: The pipeline configuration.
Raises:
Exception: If the request fails.
"""
payload = {
"pipelineTasks": [
{"taskType": "asr"},
{"taskType": "translation"},
{"taskType": "tts"}
],
"pipelineRequestConfig": {
"pipelineId": self.pipeline_id
}
}
response = requests.post(
self.PIPELINE_CONFIG_ENDPOINT,
headers=self.headers,
data=json.dumps(payload)
)
response.raise_for_status()
return response.json()
def _parse_pipeline_config(self, config):
"""
Parses the pipeline configuration and extracts necessary information.
Args:
config (dict): The pipeline configuration.
Returns:
dict: Parsed pipeline data.
"""
inference_api_key = config['pipelineInferenceAPIEndPoint']['inferenceApiKey']['value']
callback_url = config['pipelineInferenceAPIEndPoint']['callbackUrl']
pipeline_data = {
'asr': {},
'tts': {},
'translation': {},
'inferenceApiKey': inference_api_key,
'callbackUrl': callback_url
}
for pipeline in config['pipelineResponseConfig']:
task_type = pipeline['taskType']
if task_type in ['asr', 'translation', 'tts']:
for language_config in pipeline['config']:
source_language = language_config['language']['sourceLanguage']
if task_type != 'translation':
if source_language not in pipeline_data[task_type]:
pipeline_data[task_type][source_language] = []
language_info = {
'serviceId': language_config['serviceId'],
'sourceScriptCode': language_config['language'].get('sourceScriptCode')
}
if task_type == 'tts':
language_info['supportedVoices'] = language_config.get('supportedVoices', [])
pipeline_data[task_type][source_language].append(language_info)
else:
target_language = language_config['language']['targetLanguage']
if source_language not in pipeline_data[task_type]:
pipeline_data[task_type][source_language] = {}
if target_language not in pipeline_data[task_type][source_language]:
pipeline_data[task_type][source_language][target_language] = []
language_info = {
'serviceId': language_config['serviceId'],
'sourceScriptCode': language_config['language'].get('sourceScriptCode'),
'targetScriptCode': language_config['language'].get('targetScriptCode')
}
pipeline_data[task_type][source_language][target_language].append(language_info)
return pipeline_data
def list_available_languages(self, task_type):
"""
Lists the available languages for the specified task.
Args:
task_type (str): The task type ('asr', 'translation', or 'tts').
Returns:
list or dict: A list of available languages, or a dictionary for translation.
Raises:
ValueError: If an invalid task type is provided.
Usage Example:
client = BhashiniClient(user_id, api_key, pipeline_id)
asr_languages = client.list_available_languages('asr')
print("Available ASR Languages:", asr_languages)
translation_languages = client.list_available_languages('translation')
print("Available Translation Languages:", translation_languages)
"""
if task_type not in ['asr', 'translation', 'tts']:
raise ValueError("Invalid task type. Choose from 'asr', 'translation', or 'tts'.")
if task_type == 'translation':
languages = {}
for src_lang in self.pipeline_data['translation']:
languages[src_lang] = list(self.pipeline_data['translation'][src_lang].keys())
return languages
else:
return list(self.pipeline_data[task_type].keys())
def get_supported_voices(self, source_language):
"""
Returns the supported genders for TTS in the specified language.
Args:
source_language (str): The language code (e.g., 'hi' for Hindi).
Returns:
list: A list of supported genders (e.g., ['male', 'female']).
Raises:
ValueError: If TTS is not supported for the language.
Usage Example:
client = BhashiniClient(user_id, api_key, pipeline_id)
voices = client.get_supported_voices('hi')
print("Supported voices for Hindi TTS:", voices)
"""
if source_language not in self.pipeline_data['tts']:
available_languages = ', '.join(self.list_available_languages('tts'))
raise ValueError(
f"TTS not supported for language '{source_language}'. "
f"Available languages: {available_languages}"
)
service_info = self.pipeline_data['tts'][source_language][0]
supported_voices = service_info.get('supportedVoices', [])
return supported_voices
def asr(self, audio_content, source_language, audio_format='wav', sampling_rate=16000):
"""
Performs Automatic Speech Recognition on the provided audio content.
Args:
audio_content (bytes): The audio content in bytes.
source_language (str): The language code of the audio (e.g., 'hi' for Hindi).
audio_format (str): supported formats of audio content: ('wav', 'mp3', 'flac', 'ogg'.)
sampling_rate (int): The sampling rate of the audio in Hz.
Returns:
dict: The ASR response from the API.
Raises:
ValueError: If the language is not supported.
Exception: If the API request fails.
Usage Example:
client = BhashiniClient(user_id, api_key, pipeline_id)
with open('audio.wav', 'rb') as f:
audio_content = f.read()
asr_result = client.asr(audio_content, source_language='hi', audio_format='wav')
print("ASR Result:", asr_result)
"""
if source_language not in self.pipeline_data['asr']:
available_languages = ', '.join(self.list_available_languages('asr'))
raise ValueError(
f"ASR not supported for language '{source_language}'. "
f"Available languages: {available_languages}"
)
service_info = self.pipeline_data['asr'][source_language][0]
service_id = service_info['serviceId']
payload = {
"pipelineTasks": [
{
"taskType": "asr",
"config": {
"language": {
"sourceLanguage": source_language
},
"serviceId": service_id,
"audioFormat": audio_format,
"samplingRate": sampling_rate
}
}
],
"inputData": {
"audio": [
{
"audioContent": base64.b64encode(audio_content).decode('utf-8')
}
]
}
}
headers = {
'Accept': '*/*',
'Authorization': self.inference_api_key,
'Content-Type': 'application/json'
}
response = requests.post(
self.INFERENCE_ENDPOINT,
headers=headers,
data=json.dumps(payload)
)
self._handle_response_errors(response)
return response.json()
def translate(self, text, source_language, target_language):
"""
Translates the provided text from the source language to the target language.
Args:
text (str): The text to translate.
source_language (str): The source language code.
target_language (str): The target language code.
Returns:
dict: The translation response from the API.
Raises:
ValueError: If the language pair is not supported.
Exception: If the API request fails.
Usage Example:
client = BhashiniClient(user_id, api_key, pipeline_id)
translation_result = client.translate(
'मेरा नाम विहिर है।',
source_language='hi',
target_language='gu'
)
print("Translation Result:", translation_result)
"""
if source_language not in self.pipeline_data['translation']:
available_languages = ', '.join(self.list_available_languages('translation').keys())
raise ValueError(
f"Translation not supported from language '{source_language}'. "
f"Available source languages: {available_languages}"
)
if target_language not in self.pipeline_data['translation'][source_language]:
available_targets = ', '.join(self.pipeline_data['translation'][source_language].keys())
raise ValueError(
f"Translation from '{source_language}' to '{target_language}' not supported. "
f"Available target languages for '{source_language}': {available_targets}"
)
service_info = self.pipeline_data['translation'][source_language][target_language][0]
service_id = service_info['serviceId']
payload = {
"pipelineTasks": [
{
"taskType": "translation",
"config": {
"language": {
"sourceLanguage": source_language,
"targetLanguage": target_language
},
"serviceId": service_id
}
}
],
"inputData": {
"input": [
{
"source": text
}
]
}
}
headers = {
'Accept': '*/*',
'Authorization': self.inference_api_key,
'Content-Type': 'application/json'
}
response = requests.post(
self.INFERENCE_ENDPOINT,
headers=headers,
data=json.dumps(payload)
)
self._handle_response_errors(response)
return response.json()
def tts(self, text, source_language, gender='female', sampling_rate=8000):
"""
Converts the provided text to speech in the specified language.
Args:
text (str): The text to convert to speech.
source_language (str): The language code of the text.
gender (str): The desired voice gender ('male' or 'female').
sampling_rate (int): The sampling rate in Hz.
Returns:
dict: The TTS response from the API.
Raises:
ValueError: If the language or gender is not supported.
Exception: If the API request fails.
Usage Example:
client = BhashiniClient(user_id, api_key, pipeline_id)
tts_result = client.tts(
'હેલો વર્લ્ડ',
source_language='gu',
gender='female'
)
# Save the audio output
audio_base64 = tts_result['pipelineResponse'][0]['audio'][0]['audioContent']
audio_data = base64.b64decode(audio_base64)
with open('output_audio.wav', 'wb') as f:
f.write(audio_data)
"""
if source_language not in self.pipeline_data['tts']:
available_languages = ', '.join(self.list_available_languages('tts'))
raise ValueError(
f"TTS not supported for language '{source_language}'. "
f"Available languages: {available_languages}"
)
service_info = self.pipeline_data['tts'][source_language][0]
service_id = service_info['serviceId']
supported_voices = service_info.get('supportedVoices', [])
if gender not in ['male', 'female']:
raise ValueError("Gender must be 'male' or 'female'.")
if supported_voices and gender not in supported_voices:
available_genders = ', '.join(supported_voices)
raise ValueError(
f"Gender '{gender}' not supported for language '{source_language}'. "
f"Available genders: {available_genders}"
)
payload = {
"pipelineTasks": [
{
"taskType": "tts",
"config": {
"language": {
"sourceLanguage": source_language
},
"serviceId": service_id,
"gender": gender,
"samplingRate": sampling_rate
}
}
],
"inputData": {
"input": [
{
"source": text
}
]
}
}
headers = {
'Accept': '*/*',
'Authorization': self.inference_api_key,
'Content-Type': 'application/json'
}
response = requests.post(
self.INFERENCE_ENDPOINT,
headers=headers,
data=json.dumps(payload)
)
self._handle_response_errors(response)
return response.json()
def _handle_response_errors(self, response):
"""
Handles errors in the response.
Args:
response (requests.Response): The response object.
Raises:
Exception: If an HTTP error occurs.
"""
try:
response.raise_for_status()
except requests.HTTPError as http_err:
try:
error_info = response.json()
error_message = error_info.get('message', 'An error occurred.')
except json.JSONDecodeError:
error_message = response.text
raise Exception(f"HTTP error occurred: {error_message}") from http_err