Update app.py

app.py

@@ -1,21 +1,29 @@
 import gradio as gr
-import requests
 import librosa
 import numpy as np
 import os
 import hashlib
 from datetime import datetime
+from transformers import pipeline
+import soundfile as sf
+import torch
 
-# Hugging Face Inference API token (store in environment variables)
-HF_TOKEN = os.getenv("HF_TOKEN")  # Must be set in Space secrets
+# Initialize local models
+try:
+    # Whisper for speech-to-text (English-only)
+    whisper = pipeline("automatic-speech-recognition", model="openai/whisper-tiny.en", device=-1)  # CPU; use device=0 for GPU
+    print("Whisper model loaded successfully.")
+except Exception as e:
+    print(f"Failed to load Whisper model: {str(e)}")
+    whisper = None
 
-# Log HF_TOKEN status at startup
-print(f"HF_TOKEN status: {'Set' if HF_TOKEN else 'Not set'}")
-
-# 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}"}
+try:
+    # Symptom-2-Disease for health analysis
+    symptom_classifier = pipeline("text-classification", model="abhirajeshbhai/symptom-2-disease-net", device=-1)  # CPU
+    print("Symptom-2-Disease model loaded successfully.")
+except Exception as e:
+    print(f"Failed to load Symptom-2-Disease model: {str(e)}")
+    symptom_classifier = None
 
 def compute_file_hash(file_path):
     """Compute MD5 hash of a file to check uniqueness."""
@@ -26,80 +34,49 @@ def compute_file_hash(file_path):
     return hash_md5.hexdigest()
 
 def transcribe_audio(audio_file):
-    """Transcribe audio using Whisper API."""
-    if not HF_TOKEN:
-        error_msg = (
-            "Error transcribing audio: HF_TOKEN not set. Please set HF_TOKEN in Space secrets at "
-            "https://huggingface.co/spaces/your-username/HealthVoiceAnalyzer/settings. "
-            "Generate a token with Inference API access at https://huggingface.co/settings/tokens."
-        )
-        print(error_msg)
-        return error_msg
+    """Transcribe audio using local Whisper model."""
+    if not whisper:
+        return "Error: Whisper model not loaded. Check logs for details."
     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()
-        print(f"Whisper API response: {result}")
+        # Load and resample audio to 16,000 Hz
+        audio, sr = librosa.load(audio_file, sr=16000)
+        # Save as WAV for Whisper compatibility
+        temp_wav = f"/tmp/{os.path.basename(audio_file)}.wav"
+        sf.write(temp_wav, audio, sr)
+        
+        # Transcribe
+        result = whisper(temp_wav)
         transcription = result.get("text", "").strip()
+        print(f"Transcription: {transcription}")
+        
+        # Clean up temp file
+        try:
+            os.remove(temp_wav)
+        except Exception:
+            pass
+        
         if not transcription:
             return "Transcription empty. Please provide clear audio describing symptoms in English."
-        print(f"Transcription: {transcription}")
         return transcription
-    except requests.exceptions.HTTPError as e:
-        error_msg = f"Error transcribing audio: {str(e)}"
-        if e.response.status_code == 401:
-            error_msg = (
-                "Error transcribing audio: Unauthorized. Please check HF_TOKEN in Space secrets at "
-                "https://huggingface.co/spaces/your-username/HealthVoiceAnalyzer/settings. "
-                "Ensure token has Inference API access (get at https://huggingface.co/settings/tokens)."
-            )
-        print(f"Whisper API error: {error_msg}, Status: {e.response.status_code}")
-        return error_msg
     except Exception as e:
-        error_msg = f"Error transcribing audio: {str(e)}"
-        print(error_msg)
-        return error_msg
+        return f"Error transcribing audio: {str(e)}"
 
 def analyze_symptoms(text):
-    """Analyze symptoms using Symptom-2-Disease API."""
-    if not HF_TOKEN:
-        error_msg = (
-            "Error analyzing symptoms: HF_TOKEN not set. Please set HF_TOKEN in Space secrets at "
-            "https://huggingface.co/spaces/your-username/HealthVoiceAnalyzer/settings. "
-            "Generate a token with Inference API access at https://huggingface.co/settings/tokens."
-        )
-        print(error_msg)
-        return error_msg, 0.0
+    """Analyze symptoms using local Symptom-2-Disease model."""
+    if not symptom_classifier:
+        return "Error: Symptom-2-Disease model not loaded. Check logs for details.", 0.0
     try:
         if not text or "Error transcribing" in text:
             return "No valid transcription for analysis.", 0.0
-        payload = {"inputs": text}
-        response = requests.post(SYMPTOM_API_URL, headers=HEADERS, json=payload)
-        response.raise_for_status()
-        result = response.json()
-        print(f"Symptom API response: {result}")
+        result = symptom_classifier(text)
         if result and isinstance(result, list) and len(result) > 0:
-            prediction = result[0][0]["label"]
-            score = result[0][0]["score"]
+            prediction = result[0]["label"]
+            score = result[0]["score"]
             print(f"Health Prediction: {prediction}, Score: {score:.4f}")
             return prediction, score
         return "No health condition predicted", 0.0
-    except requests.exceptions.HTTPError as e:
-        error_msg = f"Error analyzing symptoms: {str(e)}"
-        if e.response.status_code == 401:
-            error_msg = (
-                "Error analyzing symptoms: Unauthorized. Please check HF_TOKEN in Space secrets at "
-                "https://huggingface.co/spaces/your-username/HealthVoiceAnalyzer/settings. "
-                "Ensure token has Inference API access (get at https://huggingface.co/settings/tokens)."
-            )
-        print(f"Symptom API error: {error_msg}, Status: {e.response.status_code}")
-        return error_msg, 0.0
     except Exception as e:
-        error_msg = f"Error analyzing symptoms: {str(e)}"
-        print(error_msg)
-        return error_msg, 0.0
+        return f"Error analyzing symptoms: {str(e)}", 0.0
 
 def analyze_voice(audio_file):
     """Analyze voice for health indicators."""
@@ -164,7 +141,7 @@ iface = gr.Interface(
     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. Ensure HF_TOKEN is set in Space secrets."
+    description="Record or upload a voice sample describing symptoms for preliminary health assessment. Supports English (transcription), with symptom analysis in English."
 )
 
 if __name__ == "__main__":