Update app.py

app.py

@@ -9,9 +9,14 @@ import soundfile
 import torch
 from tenacity import retry, stop_after_attempt, wait_fixed
 import logging
+import tempfile
 
 # Set up logging
-logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    handlers=[logging.FileHandler("voice_analyzer.log"), logging.StreamHandler()]
+)
 logger = logging.getLogger(__name__)
 
 # Initialize local models with retry logic
@@ -24,7 +29,7 @@ def load_whisper_model():
             device=-1,  # CPU; use device=0 for GPU if available
             model_kwargs={"use_safetensors": True}
         )
-        logger.info("Whisper model loaded successfully.")
+        logger.info("Whisper model loaded successfully")
         return model
     except Exception as e:
         logger.error(f"Failed to load Whisper model: {str(e)}")
@@ -39,18 +44,17 @@ def load_symptom_model():
             device=-1,  # CPU
             model_kwargs={"use_safetensors": True}
         )
-        logger.info("Symptom-2-Disease model loaded successfully.")
+        logger.info("Symptom-2-Disease model loaded successfully")
         return model
     except Exception as e:
         logger.error(f"Failed to load Symptom-2-Disease model: {str(e)}")
-        # Fallback to a generic model
         try:
             model = pipeline(
                 "text-classification",
                 model="distilbert-base-uncased",
                 device=-1
             )
-            logger.warning("Fallback to distilbert-base-uncased model.")
+            logger.warning("Fallback to distilbert-base-uncased model")
             return model
         except Exception as fallback_e:
             logger.error(f"Fallback model failed: {str(fallback_e)}")
@@ -63,131 +67,147 @@ is_fallback_model = False
 try:
     whisper = load_whisper_model()
 except Exception as e:
-    logger.error(f"Whisper model initialization failed after retries: {str(e)}")
+    logger.error(f"Whisper model initialization failed: {str(e)}")
 
 try:
     symptom_classifier = load_symptom_model()
 except Exception as e:
-    logger.error(f"Symptom model initialization failed after retries: {str(e)}")
+    logger.error(f"Symptom model initialization failed: {str(e)}")
     symptom_classifier = None
     is_fallback_model = True
 
 def compute_file_hash(file_path):
-    """Compute MD5 hash of a file to check uniqueness."""
-    hash_md5 = hashlib.md5()
-    with open(file_path, "rb") as f:
-        for chunk in iter(lambda: f.read(4096), b""):
-            hash_md5.update(chunk)
-    return hash_md5.hexdigest()
+    """Compute MD5 hash of a file."""
+    try:
+        hash_md5 = hashlib.md5()
+        with open(file_path, "rb") as f:
+            for chunk in iter(lambda: f.read(4096), b""):
+                hash_md5.update(chunk)
+        return hash_md5.hexdigest()
+    except Exception as e:
+        logger.error(f"Failed to compute file hash: {str(e)}")
+        return "unknown"
 
 def transcribe_audio(audio_file):
-    """Transcribe audio using local Whisper model."""
+    """Transcribe audio using Whisper model."""
     if not whisper:
-        return "Error: Whisper model not loaded. Check logs for details."
+        logger.error("Whisper model not loaded")
+        return "Error: Whisper model not loaded"
     try:
-        # Load and validate audio
+        logger.debug(f"Transcribing audio: {audio_file}")
         audio, sr = librosa.load(audio_file, sr=16000)
-        if len(audio) < 1600:  # Less than 0.1s
-            return "Error: Audio too short. Provide at least 1 second."
-        if np.max(np.abs(audio)) < 1e-4:  # Too quiet
-            return "Error: Audio too quiet. Provide clear audio describing symptoms."
+        if len(audio) < 1600:
+            logger.error("Audio too short")
+            return "Error: Audio too short (<0.1s)"
+        if np.max(np.abs(audio)) < 1e-4:
+            logger.error("Audio too quiet")
+            return "Error: Audio too quiet"
         
-        # Save as WAV for Whisper
-        temp_wav = f"/tmp/{datetime.now().strftime('%Y%m%d%H%M%S%f')}_temp.wav"
-        soundfile.write(audio, sr, temp_wav)
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_wav:
+            temp_path = temp_wav.name
+            soundfile.write(audio, sr, temp_path)
+            logger.debug(f"Saved temp WAV: {temp_path}")
         
