Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import librosa | |
| import numpy as np | |
| import torch | |
| from transformers import Wav2Vec2Processor, Wav2Vec2Model | |
| import requests | |
| import json | |
| import os | |
| from datetime import datetime | |
| # Salesforce API credentials (store securely in environment variables) | |
| SALESFORCE_API_URL = os.getenv("SALESFORCE_API_URL", "https://your-salesforce-instance.salesforce.com/services/data/v60.0/sobjects/HealthAssessment__c") | |
| SALESFORCE_TOKEN = os.getenv("SALESFORCE_TOKEN", "your_salesforce_token") | |
| # 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 (simplified for demo; real-world needs trained classifier) | |
| features = outputs.last_hidden_state.mean(dim=1).numpy() | |
| # Placeholder health analysis (replace with trained model) | |
| respiratory_score = np.mean(features) # Mock score | |
| mental_health_score = np.std(features) # Mock score | |
| feedback = "" | |
| if respiratory_score > 0.5: | |
| feedback += "Possible respiratory issue detected; consult a doctor. " | |
| if mental_health_score > 0.3: | |
| feedback += "Possible stress indicators detected; consider professional advice. " | |
| if not feedback: | |
| feedback = "No significant health indicators detected." | |
| feedback += "\n\n**Disclaimer**: This is not a diagnostic tool. Consult a healthcare provider for medical advice." | |
| # Store in Salesforce | |
| store_in_salesforce(audio_file, feedback, respiratory_score, mental_health_score) | |
| 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.""" | |
| headers = { | |
| "Authorization": f"Bearer {SALESFORCE_TOKEN}", | |
| "Content-Type": "application/json" | |
| } | |
| data = { | |
| "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) | |
| } | |
| response = requests.post(SALESFORCE_API_URL, headers=headers, json=data) | |
| if response.status_code != 201: | |
| print(f"Failed to store in Salesforce: {response.text}") | |
| # 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__": | |
| iface.launch(server_name="0.0.0.0", server_port=7860) |