Spaces:
Sleeping
Sleeping
File size: 4,537 Bytes
410fd66 68390a5 410fd66 68390a5 a937006 68390a5 410fd66 b86bc45 410fd66 b86bc45 410fd66 b86bc45 410fd66 b86bc45 410fd66 68390a5 410fd66 b86bc45 410fd66 68390a5 410fd66 a937006 b86bc45 a937006 410fd66 a937006 410fd66 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
import gradio as gr
import librosa
import numpy as np
import torch
from transformers import Wav2Vec2Processor, Wav2Vec2Model
from simple_salesforce import Salesforce
import os
from datetime import datetime
# 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")
# 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
# Load Wav2Vec2 model for speech feature extraction
processor = Wav2Vec2Processor.from_pretrained("facebook/wav2vec2-base-960h")
model = Wav2Vec2Model.from_pretrained("facebook/wav2vec2-base-960h")
def analyze_voice(audio_file):
"""Analyze voice for health indicators."""
try:
# Load audio file
audio, sr = librosa.load(audio_file, sr=16000)
# Process audio for Wav2Vec2
inputs = processor(audio, sampling_rate=16000, return_tensors="pt", padding=True)
with torch.no_grad():
outputs = model(**inputs)
# Extract features
features = outputs.last_hidden_state.mean(dim=1).numpy()
# Mock health analysis (for testing)
respiratory_score = np.mean(features)
mental_health_score = np.std(features)
# Debug: Print scores
print(f"Respiratory Score: {respiratory_score:.4f}, Mental Health Score: {mental_health_score:.4f}")
# Adjusted thresholds for testing
feedback = ""
if respiratory_score > 0.1:
feedback += f"Possible respiratory issue detected (score: {respiratory_score:.4f}); consult a doctor. "
if mental_health_score > 0.1:
feedback += f"Possible stress indicators detected (score: {mental_health_score:.4f}); consider professional advice. "
if not feedback:
feedback = "No significant health indicators detected."
feedback += f"\n\n**Debug Info**: Respiratory Score = {respiratory_score:.4f}, Mental Health Score = {mental_health_score:.4f}"
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, respiratory_score, mental_health_score)
# Clean up temporary audio file (for HIPAA/GDPR compliance)
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, respiratory_score, mental_health_score):
"""Store analysis results in Salesforce."""
try:
sf.HealthAssessment__c.create({
"AssessmentDate__c": datetime.utcnow().isoformat(),
"Feedback__c": feedback,
"RespiratoryScore__c": float(respiratory_score),
"MentalHealthScore__c": float(mental_health_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" # Adjust if using common_voice_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 for preliminary health assessment. Supports English, Spanish, Hindi, Mandarin."
)
if __name__ == "__main__":
print(test_with_sample_audio()) # Run test on startup
iface.launch(server_name="0.0.0.0", server_port=7860) |