import gradio as gr import json import os import ssl import base64 import urllib.request import tempfile import requests import soundfile as sf from datetime import datetime from typing import Dict, List, Optional, Tuple import edge_tts from langdetect import detect from io import BytesIO # Custom CSS for better styling CUSTOM_CSS = """ /* General styling */ .container { max-width: 900px; margin: auto; } /* Header styling */ #header { text-align: center; padding: 20px; margin-bottom: 30px; } /* Component styling */ .input-section { background: var(--background-fill-primary); padding: 20px; border-radius: 10px; margin-bottom: 20px; } .output-section { background: var(--background-fill-primary); padding: 20px; border-radius: 10px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); } /* Note styling */ .note-card { background: var(--background-fill-primary); padding: 15px; margin: 10px 0; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.05); } /* Adaptive text colors for both themes */ .output-section .prose { color: var(--body-text-color) !important; } .output-section p, .output-section h1, .output-section h2, .output-section h3, .output-section h4, .output-section h5, .output-section h6 { color: var(--body-text-color) !important; } /* Ensure Markdown content uses theme colors */ .output-section .markdown-body { color: var(--body-text-color) !important; background-color: var(--background-fill-primary) !important; } /* Style blockquotes and code blocks */ .output-section blockquote { border-left: 3px solid var(--border-color-primary); color: var(--body-text-color); background: var(--background-fill-secondary); } .output-section code { color: var(--body-text-color); background: var(--background-fill-secondary); } """ # Create a light theme LIGHT_THEME = gr.themes.Default( primary_hue="blue", secondary_hue="blue", neutral_hue="slate", font=["Source Sans Pro", "ui-sans-serif", "system-ui", "sans-serif"], font_mono=["JetBrains Mono", "ui-monospace", "Consolas", "monospace"], ) MULTILINGUAL_SYSYTEM_PROMPTS = { 'english': { 'system': """You are a world class AI assistant. From users' collection of random thoughts, you need to organize into a clear, concise, and actionable format. Please structure them into the following categories that STRICTLY follows this format: 🧠 Main Idea/Theme – Summarize the central topic. ✅ Actionable Steps – Break down what can be done next. ❓ Open Questions – Highlight any uncertainties or areas that need further exploration. Here are thoughts:""", 'summarize': "Summarize the text above." }, 'chinese': { 'system': """你是一位世界级的AI助手。你需要将用户的随想整理成清晰、简洁和可行的格式。请严格按照以下格式将内容分类: 🧠 主要想法/主题 – 总结核心话题。 ✅ 可行步骤 – 分解下一步可以做什么。 ❓ 开放性问题 – 突出任何不确定性或需要进一步探索的领域。 以下是想法:""", 'summarize': "总结上述文本。" }, 'german': { 'system': """Sie sind ein erstklassiger KI-Assistent. Sie müssen die zufälligen Gedanken der Benutzer in ein klares, prägnantes und umsetzbares Format bringen. Bitte strukturieren Sie die Inhalte streng nach folgendem Format: 🧠 Hauptidee/Thema – Fassen Sie das zentrale Thema zusammen. ✅ Umsetzbare Schritte – Schlüsseln Sie auf, was als nächstes getan werden kann. ❓ Offene Fragen – Heben Sie Unklarheiten oder Bereiche hervor, die weiterer Erforschung bedürfen. Hier sind die Gedanken:""", 'summarize': "Fassen Sie den obigen Text zusammen." }, 'french': { 'system': """Vous êtes un assistant IA de classe mondiale. Vous devez organiser les pensées aléatoires des utilisateurs dans un format clair, concis et exploitable. Veuillez structurer le contenu en respectant strictement le format suivant: 🧠 Idée/Thème principal – Résumez le sujet central. ✅ Étapes exploitables – Décomposez ce qui peut être fait ensuite. ❓ Questions ouvertes – Mettez en évidence les incertitudes ou les domaines nécessitant une exploration plus approfondie. Voici les pensées :""", 'summarize': "Résumez le texte ci-dessus." }, 'italian': { 'system': """Sei un assistente IA di classe mondiale. Devi organizzare i pensieri casuali degli utenti in un formato chiaro, conciso e pratico. Si prega di strutturare il contenuto seguendo rigorosamente questo formato: 🧠 Idea/Tema principale – Riassumi l'argomento centrale. ✅ Passi attuabili – Scomponi cosa si può fare dopo. ❓ Domande aperte – Evidenzia eventuali incertezze o aree che necessitano di ulteriore esplorazione. Ecco i pensieri:""", 'summarize': "Riassumi il testo sopra." }, 'japanese': { 'system': """あなたは世界クラスのAIアシスタントです。ユーザーのランダムな考えを、明確で簡潔で実行可能な形式に整理する必要があります。以下の形式に厳密に従って内容を構造化してください: 🧠 メインアイデア/テーマ – 中心的なトピックを要約します。 ✅ 実行可能なステップ – 次に何ができるかを分解します。 ❓ オープンな質問 – 不確実性や更なる探求が必要な領域を強調します。 以下が考えです:""", 'summarize': "上記のテキストを要約してください。" }, 'spanish': { 'system': """Eres un asistente de IA de clase mundial. Necesitas organizar los pensamientos aleatorios de los usuarios en un formato claro, conciso y procesable. Por favor, estructure el contenido siguiendo estrictamente este formato: 🧠 Idea/Tema principal – Resume el tema central. ✅ Pasos procesables – Desglose lo que se puede hacer a continuación. ❓ Preguntas abiertas – Destaca cualquier incertidumbre o áreas que necesiten más exploración. Aquí están los pensamientos:""", 'summarize': "Resume el texto anterior." }, 'portuguese': { 'system': """Você é um assistente de IA de classe mundial. Você precisa organizar os pensamentos aleatórios dos usuários em um formato claro, conciso e acionável. Por favor, estruture o conteúdo seguindo rigorosamente este formato: 🧠 Ideia/Tema principal – Resuma o tópico central. ✅ Passos acionáveis – Detalhe o que pode ser feito a seguir. ❓ Questões em aberto – Destaque quaisquer incertezas ou áreas que precisam de mais exploração. Aqui estão os pensamentos:""", 'summarize': "Resuma o texto acima." } } def load_audio_from_url(url): """ Load audio from a URL using soundfile Args: url (str): URL of the audio file Returns: tuple: (audio_data, sample_rate) if successful, None otherwise """ try: # Get the audio file from the URL response = requests.get(url) response.raise_for_status() # Raise exception for bad status codes # For other formats that soundfile supports directly (WAV, FLAC, etc.) audio_data, sample_rate = sf.read(BytesIO(response.content)) return sample_rate, audio_data except Exception as e: print(f"Error loading audio from URL: {e}") return None class VoiceNotesApp: def __init__(self): # Azure endpoint configuration self.azure_url = os.getenv("AZURE_ENDPOINT") self.api_key = os.getenv("AZURE_API_KEY") # Initialize sample audio files - all using HTTPS URLs self.sample_audios = { "English - Weekend Plan": "https://diamondfan.github.io/audio_files/english.weekend.plan.wav", "Chinese - Kids & Work": "https://diamondfan.github.io/audio_files/chinese.kid.work.wav", "German - Vacation Planning": "https://diamondfan.github.io/audio_files/german.vacation.work.wav", "French - Random Thoughts": "https://diamondfan.github.io/audio_files/french.random.vacation.wav", "Italian - Daily Life": "https://diamondfan.github.io/audio_files/italian.daily.life.wav", "Japanese - Seattle Trip Report": "https://diamondfan.github.io/audio_files/japanese.seattle.trip.report.wav", "Spanish - Soccer Class": "https://diamondfan.github.io/audio_files/spanish.soccer.class.wav", "Portuguese - Buying House & Friends": "https://diamondfan.github.io/audio_files/portugese.house.friends.wav" } # Initialize storage self.notes_file = "voice_notes.json" self.notes = self.load_notes() def load_notes(self): if os.path.exists(self.notes_file): with open(self.notes_file, 'r') as f: notes = json.load(f) # Sort notes by timestamp in descending order (most recent first) return sorted(notes, key=lambda x: x['timestamp'], reverse=True) return [] def save_notes(self): with open(self.notes_file, 'w') as f: json.dump(self.notes, f, indent=2) def encode_base64_from_file(self, file_path): """Encode file content to base64 string and determine MIME type.""" file_extension = os.path.splitext(file_path)[1].lower() # Map file extensions to MIME types if file_extension in ['.jpg', '.jpeg']: mime_type = "image/jpeg" elif file_extension == '.png': mime_type = "image/png" elif file_extension == '.gif': mime_type = "image/gif" elif file_extension in ['.bmp', '.tiff', '.webp']: mime_type = f"image/{file_extension[1:]}" elif 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 = "application/octet-stream" # Read and encode file content with open(file_path, "rb") as file: encoded_string = base64.b64encode(file.read()).decode('utf-8') return encoded_string, mime_type def call_azure_endpoint(self, data): """Call Azure ML endpoint with the given data.""" parameters = {"temperature": 0.0} data["input_data"]["parameters"] = parameters def allowSelfSignedHttps(allowed): # bypass the server certificate verification on client side if allowed and not os.environ.get('PYTHONHTTPSVERIFY', '') and getattr(ssl, '_create_unverified_context', None): ssl._create_default_https_context = ssl._create_unverified_context allowSelfSignedHttps(True) body = str.encode(json.dumps(data)) if not self.api_key: raise Exception("A key should be provided to invoke the endpoint") headers = {'Content-Type': 'application/json', 'Authorization': ('Bearer ' + self.api_key)} req = urllib.request.Request(self.azure_url, body, headers) try: response = urllib.request.urlopen(req) result = response.read().decode('utf-8') return result except urllib.error.HTTPError as error: print("The request failed with status code: " + str(error.code)) print(error.info()) print(error.read().decode("utf8", 'ignore')) return f"Error: {error.code}" def transcribe_audio(self, audio_input): """Convert speech to text using Azure endpoint""" try: # Encode audio file to base64 encoded_audio, mime_type = self.encode_base64_from_file(audio_input) # Prepare the request payload speech_prompt = "Based on the attached audio, generate a comprehensive text transcription of the spoken content." payload = { "input_data": { "input_string": [ { "role": "user", "content": [ { "type": "text", "text": speech_prompt }, { "type": "audio_url", "audio_url": { "url": f"data:{mime_type};base64,{encoded_audio}" } } ] } ] } } # Call Azure endpoint response_json = self.call_azure_endpoint(payload) # Parse response try: response_data = json.loads(response_json) # Extract the actual response text if isinstance(response_data, dict) and "output" in response_data: transcription = response_data["output"] else: transcription = response_json except: transcription = response_json print(f"Debug transcription:\n{transcription}") return transcription except Exception as e: print(f"Transcription error: {str(e)}") return f"Error transcribing: {str(e)}" def summarize_text(self, text): """Generate a summary in the detected language using Azure endpoint""" if not text: return "No text to summarize" try: # First, detect language detected_language = detect(text) # Map detected language to supported languages language_mapping = { 'en': 'english', 'zh-cn': 'chinese', 'de': 'german', 'fr': 'french', 'it': 'italian', 'ja': 'japanese', 'es': 'spanish', 'pt': 'portuguese' } # Default to English if language not supported selected_language = language_mapping.get(detected_language, 'english') print(f"Detected language: {detected_language} ; Selected language: {selected_language} ; Text: {text}") # Generate summary in detected language prompts = MULTILINGUAL_SYSYTEM_PROMPTS[selected_language] # Prepare the request payload for summarization payload = { "input_data": { "input_string": [ { "role": "system", "content": [ { "type": "text", "text": prompts["system"] } ] }, { "role": "user", "content": [ { "type": "text", "text": text } ] }, { "role": "user", "content": [ { "type": "text", "text": prompts["summarize"] } ] } ] } } # Call Azure endpoint response_json = self.call_azure_endpoint(payload) # Parse response try: response_data = json.loads(response_json) # Extract the actual response text if isinstance(response_data, dict) and "output" in response_data: summary = response_data["output"] else: summary = response_json except: summary = response_json print(f"Debug summary:\n{summary}") return summary, selected_language except Exception as e: print(f"Summarization error: {str(e)}") return f"Error summarizing: {str(e)}", "english" def format_note_display(self, note): """Format note for display in a Gradio interface""" return f"""## 📝 Summary {note['summary']} --- ## 🎙️ Transcription {note['transcription']} --- ## 🕒 Timestamp **{note['timestamp']}** --- ## 🌐 Detected Language **{note['language']}**""" def search_notes(self, query): """Search through notes content""" if not query: return self.list_all_notes() matching_notes = [] query = query.lower() for note in self.notes: if (query in note['transcription'].lower() or query in note['summary'].lower()): matching_notes.append(note) if not matching_notes: return "No matching notes found" return "\n\n---\n\n".join([self.format_note_display(note) for note in matching_notes]) def list_all_notes(self): """Return all notes in formatted string""" if not self.notes: return "No notes found" # Notes are already sorted by timestamp in load_notes() return "\n\n---\n\n".join([self.format_note_display(note) for note in self.notes]) def process_note(self, audio, progress=gr.Progress()): """Process audio input and generate note with summary""" if audio is None: return "Please record or upload an audio file.", None, "❌ No audio provided" try: # Start processing progress(0, desc="Starting audio processing...") # Transcribe audio (25% of progress) progress(0.25, desc="Transcribing audio...") transcription = self.transcribe_audio(audio) # Generate summary (35% more progress) progress(0.60, desc="Generating summary...") summary, selected_language = self.summarize_text(transcription) # Create and save note (25% more progress) progress(0.85, desc="Saving note...") note = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "transcription": transcription, "summary": summary, "language": selected_language } # Save note self.notes.append(note) self.save_notes() # Complete progress(1.0, desc="Complete!") return self.format_note_display(note), audio, "✅ Processing complete!" except Exception as e: return str(e), None, f"❌ Error: {str(e)}" async def text_to_speech(self, text, detected_lang): """Convert text to speech using Edge TTS with language-specific voices""" if not text.strip(): return None # Map of language codes to Edge TTS voices voice_mapping = { 'english': 'en-US-RogerNeural', # English 'chinese': 'zh-CN-XiaoxiaoNeural', # Chinese 'german': 'de-DE-KatjaNeural', # German 'french': 'fr-FR-HenriNeural', # French 'italian': 'it-IT-DiegoNeural', # Italian 'japanese': 'ja-JP-KeitaNeural', # Japanese 'spanish': 'es-ES-XimenaNeural', # Spanish 'portuguese': 'pt-BR-AntonioNeural', # Portuguese } # Default to English if language not supported voice = voice_mapping.get(detected_lang, 'en-US-EricNeural') try: communicate = edge_tts.Communicate(text, voice) with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file: tmp_path = tmp_file.name await communicate.save(tmp_path) return tmp_path except Exception as e: print(f"TTS Error: {str(e)}") return None def create_interface(self): """Create an enhanced Gradio interface with improved styling and layout""" # Preload Chinese sample as default default_audio_path = self.sample_audios["English - Weekend Plan"] # Create the layout with tabs with gr.Blocks( css=CUSTOM_CSS, theme=LIGHT_THEME, title="Thoughts Organizer" ) as interface: # Header section with gr.Row(elem_id="header"): gr.Markdown( """ # 🎙️ Thoughts Organizer with Phi-4-Multimodal ### Transform your spoken thoughts into organized, actionable insights Capture, organize, and act on your spoken thoughts with AI-powered voice notes in multiple languages, including English, Chinese, German, French, Italian, Japanese, Spanish, and Portuguese. Other demos include [Phi-4-multimodal](https://huggingface.co/spaces/microsoft/phi-4-multimodal), [Phi-4-Mini](https://huggingface.co/spaces/microsoft/phi-4-mini) playgrounds, [Stories Come Alive](https://huggingface.co/spaces/microsoft/StoriesComeAlive), [Phine Speech Translator](https://huggingface.co/spaces/microsoft/PhineSpeechTranslator) """ ) # Main content tabs with gr.Tabs(): # Record New Note Tab with gr.Tab("📝 New Note", id="new_note"): with gr.Column(elem_classes="input-section"): # Audio input with clear instructions gr.Markdown("### Record or Upload Your Voice Note") audio_input = gr.Audio( sources=["microphone", "upload"], type="filepath", label="", interactive=True, show_download_button=True ) with gr.Row(): sample_audio = gr.Dropdown( choices=list(self.sample_audios.keys()), label="Select a sample audio to try", value="English - Weekend Plan" ) # Process button with loading state process_btn = gr.Button("🔄 Process Note", variant="primary") clear_btn = gr.Button("🗑️ Clear", variant="secondary") # Add progress bar progress_bar = gr.Progress() with gr.Column(elem_classes="output-section"): # Status message display processing_status = gr.Markdown(label="Status", value="Ready to process...") # Note output display note_display = gr.Markdown(label="") # Add TTS playback controls with gr.Row(): play_btn = gr.Button("🔊 Listen Note", variant="secondary") tts_audio = gr.Audio(label="TTS Output", visible=True, interactive=False) # Enhanced All Notes Tab with gr.Tab("📚 All Notes", id="all_notes"): with gr.Column(): # Search box search_box = gr.Textbox( label="🔍 Search Notes", placeholder="Enter keywords to search...", show_label=True ) # Refresh button for notes refresh_btn = gr.Button("🔄 Refresh Notes") # All notes display all_notes_display = gr.Markdown() # Function to load sample audio def load_sample(sample_name): if not sample_name: return None try: audio_url = self.sample_audios[sample_name] # Use the load_audio_from_url function to fetch the audio result = load_audio_from_url(audio_url) if result: sample_rate, audio_data = result return (sample_rate, audio_data) return None except Exception as e: print(f"Error loading sample audio: {e}") return None # Automatically load sample when selected sample_audio.change( fn=load_sample, inputs=[sample_audio], outputs=[audio_input], api_name="load_sample" ) def process_and_get_note(audio, progress=gr.Progress()): try: note_text, _, status = self.process_note(audio, progress) # Return both the note text and updated all notes display all_notes = self.list_all_notes() return note_text, all_notes, status except Exception as e: error_msg = f"❌ Error processing note: {str(e)}" return "", "", error_msg # Update the process button click event process_btn.click( fn=process_and_get_note, inputs=[audio_input], outputs=[note_display, all_notes_display, processing_status], api_name="process_note" ) # Function to handle TTS playback async def play_note(note_text): if not note_text: return None try: # Extract the detected language from the note display lang_section = note_text.split("Detected Language")[-1].strip() detected_lang = lang_section.strip('*').strip() # Extract the summary section (everything before the first ---) summary_section = note_text.split("---")[0].strip() # Remove Markdown headers (#) cleaned_text = summary_section.replace('#', '') # Remove emojis and section labels cleaned_text = cleaned_text.replace('📝 Summary', '').strip() # Remove all special characters except basic punctuation # Keep: letters, numbers, spaces, and basic punctuation clean_chars = [] for char in cleaned_text: if (char.isalnum() or char.isspace() or char in '.,!?-:;()[]{}"\''): clean_chars.append(char) cleaned_text = ''.join(clean_chars) # Remove multiple spaces cleaned_text = ' '.join(cleaned_text.split()) print(f"Debug - Original text: {summary_section}") print(f"Debug - Cleaned text: {cleaned_text}") audio_path = await self.text_to_speech(cleaned_text, detected_lang) return audio_path except Exception as e: print(f"Error in play_note: {str(e)}") return None # Update event handlers play_btn.click( fn=play_note, inputs=[note_display], outputs=[tts_audio], api_name="play_note" ) # Clear button functionality def clear_all(): # Return default/empty values for all components return None, "", "Ready to process...", "" clear_btn.click( fn=clear_all, inputs=[], outputs=[ audio_input, # Clear audio input note_display, # Clear note display processing_status, # Reset status message all_notes_display # Clear all notes display ] ) def refresh_notes(): # Reload notes from disk self.notes = self.load_notes() # Return updated notes display return self.list_all_notes() refresh_btn.click( fn=refresh_notes, inputs=[], outputs=[all_notes_display], api_name="refresh_notes" ) # Add search functionality search_box.change( fn=self.search_notes, inputs=[search_box], outputs=[all_notes_display], api_name="search_notes" ) # Instructions and tips with gr.Accordion("ℹ️ Tips & Instructions", open=False): gr.Markdown( """ ### How to Use: 1. **Record or Upload**: Use the microphone to record directly or upload an audio file 2. **Process**: Click 'Process Note' to convert your voice note into organized text 3. **Review**: View your processed note with main ideas, action items, and questions 4. **Listen**: Click the '🔊 Play Note' button to hear the summary read aloud 5. **Browse**: Switch to 'All Notes' tab to view your note history ### Features: - 🎙️ Record or upload voice notes - 📝 Automatic transcription - 🧠 Smart organization of ideas - 📚 Historical note tracking """ ) # Footer with gr.Column(elem_classes="output-section"): gr.Markdown("Powered by Microsoft [Phi-4 multimodal model](https://aka.ms/phi-4-multimodal/azure) on Azure AI. © 2025") return interface def run_app(): # Create app instance app = VoiceNotesApp() # Launch Gradio interface interface = app.create_interface() interface.launch( share=True, server_name="0.0.0.0", ) run_app()