Spaces:
Running
Running
| import gradio as gr | |
| import json | |
| import requests | |
| import os | |
| import urllib.request | |
| import ssl | |
| import base64 | |
| import soundfile as sf | |
| from io import BytesIO | |
| import tempfile | |
| from datetime import datetime | |
| import logging | |
| # Set up logging | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| logger = logging.getLogger(__name__) | |
| class AzureSpeechTranslatorApp: | |
| def __init__(self): | |
| # Azure ML endpoint configuration | |
| self.url = os.getenv("AZURE_ENDPOINT") | |
| self.api_key = os.getenv("AZURE_API_KEY") | |
| # Define supported languages with their codes and native names | |
| self.languages = { | |
| "English": { | |
| "code": "en", | |
| "native": "English" | |
| }, | |
| "Chinese": { | |
| "code": "zh", | |
| "native": "中文" | |
| }, | |
| "German": { | |
| "code": "de", | |
| "native": "Deutsch" | |
| }, | |
| "French": { | |
| "code": "fr", | |
| "native": "Français" | |
| }, | |
| "Italian": { | |
| "code": "it", | |
| "native": "Italiano" | |
| }, | |
| "Japanese": { | |
| "code": "ja", | |
| "native": "日本語" | |
| }, | |
| "Spanish": { | |
| "code": "es", | |
| "native": "Español" | |
| }, | |
| "Portuguese": { | |
| "code": "pt", | |
| "native": "Português" | |
| } | |
| } | |
| # Initialize storage | |
| self.translations_dir = "translations" | |
| os.makedirs(self.translations_dir, exist_ok=True) | |
| self.translations = self.load_translations() | |
| def get_translation_file_path(self, lang_code): | |
| """Get path for language-specific translation file""" | |
| return os.path.join(self.translations_dir, f"translations_{lang_code}.json") | |
| def load_translations(self): | |
| """Load translations for all languages""" | |
| translations = {} | |
| for lang_info in self.languages.values(): | |
| file_path = self.get_translation_file_path(lang_info["code"]) | |
| if os.path.exists(file_path): | |
| with open(file_path, 'r', encoding='utf-8') as f: | |
| translations[lang_info["code"]] = json.load(f) | |
| else: | |
| translations[lang_info["code"]] = [] | |
| return translations | |
| def save_translation(self, lang_code, translation): | |
| """Save translation for specific language""" | |
| file_path = self.get_translation_file_path(lang_code) | |
| with open(file_path, 'w', encoding='utf-8') as f: | |
| json.dump(translation, f, ensure_ascii=False, indent=2) | |
| def call_azure_endpoint(self, payload): | |
| """Call Azure ML endpoint with the given payload.""" | |
| # Allow self-signed HTTPS certificates | |
| def allow_self_signed_https(allowed): | |
| if allowed and not os.environ.get('PYTHONHTTPSVERIFY', '') and getattr(ssl, '_create_unverified_context', None): | |
| ssl._create_default_https_context = ssl._create_unverified_context | |
| allow_self_signed_https(True) | |
| # Set parameters | |
| parameters = {"temperature": 0.7} | |
| if "parameters" not in payload["input_data"]: | |
| payload["input_data"]["parameters"] = parameters | |
| # Encode the request body | |
| body = str.encode(json.dumps(payload)) | |
| if not self.api_key: | |
| raise Exception("A key should be provided to invoke the endpoint") | |
| # Set up headers | |
| headers = {'Content-Type': 'application/json', 'Authorization': ('Bearer ' + self.api_key)} | |
| # Create and send the request | |
| req = urllib.request.Request(self.url, body, headers) | |
| try: | |
| logger.info(f"Sending request to {self.url}") | |
| response = urllib.request.urlopen(req) | |
| result = response.read().decode('utf-8') | |
| logger.info("Received response successfully") | |
| return json.loads(result) | |
| except urllib.error.HTTPError as error: | |
| logger.error(f"Request failed with status code: {error.code}") | |
| logger.error(f"Headers: {error.info()}") | |
| error_message = error.read().decode("utf8", 'ignore') | |
| logger.error(f"Error message: {error_message}") | |
| return {"error": error_message} | |
| def encode_audio_base64(self, audio_path): | |
| """Encode audio file to base64 and determine MIME type""" | |
| file_extension = os.path.splitext(audio_path)[1].lower() | |
| # Map file extensions to MIME types | |
| if file_extension == '.flac': | |
| mime_type = "audio/flac" | |
| elif file_extension == '.wav': | |
| mime_type = "audio/wav" | |
| elif file_extension == '.mp3': | |
| mime_type = "audio/mpeg" | |
| elif file_extension in ['.m4a', '.aac']: | |
| mime_type = "audio/aac" | |
| elif file_extension == '.ogg': | |
| mime_type = "audio/ogg" | |
| else: | |
| mime_type = "audio/wav" # Default to WAV | |
| # Read and encode file content | |
| with open(audio_path, "rb") as file: | |
| encoded_string = base64.b64encode(file.read()).decode('utf-8') | |
| return encoded_string, mime_type | |
| def transcribe_audio(self, audio_input, source_lang="English"): | |
| """Transcribe audio to text using Azure endpoint""" | |
| try: | |
| # Encode audio to base64 | |
| base64_audio, mime_type = self.encode_audio_base64(audio_input) | |
| # Create input content for Azure ML | |
| content_items = [ | |
| { | |
| "type": "text", | |
| "text": f"Transcribe this {source_lang} audio to text." | |
| }, | |
| { | |
| "type": "audio_url", | |
| "audio_url": { | |
| "url": f"data:{mime_type};base64,{base64_audio}" | |
| } | |
| } | |
| ] | |
| # Create conversation state for Azure ML | |
| conversation_state = [ | |
| { | |
| "role": "user", | |
| "content": content_items | |
| } | |
| ] | |
| # Create the payload | |
| payload = { | |
| "input_data": { | |
| "input_string": conversation_state | |
| } | |
| } | |
| # Call Azure ML endpoint | |
| response = self.call_azure_endpoint(payload) | |
| # Extract text response | |
| try: | |
| if isinstance(response, dict): | |
| if "result" in response: | |
| result = response["result"] | |
| elif "output" in response: | |
| if isinstance(response["output"], list) and len(response["output"]) > 0: | |
| result = response["output"][0] | |
| else: | |
| result = str(response["output"]) | |
| elif "error" in response: | |
| result = f"Error: {response['error']}" | |
| else: | |
| result = f"Unexpected response format: {json.dumps(response)}" | |
| else: | |
| result = str(response) | |
| except Exception as e: | |
| result = f"Error processing response: {str(e)}" | |
| return result.strip() | |
| except Exception as e: | |
| logger.error(f"Error in transcription: {str(e)}") | |
| return f"Transcription failed: {str(e)}" | |
| def translate_text(self, text, source_lang, target_lang): | |
| """Translate text between languages using Azure endpoint""" | |
| if not text: | |
| return "No text to translate" | |
| try: | |
| # Create input content for Azure ML | |
| content_items = [ | |
| { | |
| "type": "text", | |
| "text": f"Translate the following {source_lang} text to {target_lang}. Provide only the translation without any additional text or explanation:\n\n{text}" | |
| } | |
| ] | |
| # Create conversation state for Azure ML | |
| conversation_state = [ | |
| { | |
| "role": "system", | |
| "content": [{"type": "text", "text": "You are a professional translator."}] | |
| }, | |
| { | |
| "role": "user", | |
| "content": content_items | |
| } | |
| ] | |
| # Create the payload | |
| payload = { | |
| "input_data": { | |
| "input_string": conversation_state | |
| } | |
| } | |
| # Call Azure ML endpoint | |
| response = self.call_azure_endpoint(payload) | |
| # Extract text response | |
| try: | |
| if isinstance(response, dict): | |
| if "result" in response: | |
| result = response["result"] | |
| elif "output" in response: | |
| if isinstance(response["output"], list) and len(response["output"]) > 0: | |
| result = response["output"][0] | |
| else: | |
| result = str(response["output"]) | |
| elif "error" in response: | |
| result = f"Error: {response['error']}" | |
| else: | |
| result = f"Unexpected response format: {json.dumps(response)}" | |
| else: | |
| result = str(response) | |
| except Exception as e: | |
| result = f"Error processing response: {str(e)}" | |
| return result.strip() | |
| except Exception as e: | |
| logger.error(f"Error in translation: {str(e)}") | |
| return f"Translation failed: {str(e)}" | |
| def process_translation(self, audio, source_lang, target_lang): | |
| """Process audio input and generate translation""" | |
| if not audio: | |
| return "Please provide an audio file to translate." | |
| # Transcribe audio to text | |
| source_text = self.transcribe_audio(audio, source_lang) | |
| # Translate to target language | |
| translation = self.translate_text(source_text, source_lang, target_lang) | |
| # Create translation entry | |
| translation_entry = { | |
| "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| "source_language": source_lang, | |
| "target_language": target_lang, | |
| "source_text": source_text, | |
| "translated_text": translation | |
| } | |
| # Save translation | |
| source_code = self.languages[source_lang]["code"] | |
| target_code = self.languages[target_lang]["code"] | |
| if source_code not in self.translations: | |
| self.translations[source_code] = [] | |
| if target_code not in self.translations: | |
| self.translations[target_code] = [] | |
| self.translations[source_code].append(translation_entry) | |
| self.translations[target_code].append(translation_entry) | |
| self.save_translation(source_code, self.translations[source_code]) | |
| self.save_translation(target_code, self.translations[target_code]) | |
| return self.format_translation_display(translation_entry) | |
| def format_translation_display(self, entry): | |
| """Format translation for display""" | |
| output = f"""Timestamp: {entry['timestamp']}\n\n""" | |
| output += f"""Source Language ({entry['source_language']}):\n{entry['source_text']}\n\n""" | |
| output += f"""Target Language ({entry['target_language']}):\n{entry['translated_text']}\n""" | |
| return output | |
| def list_translations(self, lang_code): | |
| """List translations for specific language""" | |
| if lang_code not in self.translations or not self.translations[lang_code]: | |
| return "No translations found" | |
| return "\n\n---\n\n".join([ | |
| self.format_translation_display(entry) | |
| for entry in self.translations[lang_code] | |
| ]) | |
| def create_interface(self): | |
| """Create Gradio interface""" | |
| with gr.Blocks(theme=gr.themes.Soft()) as interface: | |
| gr.Markdown("# Phine Speech Translator with Phi-4-Multimodal") | |
| gr.Markdown("Record speech or upload audio file for translation between multiple languages using [Phi-4-Multimodal](https://aka.ms/phi-4-multimodal/azure). Other demos include [Phi-4-Mini playground](https://huggingface.co./spaces/microsoft/phi-4-mini), [Stories Come Alive](https://huggingface.co./spaces/microsoft/StoriesComeAlive), [Thoughts Organizer](https://huggingface.co./spaces/microsoft/ThoughtsOrganizer)") | |
| with gr.Row(): | |
| source_lang = gr.Dropdown( | |
| choices=list(self.languages.keys()), | |
| value="English", | |
| label="Source Language" | |
| ) | |
| target_lang = gr.Dropdown( | |
| choices=list(self.languages.keys()), | |
| value="Chinese", | |
| label="Target Language" | |
| ) | |
| with gr.Row(): | |
| audio_input = gr.Audio( | |
| sources=["microphone", "upload"], | |
| type="filepath", | |
| label="Record or Upload Audio" | |
| ) | |
| with gr.Row(): | |
| translate_btn = gr.Button("Translate") | |
| with gr.Row(): | |
| output = gr.Textbox( | |
| label="Translation Results", | |
| lines=10 | |
| ) | |
| # History viewer | |
| with gr.Accordion("Translation History", open=False): | |
| lang_select = gr.Dropdown( | |
| choices=list(self.languages.keys()), | |
| value="English", | |
| label="Select Language" | |
| ) | |
| history_output = gr.Textbox( | |
| label="Translation History", | |
| lines=20 | |
| ) | |
| # Event handlers | |
| translate_btn.click( | |
| fn=self.process_translation, | |
| inputs=[audio_input, source_lang, target_lang], | |
| outputs=output | |
| ) | |
| lang_select.change( | |
| fn=lambda x: self.list_translations(self.languages[x]["code"]), | |
| inputs=[lang_select], | |
| outputs=history_output | |
| ) | |
| return interface | |
| def run_app(): | |
| # Create app instance | |
| app = AzureSpeechTranslatorApp() | |
| # Launch Gradio interface | |
| interface = app.create_interface() | |
| interface.launch( | |
| share=True, | |
| server_name="0.0.0.0" | |
| ) | |
| if __name__ == "__main__": | |
| run_app() |