-        # Transcribe with beam search
         with torch.no_grad():
-            result = whisper(temp_wav, generate_kwargs={"num_beams": 5})
+            result = whisper(temp_path, generate_kwargs={"num_beams": 5})
         transcription = result.get("text", "").strip()
         logger.info(f"Transcription: {transcription}")
         
-        # Clean up temp file
         try:
-            os.remove(temp_wav)
-            logger.info(f"Deleted temp file: {temp_wav}")
+            os.remove(temp_path)
+            logger.debug(f"Deleted temp WAV: {temp_path}")
         except Exception as e:
-            logger.error(f"Failed to delete temp file: {str(e)}")
+            logger.error(f"Failed to delete temp WAV: {str(e)}")
         
         if not transcription:
-            return "Error: Transcription empty. Provide clear audio describing symptoms."
-        # Check for repetitive transcription
+            logger.error("Transcription empty")
+            return "Error: Transcription empty"
         words = transcription.split()
         if len(words) > 5 and len(set(words)) < len(words) / 2:
-            return "Error: Transcription repetitive. Provide clear, non-repetitive audio."
+            logger.error("Transcription repetitive")
+            return "Error: Transcription repetitive"
         return transcription
     except Exception as e:
-        logger.error(f"Error transcribing audio: {str(e)}")
+        logger.error(f"Transcription failed: {str(e)}")
         return f"Error: {str(e)}"
 
 def analyze_symptoms(text):
-    """Analyze symptoms using local Symptom-2-Disease model."""
+    """Analyze symptoms using Symptom-2-Disease model."""
     if not symptom_classifier:
-        return "Error: Symptom-2-Disease model not loaded.", 0.0
+        logger.error("Symptom-2-Disease model not loaded")
+        return "Error: Symptom-2-Disease model not loaded", 0.0
     try:
         if not text or "Error" in text:
-            return "Error: No valid transcription for analysis.", 0.0
+            logger.error(f"Invalid transcription: {text}")
+            return "Error: No valid transcription", 0.0
         with torch.no_grad():
             result = symptom_classifier(text)
         if result and isinstance(result, list) and len(result) > 0:
             prediction = result[0]["label"]
             score = result[0]["score"]
             if is_fallback_model:
-                logger.warning("Using fallback model; results may be less accurate.")
-                prediction = f"{prediction} (fallback model)"
+                logger.warning("Using fallback model")
+                prediction = f"{prediction} (fallback)"
             logger.info(f"Prediction: {prediction}, Score: {score:.4f}")
             return prediction, score
+        logger.warning("No prediction returned")
         return "No health condition detected", 0.0
     except Exception as e:
-        logger.error(f"Error analyzing symptoms: {str(e)}")
+        logger.error(f"Symptom analysis failed: {str(e)}")
         return f"Error: {str(e)}", 0.0
 
 def analyze_voice(audio_file):
     """Analyze voice for health indicators."""
     try:
