Spaces:
Sleeping
Sleeping
File size: 3,332 Bytes
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 |
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) |