audio_samples/

app.py

@@ -4,10 +4,10 @@ import numpy as np
 import os
 import hashlib
 from datetime import datetime
+from transformers import pipeline
 import soundfile as sf
 import torch
 from tenacity import retry, stop_after_attempt, wait_fixed
-from transformers import pipeline
 
 # Initialize local models with retry logic
 @retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
@@ -15,7 +15,7 @@ def load_whisper_model():
     try:
         model = pipeline(
             "automatic-speech-recognition",
-            model="openai/whisper-tiny",  # Multilingual model
+            model="openai/whisper-tiny.en",
             device=-1,  # CPU; use device=0 for GPU if available
             model_kwargs={"use_safetensors": True}
         )
@@ -65,7 +65,7 @@ try:
 except Exception as e:
     print(f"Symptom model initialization failed after retries: {str(e)}")
     symptom_classifier = None
-    is_fallback_model = True
+    is_fallback_model = True  # Track if fallback model is used
 
 def compute_file_hash(file_path):
     """Compute MD5 hash of a file to check uniqueness."""
@@ -75,7 +75,7 @@ def compute_file_hash(file_path):
             hash_md5.update(chunk)
     return hash_md5.hexdigest()
 
-def transcribe_audio(audio_file, language="en"):
+def transcribe_audio(audio_file):
     """Transcribe audio using local Whisper model."""
     if not whisper:
         return "Error: Whisper model not loaded. Check logs for details or ensure sufficient compute resources."
@@ -85,15 +85,15 @@ def transcribe_audio(audio_file, language="en"):
         if len(audio) < 1600:  # Less than 0.1s
             return "Error: Audio too short. Please provide audio of at least 1 second."
         if np.max(np.abs(audio)) < 1e-4:  # Too quiet
-            return "Error: Audio too quiet. Please provide clear audio describing symptoms."
+            return "Error: Audio too quiet. Please provide clear audio describing symptoms in English."
         
         # Save as WAV for Whisper
         temp_wav = f"/tmp/{os.path.basename(audio_file)}.wav"
         sf.write(temp_wav, audio, sr)
         
-        # Transcribe with beam search and language
+        # Transcribe with beam search
         with torch.no_grad():
-            result = whisper(temp_wav, generate_kwargs={"num_beams": 5, "language": language})
+            result = whisper(temp_wav, generate_kwargs={"num_beams": 5})
         transcription = result.get("text", "").strip()
         print(f"Transcription: {transcription}")
         
@@ -104,7 +104,7 @@ def transcribe_audio(audio_file, language="en"):
             pass
         
         if not transcription:
-            return "Transcription empty. Please provide clear audio describing symptoms."
+            return "Transcription empty. Please provide clear audio describing symptoms in English."
         # Check for repetitive transcription
         words = transcription.split()
         if len(words) > 5 and len(set(words)) < len(words) / 2:
@@ -134,20 +134,10 @@ def analyze_symptoms(text):
     except Exception as e:
         return f"Error analyzing symptoms: {str(e)}", 0.0
 
-def handle_health_query(query, language="en"):
-    """Handle health-related queries with a general response."""
-    if not query:
-        return "Please provide a valid health query."
-    # Placeholder for Q&A logic (could integrate a model like BERT for Q&A)
-    restricted_terms = ["medicine", "treatment", "drug", "prescription"]
-    if any(term in query.lower() for term in restricted_terms):
-        return "This tool does not provide medication or treatment advice. Please ask about symptoms or general health information (e.g., 'What are symptoms of asthma?')."
-    return f"Response to query '{query}': For accurate health information, consult a healthcare provider."
-
-def analyze_voice(audio_file, language="en"):
-    """Analyze voice for health indicators and handle queries."""
+def analyze_voice(audio_file):
+    """Analyze voice for health indicators."""
     try:
-        # Ensure unique file name
+        # Ensure unique file name to avoid Gradio reuse
         unique_path = f"/tmp/gradio/{datetime.now().strftime('%Y%m%d%H%M%S%f')}_{os.path.basename(audio_file)}"
         os.rename(audio_file, unique_path)
         audio_file = unique_path
@@ -161,43 +151,29 @@ def analyze_voice(audio_file, language="en"):
         print(f"Audio shape: {audio.shape}, Sampling rate: {sr}, Duration: {len(audio)/sr:.2f}s, Mean: {np.mean(audio):.4f}, Std: {np.std(audio):.4f}")
         
         # Transcribe audio
-        transcription = transcribe_audio(audio_file, language)
+        transcription = transcribe_audio(audio_file)
         if "Error transcribing" in transcription:
             return transcription
         