-        # Ensure unique file name
-        unique_path = f"/tmp/gradio/{datetime.now().strftime('%Y%m%d%H%M%S%f')}_{os.path.basename(audio_file)}"
+        logger.debug(f"Starting analysis for: {audio_file}")
+        if not os.path.exists(audio_file):
+            logger.error(f"Audio file not found: {audio_file}")
+            return "Error: Audio file not found"
+        
+        unique_path = os.path.join(
+            tempfile.gettempdir(),
+            f"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
+        logger.debug(f"Renamed to: {audio_file}")
         
-        # Log audio file info
         file_hash = compute_file_hash(audio_file)
-        logger.info(f"Processing audio: {audio_file}, Hash: {file_hash}")
+        logger.info(f"Processing audio, Hash: {file_hash}")
         
-        # Load audio to verify
         audio, sr = librosa.load(audio_file, sr=16000)
-        logger.info(f"Audio shape: {audio.shape}, SR: {sr}, Duration: {len(audio)/sr:.2f}s, Mean: {np.mean(audio):.4f}, Std: {np.std(audio):.4f}")
+        logger.info(f"Audio loaded: shape={audio.shape}, SR={sr}, Duration={len(audio)/sr:.2f}s")
         
-        # Transcribe audio
         transcription = transcribe_audio(audio_file)
         if "Error" in transcription:
+            logger.error(f"Transcription error: {transcription}")
             return transcription
         
-        # Check for medication queries
         if any(keyword in transcription.lower() for keyword in ["medicine", "treatment"]):
-            return "Error: This tool does not provide medication or treatment advice."
+            logger.warning("Medication query detected")
+            return "Error: This tool does not provide medication advice"
         
-        # Analyze symptoms
         prediction, score = analyze_symptoms(transcription)
         if "Error" in prediction:
+            logger.error(f"Symptom analysis error: {prediction}")
             return prediction
         
-        # Generate one-line feedback
-        feedback = "No health condition detected, consult a doctor if symptoms persist." if prediction == "No health condition detected" else f"Possible {prediction.lower()} detected, consult a doctor."
-        
-        # Log debug info
-        logger.info(f"Feedback: {feedback}, Transcription: {transcription}, Prediction: {prediction}, Confidence: {score:.4f}, Hash: {file_hash}")
+        feedback = (
+            "No health condition detected, consult a doctor if symptoms persist."
+            if prediction == "No health condition detected"
+            else f"Possible {prediction.lower()} detected, consult a doctor."
+        )
+        logger.info(f"Feedback: {feedback}, Transcription: {transcription}, Prediction: {prediction}, Score: {score:.4f}")
         
-        # Clean up audio file
         try:
             os.remove(audio_file)
-            logger.info(f"Deleted audio file: {audio_file}")
+            logger.debug(f"Deleted audio file: {audio_file}")
         except Exception as e:
             logger.error(f"Failed to delete audio file: {str(e)}")
         
         return feedback
     except Exception as e:
-        logger.error(f"Error processing audio: {str(e)}")
+        logger.error(f"Voice analysis failed: {str(e)}")
         return f"Error: {str(e)}"
 
 def test_with_sample_audio():
@@ -195,28 +215,35 @@ def test_with_sample_audio():
     sample_audio_path = "audio_samples/sample.wav"
     if not os.path.exists(sample_audio_path):
         logger.warning("Sample audio not found; generating synthetic audio")
-        # Generate synthetic audio (sine wave to simulate voice)
         sr = 16000
         t = np.linspace(0, 2, 2 * sr)
         freq_mod = 440 + 10 * np.sin(2 * np.pi * 0.5 * t)
         amplitude_mod = 0.5 + 0.1 * np.sin(2 * np.pi * 0.3 * t)
         noise = 0.01 * np.random.normal(0, 1, len(t))
         dummy_audio = amplitude_mod * np.sin(2 * np.pi * freq_mod * t) + noise
-        sample_audio_path = "audio_samples/dummy_test.wav"
-        os.makedirs("audio_samples", exist_ok=True)
+        sample_audio_path = os.path.join(tempfile.gettempdir(), "dummy_test.wav")
+        os.makedirs(os.path.dirname(sample_audio_path), exist_ok=True)
         try:
             soundfile.write(dummy_audio, sr, sample_audio_path)
-            logger.info(f"Generated synthetic audio at: {sample_audio_path}")
+            logger.info(f"Generated synthetic audio: {sample_audio_path}")
         except Exception as e:
             logger.error(f"Failed to write synthetic audio: {str(e)}")
             return f"Error: Failed to generate synthetic audio: {str(e)}"
     
-    # Mock transcription for synthetic audio
     mock_transcription = "I have a cough and sore throat"
-    logger.info(f"Mock transcription for synthetic audio: {mock_transcription}")
+    logger.info(f"Mock transcription: {mock_transcription}")
     prediction, score = analyze_symptoms(mock_transcription)
-    feedback = "No health condition detected, consult a doctor if symptoms persist." if prediction == "No health condition detected" else f"Possible {prediction.lower()} detected, consult a doctor."
+    feedback = (
+        "No health condition detected, consult a doctor if symptoms persist."
+        if prediction == "No health condition detected"
+        else f"Possible {prediction.lower()} detected, consult a doctor."
+    )
     logger.info(f"Test feedback: {feedback}, Prediction: {prediction}, Score: {score:.4f}")
+    try:
+        os.remove(sample_audio_path)
+        logger.debug(f"Deleted test audio: {sample_audio_path}")
+    except Exception:
+        pass
     return feedback
 
 # Gradio interface