Update app.py

app.py

@@ -235,30 +235,32 @@ def analyze_symptoms(text):
         elif isinstance(result, (tuple, list)):
             # Flatten nested tuples/lists
             flattened = []
-            def flatten(item):
+            def flatten(item, depth=0, max_depth=10):
+                if depth > max_depth:
+                    logger.warning(f"Max recursion depth exceeded: {item}")
+                    return
                 if isinstance(item, (tuple, list)):
                     for subitem in item:
-                        flatten(subitem)
-                else:
+                        flatten(subitem, depth + 1, max_depth)
+                elif isinstance(item, dict):
                     flattened.append(item)
+                else:
+                    logger.warning(f"Skipping non-dict item: {item}")
             flatten(result)
-            result = flattened
-            if not result:
+            if not flattened:
                 logger.warning("Flattened model output is empty")
-            elif isinstance(result, list):
-                if not all(isinstance(item, dict) for item in result):
-                    logger.warning(f"Non-dictionary items in result: {result}")
-                elif not all("label" in item and "score" in item for item in result):
-                    logger.warning(f"Missing label or score in result: {result}")
-                else:
-                    prediction = result[0]["label"]
-                    score = result[0]["score"]
+            elif not all(isinstance(item, dict) for item in flattened):
+                logger.warning(f"Non-dictionary items in flattened result: {flattened}")
+            elif not all("label" in item and "score" in item for item in flattened):
+                logger.warning(f"Missing label or score in flattened result: {flattened}")
+            else:
+                prediction = flattened[0]["label"]
+                score = flattened[0]["score"]
         elif isinstance(result, dict):
-            logger.debug("Model returned single dictionary; wrapping in list")
-            result = [result]
-            if "label" in result[0] and "score" in result[0]:
-                prediction = result[0]["label"]
-                score = result[0]["score"]
+            logger.debug("Model returned single dictionary")
+            if "label" in result and "score" in result:
+                prediction = result["label"]
+                score = result["score"]
             else:
                 logger.warning(f"Missing label or score in dictionary: {result}")
 
@@ -390,97 +392,6 @@ async def analyze_voice(audio_file, language="en", user_id="anonymous", consent_
         feedback = (
             "No health condition detected, consult a doctor if symptoms persist. This is not a medical diagnosis."
             if prediction == "No health condition detected"
-            else f"Possible {prediction.lower()} detected, consult a doctor. This is not a medical diagnosis."
+            else f"Possible {prediction.lower()} detected based on symptoms like '{transcription.lower()}', consult a doctor. This is not a medical diagnosis."
         )
-        logger.info(f"Feedback: {feedback}, Transcription: {transcription}, Prediction: {prediction}, Score: {score:.4f}")
-        
-        # Save to Salesforce
-        save_to_salesforce(user_id, transcription, prediction, score, feedback, consent_granted)
-        
-        try:
-            os.remove(audio_file)
-            logger.debug(f"Deleted audio file: {audio_file}")
-        except Exception as e:
-            logger.error(f"Failed to delete audio file: {str(e)}")
-        
-        # Speak response
-        await speak_response(feedback)
-        
-        return feedback
-    except Exception as e:
-        logger.error(f"Voice analysis failed: {str(e)}")
-        return f"Error: {str(e)}"
-
-async def test_with_sample_audio(language="en", user_id="anonymous", consent_granted=True):
-    """Test with synthetic audio."""
-    temp_dir = os.path.join(tempfile.gettempdir(), "audio_samples")
-    if not ensure_writable_dir(temp_dir):
-        fallback_dir = os.path.join(os.getcwd(), "temp_audio_samples")
-        if not ensure_writable_dir(fallback_dir):
-            logger.error(f"Temp directories {temp_dir} and {fallback_dir} not writable")
-            return f"Error: Temp directories not writable"
-        temp_dir = fallback_dir
-    
-    sample_audio_path = os.path.join(temp_dir, "dummy_test.wav")
-    logger.info(f"Generating synthetic audio at: {sample_audio_path}")
-    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
-    try:
-        soundfile.write(dummy_audio, sr, 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)}"
-    
-    if not os.path.exists(sample_audio_path):
-        logger.error(f"Synthetic audio not created: {sample_audio_path}")
-        return f"Error: Synthetic audio not created: {sample_audio_path}"
-    
-    mock_transcription = "I have a cough and sore throat"
-    logger.info(f"Mock transcription: {mock_transcription}")
-    prediction, score = analyze_symptoms(mock_transcription)
-    feedback = (
-        "No health condition detected, consult a doctor if symptoms persist. This is not a medical diagnosis."
-        if prediction == "No health condition detected"
-        else f"Possible {prediction.lower()} detected, consult a doctor. This is not a medical diagnosis."
-    )
-    logger.info(f"Test feedback: {feedback}, Prediction: {prediction}, Score: {score:.4f}")
-    
-    # Save to Salesforce
-    save_to_salesforce(user_id, mock_transcription, prediction, score, feedback, consent_granted)
-    
-    try:
-        os.remove(sample_audio_path)
-        logger.debug(f"Deleted test audio: {sample_audio_path}")
-    except Exception:
-        pass
-    return feedback
-
-async def voicebot_interface(audio_file, language="en", user_id="anonymous", consent_granted=True):
-    """Gradio interface wrapper."""
-    return await analyze_voice(audio_file, language, user_id, consent_granted)
-
-# Gradio interface
-iface = gr.Interface(
-    fn=voicebot_interface,
-    inputs=[
-        gr.Audio(type="filepath", label="Record or Upload Voice (WAV, MP3, FLAC, 1+ sec)"),
-        gr.Dropdown(["en", "es", "hi", "zh"], label="Language", value="en"),
-        gr.Textbox(label="User ID (optional)", value="anonymous"),
-        gr.Checkbox(label="Consent to store data", value=True)
-    ],
-    outputs=gr.Textbox(label="Health Assessment Feedback"),
-    title="Smart Voicebot for Public Health",
-    description="Record or upload a voice sample describing symptoms (e.g., 'I have a cough') for preliminary health assessment. Supports English, Spanish, Hindi, Mandarin. Not a diagnostic tool. Data is encrypted and stored with consent. Complies with HIPAA/GDPR."
-)
-
-if __name__ == "__main__":
-    logger.info("Starting Voice Health Analyzer")
-    # Test with synthetic audio
-    loop = asyncio.get_event_loop()
-    print(loop.run_until_complete(test_with_sample_audio()))
-    iface.launch(server_name="0.0.0.0", server_port=7860)
+        logger.info(f"Feedback: {fat the BRD requirements, and ensures accurate feedback reflecting the voice input. Let me know how it performs, and share any logs or errors if issues persist!