-        # Split transcription into symptom and query parts
-        symptom_text = transcription
-        query_text = None
-        restricted_terms = ["medicine", "treatment", "drug", "prescription"]
-        for term in restricted_terms:
-            if term in transcription.lower():
-                # Split at the first restricted term
-                split_index = transcription.lower().find(term)
-                symptom_text = transcription[:split_index].strip()
-                query_text = transcription[split_index:].strip()
-                break
+        # Check for medication-related queries
+        if "medicine" in transcription.lower() or "treatment" in transcription.lower():
+            feedback = "Error: This tool does not provide medication or treatment advice. Please describe symptoms only (e.g., 'I have a fever')."
+            feedback += f"\n\n**Debug Info**: Transcription = '{transcription}', File Hash = {file_hash}"
+            feedback += "\n**Disclaimer**: This is not a diagnostic tool. Consult a healthcare provider for medical advice."
+            return feedback
         
-        feedback = ""
+        # Analyze symptoms
+        prediction, score = analyze_symptoms(transcription)
+        if "Error analyzing" in prediction:
+            return prediction
         
-        # Analyze symptoms if present
-        if symptom_text:
-            prediction, score = analyze_symptoms(symptom_text)
-            if "Error analyzing" in prediction:
-                feedback += prediction + "\n"
-            elif prediction == "No health condition predicted":
-                feedback += "No significant health indicators detected.\n"
-            else:
-                feedback += f"Possible health condition: {prediction} (confidence: {score:.4f}). Consult a doctor.\n"
+        # Generate feedback
+        if prediction == "No health condition predicted":
+            feedback = "No significant health indicators detected."
         else:
-            feedback += "No symptoms detected in the audio.\n"
+            feedback = f"Possible health condition: {prediction} (confidence: {score:.4f}). Consult a doctor."
         
-        # Handle query if present
-        if query_text:
-            feedback += f"\nQuery detected: '{query_text}'\n"
-            feedback += handle_health_query(query_text, language) + "\n"
-        
-        # Add debug info and disclaimer
-        feedback += f"\n**Debug Info**: Transcription = '{transcription}', File Hash = {file_hash}"
+        feedback += f"\n\n**Debug Info**: Transcription = '{transcription}', Prediction = {prediction}, Confidence = {score:.4f}, File Hash = {file_hash}"
         feedback += "\n**Disclaimer**: This is not a diagnostic tool. Consult a healthcare provider for medical advice."
         
         # Clean up temporary audio file
@@ -211,48 +187,26 @@ def analyze_voice(audio_file, language="en"):
     except Exception as e:
         return f"Error processing audio: {str(e)}"
 
+def test_with_sample_audio():
+    """Test the app with sample audio files."""
+    samples = ["audio_samples/sample.wav", "audio_samples/common_voice_en.wav"]
+    results = []
+    for sample in samples:
+        if os.path.exists(sample):
+            results.append(analyze_voice(sample))
+        else:
+            results.append(f"Sample not found: {sample}")
+    return "\n".join(results)
+
 # Gradio interface
-def create_gradio_interface():
-    with gr.Blocks(theme=gr.themes.Soft()) as demo:
-        gr.Markdown(
-            """
-            # Health Voice Analyzer
-            Record or upload a voice sample describing symptoms in English, Spanish, Hindi, or Mandarin (e.g., 'I have a fever').  
-            Ask health questions in the text box below (e.g., 'What are symptoms of asthma?').  
-            **Note**: Do not ask for medication or treatment advice; focus on symptoms or general health questions.  
-            **Disclaimer**: This is not a diagnostic tool. Consult a healthcare provider for medical advice.  
-            **Text-to-Speech**: Available in the web frontend (Salesforce Sites) using the browser's Web Speech API.
-            """
-        )
-        with gr.Row():
-            language = gr.Dropdown(
-                choices=["en", "es", "hi", "zh"],
-                label="Select Language",
-                value="en"
-            )
-        with gr.Row():
-            audio_input = gr.Audio(type="filepath", label="Record or Upload Voice")
-        with gr.Row():
-            query_input = gr.Textbox(label="Ask a Health Question (e.g., 'What are symptoms of asthma?')")
-        with gr.Row():
-            output = gr.Textbox(label="Health Assessment Feedback")
-        with gr.Row():
-            analyze_button = gr.Button("Analyze Voice")
-            query_button = gr.Button("Submit Query")
-        
-        analyze_button.click(
-            fn=analyze_voice,
-            inputs=[audio_input, language],
-            outputs=output
-        )
-        query_button.click(
-            fn=handle_health_query,
-            inputs=[query_input, language],
-            outputs=output
-        )
-    
-    return demo
+iface = gr.Interface(
+    fn=analyze_voice,
+    inputs=gr.Audio(type="filepath", label="Record or Upload Voice"),
+    outputs=gr.Textbox(label="Health Assessment Feedback"),
+    title="Health Voice Analyzer",
+    description="Record or upload a voice sample describing symptoms (e.g., 'I have a fever') for preliminary health assessment. Supports English only. Use clear audio (WAV, 16kHz). Do not ask for medication or treatment advice."
+)
 
 if __name__ == "__main__":
-    demo = create_gradio_interface()
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    print(test_with_sample_audio())
+    iface.launch(server_name="0.0.0.0", server_port=7860)