app.py

import gradio as gr
import requests
import librosa
import numpy as np
import os
import hashlib
from datetime import datetime
from simple_salesforce import Salesforce

# Salesforce credentials (store securely in environment variables)
SF_USERNAME = os.getenv("SF_USERNAME", "your_salesforce_username")
SF_PASSWORD = os.getenv("SF_PASSWORD", "your_salesforce_password")
SF_SECURITY_TOKEN = os.getenv("SF_SECURITY_TOKEN", "your_salesforce_security_token")
SF_INSTANCE_URL = os.getenv("SF_INSTANCE_URL", "https://your-salesforce-instance.salesforce.com")

# Hugging Face Inference API token (store in environment variables)
HF_TOKEN = os.getenv("HF_TOKEN", "your_huggingface_token")

# Initialize Salesforce connection
try:
    sf = Salesforce(
        username=SF_USERNAME,
        password=SF_PASSWORD,
        security_token=SF_SECURITY_TOKEN,
        instance_url=SF_INSTANCE_URL
    )
except Exception as e:
    print(f"Failed to connect to Salesforce: {str(e)}")
    sf = None

# Hugging Face API endpoints
WHISPER_API_URL = "https://api-inference.huggingface.co/models/openai/whisper-tiny.en"
SYMPTOM_API_URL = "https://api-inference.huggingface.co/models/abhirajeshbhai/symptom-2-disease-net"
HEADERS = {"Authorization": f"Bearer {HF_TOKEN}"}

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()

def transcribe_audio(audio_file):
    """Transcribe audio using Whisper API."""
    try:
        with open(audio_file, "rb") as f:
            data = f.read()
        response = requests.post(WHISPER_API_URL, headers=HEADERS, data=data)
        response.raise_for_status()
        result = response.json()
        transcription = result.get("text", "")
        print(f"Transcription: {transcription}")
        return transcription
    except Exception as e:
        print(f"Whisper API error: {str(e)}")
        return f"Error transcribing audio: {str(e)}"

def analyze_symptoms(text):
    """Analyze symptoms using Symptom-2-Disease API."""
    try:
        payload = {"inputs": text}
        response = requests.post(SYMPTOM_API_URL, headers=HEADERS, json=payload)
        response.raise_for_status()
        result = response.json()
        if result and isinstance(result, list) and len(result) > 0:
            prediction = result[0][0]["label"]
            score = result[0][0]["score"]
            print(f"Health Prediction: {prediction}, Score: {score:.4f}")
            return prediction, score
        return "No health condition predicted", 0.0
    except Exception as e:
        print(f"Symptom API error: {str(e)}")
        return f"Error analyzing symptoms: {str(e)}", 0.0

def analyze_voice(audio_file):
    """Analyze voice for health indicators."""
    try:
        # Log audio file info
        file_hash = compute_file_hash(audio_file)
        print(f"Processing audio file: {audio_file}, Hash: {file_hash}")
        
        # Load audio to verify format
        audio, sr = librosa.load(audio_file, sr=16000)
        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)
        if "Error transcribing" in transcription:
            return transcription
        
        # Analyze symptoms
        prediction, score = analyze_symptoms(transcription)
        if "Error analyzing" in prediction:
            return prediction
        
        # Generate feedback
        if prediction == "No health condition predicted":
            feedback = "No significant health indicators detected."
        else:
            feedback = f"Possible health condition: {prediction} (confidence: {score:.4f}). Consult a doctor."
        
        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."
        
        # Store in Salesforce
        if sf:
            store_in_salesforce(audio_file, feedback, transcription, prediction, score)
        
        # Clean up temporary audio file
        try:
            os.remove(audio_file)
            print(f"Deleted temporary audio file: {audio_file}")
        except Exception as e:
            print(f"Failed to delete audio file: {str(e)}")
        
        return feedback
    except Exception as e:
        return f"Error processing audio: {str(e)}"

def store_in_salesforce(audio_file, feedback, transcription, prediction, score):
    """Store analysis results in Salesforce."""
    try:
        sf.HealthAssessment__c.create({
            "AssessmentDate__c": datetime.utcnow().isoformat(),
            "Feedback__c": feedback,
            "Transcription__c": transcription,
            "Prediction__c": prediction,
            "Confidence__c": float(score),
            "AudioFileName__c": os.path.basename(audio_file)
        })
    except Exception as e:
        print(f"Failed to store in Salesforce: {str(e)}")

def test_with_sample_audio():
    """Test the app with a sample audio file."""
    sample_audio_path = "audio_samples/sample.wav"
    if os.path.exists(sample_audio_path):
        return analyze_voice(sample_audio_path)
    return "Sample audio file not found."

# Gradio interface
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 for preliminary health assessment. Supports English (transcription), with symptom analysis in English."
)

if __name__ == "__main__":
    print(test_with_sample_audio())
    iface.launch(server_name="0.0.0.0", server_port=7860)