Spaces:
Running
Running
| 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 | |