Update app.py

app.py

@@ -1,8 +1,9 @@
+```python
 import gradio as gr
 import librosa
 import numpy as np
 import torch
-from transformers import Wav2Vec2Processor, Wav2Vec2Model
+from transformers import WhisperProcessor, WhisperForConditionalGeneration, pipeline
 from simple_salesforce import Salesforce
 import os
 from datetime import datetime
@@ -36,9 +37,10 @@ try:
 except Exception as e:
     logger.error(f"Salesforce connection failed: {str(e)}")
 
-# Load Wav2Vec2 model (optional context features)
-processor = Wav2Vec2Processor.from_pretrained("facebook/wav2vec2-base-960h")
-model = Wav2Vec2Model.from_pretrained("facebook/wav2vec2-base-960h")
+# Load Whisper model for speech-to-text
+whisper_processor = WhisperProcessor.from_pretrained("openai/whisper-tiny")
+whisper_model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-tiny")
+whisper_model.config.forced_decoder_ids = whisper_processor.get_decoder_prompt_ids(language="english", task="transcribe")
 
 # Initialize VAD
 vad = webrtcvad.Vad(mode=2)  # Moderate mode for balanced voice detection
@@ -84,6 +86,33 @@ def extract_health_features(audio, sr):
         logger.error(f"Feature extraction failed: {str(e)}")
         raise
 
+def transcribe_audio(audio_file):
+    """Transcribe audio to text using Whisper."""
+    try:
+        audio, sr = librosa.load(audio_file, sr=16000)
+        inputs = whisper_processor(audio, sampling_rate=sr, return_tensors="pt")
+        with torch.no_grad():
+            generated_ids = whisper_model.generate(inputs["input_features"])
+        transcription = whisper_processor.batch_decode(generated_ids, skip_special_tokens=True)[0]
+        logger.info(f"Transcription: {transcription}")
+        return transcription
+    except Exception as e:
+        logger.error(f"Transcription failed: {str(e)}")
+        return ""
+
+def analyze_symptoms(text):
+    """Mock symptom-to-disease analysis (placeholder for symptom-2-disease-net)."""
+    # Since abhirajeshbhai/symptom-2-disease-net is not locally available, use rule-based analysis
+    text = text.lower()
+    feedback = []
+    if "cough" in text or "difficulty breathing" in text:
+        feedback.append("Symptoms like cough or difficulty breathing may indicate a respiratory condition, such as bronchitis or asthma. Consult a doctor.")
+    if "tired" in text or "fatigue" in text:
+        feedback.append("Reported fatigue may suggest conditions like anemia or chronic fatigue syndrome. Seek medical advice.")
+    if not feedback:
+        feedback.append("No specific conditions detected from reported symptoms.")
+    return "\n".join(feedback)
+
 def analyze_voice(audio_file):
     """Analyze voice for health indicators."""
     try:
@@ -96,10 +125,14 @@ def analyze_voice(audio_file):
         if len(audio) < sr:
             raise ValueError("Audio too short (minimum 1 second)")
 
-        # Extract features
+        # Extract voice features
         features = extract_health_features(audio, sr)
 
-        # Analyze for health indicators
+        # Transcribe audio for symptom analysis
+        transcription = transcribe_audio(audio_file)
+        symptom_feedback = analyze_symptoms(transcription) if transcription else "No transcription available for symptom analysis."
+
+        # Analyze voice features for health indicators
         feedback = []
         respiratory_score = features["jitter"]
         mental_health_score = features["shimmer"]
@@ -113,21 +146,24 @@ def analyze_voice(audio_file):
             feedback.append(f"Low vocal energy ({features['energy']:.4f}) may indicate fatigue or reduced vocal effort, potentially linked to physical or mental exhaustion.")
 
         if not feedback:
-            feedback.append("No significant health indicators detected. Your voice appears stable based on the analyzed features.")
+            feedback.append("No significant health indicators detected from voice features.")
 
-        # Debug info
-        feedback.append("\n**Analysis Details**:")
+        # Combine voice and symptom feedback
+        feedback.append("\n**Symptom Analysis (from transcription)**:")
+        feedback.append(symptom_feedback)
+        feedback.append("\n**Voice Analysis Details**:")
         feedback.append(f"Pitch: {features['pitch']:.2f} Hz (average fundamental frequency)")
         feedback.append(f"Jitter: {respiratory_score:.2f}% (pitch variation, higher values may indicate respiratory issues)")
         feedback.append(f"Shimmer: {mental_health_score:.2f}% (amplitude variation, higher values may indicate stress)")
         feedback.append(f"Energy: {features['energy']:.4f} (vocal intensity, lower values may indicate fatigue)")
+        feedback.append(f"Transcription: {transcription if transcription else 'None'}")
         feedback.append("\n**Disclaimer**: This is a preliminary analysis, not a medical diagnosis. Always consult a healthcare provider for professional evaluation.")
 
         feedback_str = "\n".join(feedback)
 
         # Store in Salesforce
         if sf:
-            store_in_salesforce(audio_file, feedback_str, respiratory_score, mental_health_score, features)
+            store_in_salesforce(audio_file, feedback_str, respiratory_score, mental_health_score, features, transcription)
 
         # Clean up
         try:
@@ -141,7 +177,7 @@ def analyze_voice(audio_file):
         logger.error(f"Audio processing failed: {str(e)}")
         return f"Error: {str(e)}"
 
-def store_in_salesforce(audio_file, feedback, respiratory_score, mental_health_score, features):
+def store_in_salesforce(audio_file, feedback, respiratory_score, mental_health_score, features, transcription):
     """Store results in Salesforce."""
     try:
         sf.HealthAssessment__c.create({
@@ -153,7 +189,8 @@ def store_in_salesforce(audio_file, feedback, respiratory_score, mental_health_s
             "Pitch__c": float(features["pitch"]),
             "Jitter__c": float(features["jitter"]),
             "Shimmer__c": float(features["shimmer"]),
-            "Energy__c": float(features["energy"])
+            "Energy__c": float(features["energy"]),
+            "Transcription__c": transcription
         })
         logger.info("Stored in Salesforce")
     except Exception as e:
@@ -163,7 +200,7 @@ def test_with_sample_audio():
     """Test with sample or dummy audio simulating a user's voice."""
     sample_audio_path = "audio_samples/sample.wav"
     if not os.path.exists(sample_audio_path):
-        logger.warning("Sample audio not found; generating dummy audio to simulate user voice")
+        logger.warning("Sample audio not found; generating dummy audio to simulate user voice saying 'I have a cough'")
         # Generate synthetic audio: 150 Hz base frequency to mimic human voice
         sr = 16000
         t = np.linspace(0, 2, 2 * sr)
@@ -172,16 +209,20 @@ def test_with_sample_audio():
         noise = 0.05 * np.random.normal(0, 1, len(t))  # Moderate noise for realism
         dummy_audio = amplitude_mod * np.sin(2 * np.pi * freq_mod * t) + noise
         # Ensure dummy_audio is a 1D NumPy array
-        dummy_audio = np.asarray(dummy_audio, dtype=np.float64).flatten()
+        dummy_audio = np.asarray(dummy_audio, dtype=np.float32).flatten()
         if not isinstance(dummy_audio, np.ndarray) or dummy_audio.ndim != 1:
             logger.error(f"Invalid dummy_audio: type={type(dummy_audio)}, shape={dummy_audio.shape if hasattr(dummy_audio, 'shape') else 'N/A'}")
             raise ValueError("Generated audio is not a 1D NumPy array")
-        logger.info(f"Dummy audio shape: {dummy_audio.shape}, type: {type(dummy_audio)}")
+        logger.info(f"Dummy audio shape: {dummy_audio.shape}, type: {type(dummy_audio)}, dtype: {dummy_audio.dtype}")
         sample_audio_path = "audio_samples/dummy_test.wav"
         os.makedirs("audio_samples", exist_ok=True)
         try:
+            # Test audio writing
             soundfile.write(dummy_audio, sr, sample_audio_path)
             logger.info(f"Generated dummy audio at: {sample_audio_path}")
+            # Verify file exists
+            if not os.path.exists(sample_audio_path):
+                raise ValueError(f"Audio file not created: {sample_audio_path}")
         except Exception as e:
             logger.error(f"Failed to write dummy audio: {str(e)}")
             raise
@@ -200,3 +241,4 @@ if __name__ == "__main__":
     logger.info("Starting Voice Health Analyzer")
     print(test_with_sample_audio())
     iface.launch(server_name="0.0.0.0", server_port=7860)
+```