Update app.py

app.py

@@ -7,7 +7,6 @@ from simple_salesforce import Salesforce
 import os
 from datetime import datetime
 import logging
-from scipy.io import wavfile
 import webrtcvad
 
 # Set up logging
@@ -85,11 +84,10 @@ def extract_health_features(audio, sr):
         logger.error(f"Feature extraction failed: {str(e)}")
         raise
 
-def transcribe_audio(audio_file):
+def transcribe_audio(audio):
     """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")
+        inputs = whisper_processor(audio, sampling_rate=16000, 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]
@@ -105,21 +103,24 @@ def analyze_symptoms(text):
     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 "stressed" in text or "stress" in text or "fatigue" in text:
+        feedback.append("Reported stress or fatigue may suggest conditions like anxiety 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):
+def analyze_voice(audio_file=None, audio_data=None):
     """Analyze voice for health indicators."""
     try:
-        # Validate input
-        if not os.path.exists(audio_file):
-            raise ValueError("Audio file not found")
-        if not audio_file.lower().endswith((".wav", ".mp3", ".flac")):
-            raise ValueError("Supported formats: WAV, MP3, FLAC")
-        audio, sr = librosa.load(audio_file, sr=16000)
+        # Use provided audio file or in-memory audio data
+        if audio_file and os.path.exists(audio_file):
+            audio, sr = librosa.load(audio_file, sr=16000)
+        elif audio_data is not None:
+            audio = audio_data
+            sr = 16000
+        else:
+            raise ValueError("No audio input provided")
+
         if len(audio) < sr:
             raise ValueError("Audio too short (minimum 1 second)")
 
@@ -127,7 +128,7 @@ def analyze_voice(audio_file):
         features = extract_health_features(audio, sr)
 
         # Transcribe audio for symptom analysis
-        transcription = transcribe_audio(audio_file)
+        transcription = transcribe_audio(audio)
         symptom_feedback = analyze_symptoms(transcription) if transcription else "No transcription available for symptom analysis."
 
         # Analyze voice features for health indicators
@@ -160,16 +161,9 @@ def analyze_voice(audio_file):
         feedback_str = "\n".join(feedback)
 
         # Store in Salesforce
-        if sf:
+        if sf and audio_file:
             store_in_salesforce(audio_file, feedback_str, respiratory_score, mental_health_score, features, transcription)
 
-        # Clean up
-        try:
-            os.remove(audio_file)
-            logger.info(f"Deleted audio: {audio_file}")
-        except Exception as e:
-            logger.error(f"Failed to delete audio: {str(e)}")
-
         return feedback_str
     except Exception as e:
         logger.error(f"Audio processing failed: {str(e)}")
@@ -183,7 +177,7 @@ def store_in_salesforce(audio_file, feedback, respiratory_score, mental_health_s
             "Feedback__c": feedback,
             "RespiratoryScore__c": float(respiratory_score),
             "MentalHealthScore__c": float(mental_health_score),
-            "AudioFileName__c": os.path.basename(audio_file),
+            "AudioFileName__c": os.path.basename(audio_file) if audio_file else "in_memory_audio",
             "Pitch__c": float(features["pitch"]),
             "Jitter__c": float(features["jitter"]),
             "Shimmer__c": float(features["shimmer"]),
@@ -195,38 +189,22 @@ def store_in_salesforce(audio_file, feedback, respiratory_score, mental_health_s
         logger.error(f"Salesforce storage failed: {str(e)}")
 
 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 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)
-        freq_mod = 150 + 20 * np.sin(2 * np.pi * 0.5 * t)  # Increased frequency variation
-        amplitude_mod = 0.5 + 0.2 * np.sin(2 * np.pi * 0.3 * t)  # Increased amplitude variation
-        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.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")
-        # Normalize to int16 for scipy.io.wavfile
-        dummy_audio = (dummy_audio * 32767).astype(np.int16)
-        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:
-            # Write audio using scipy.io.wavfile
-            wavfile.write(sample_audio_path, sr, dummy_audio)
-            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
-    return analyze_voice(sample_audio_path)
+    """Test with dummy audio simulating a user's voice saying 'I have a cough and feel stressed'."""
+    logger.info("Starting test with in-memory audio simulation")
+    # Generate synthetic audio: 150 Hz base frequency with variations to mimic a stressed voice with cough
+    sr = 16000
+    t = np.linspace(0, 2, 2 * sr)
+    freq_mod = 150 + 25 * np.sin(2 * np.pi * 0.5 * t)  # Increased jitter for respiratory hint
+    amplitude_mod = 0.5 + 0.25 * np.sin(2 * np.pi * 0.3 * t)  # Increased shimmer for stress hint
+    noise = 0.05 * np.random.normal(0, 1, len(t))  # Moderate noise
+    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.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)}, dtype: {dummy_audio.dtype}")
+    return analyze_voice(audio_data=dummy_audio)
 
 # Gradio interface
 iface = gr.Interface(
@@ -238,6 +216,6 @@ iface = gr.Interface(
 )
 
 if __name__ == "__main__":
-    logger.info("Starting Voice Health Analyzer")
+    logger.info("Starting Voice Health Analyzer at 10:31 AM IST, June 23, 2025")
     print(test_with_sample_audio())
     iface.launch(server_name="0.0.0.0", server_port=7860)