Upload app.py

app.py

@@ -0,0 +1,486 @@
+"""
+Gradio interface for Speech Disfluency Detection
+-------------------------------------------------
+Wraps ChildFluencyNet (stage2_fyp_best.pt).
+Part of a multi-modal neurodevelopmental assessment system.
+
+Same inference logic as FastAPI version, but deployed on HF Spaces via Gradio.
+"""
+
+import io
+import json
+import os
+import re
+import sys
+import time
+import yaml
+import tempfile
+import traceback
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+import torch
+import numpy as np
+import librosa
+import gradio as gr
+from huggingface_hub import hf_hub_download
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# ── Path setup ────────────────────────────────────────────────────────────
+ROOT = Path(__file__).resolve().parent
+sys.path.insert(0, str(ROOT))
+
+# Import your model
+try:
+    from childfluency import ChildFluencyNet
+except ImportError:
+    print("[startup] Warning: Could not import ChildFluencyNet from local childfluency.py")
+    print("[startup] Make sure childfluency.py is in the same directory as app.py")
+
+# ── Load params ───────────────────────────────────────────────────────────
+with open(ROOT / 'params.yaml') as f:
+    params = yaml.safe_load(f)
+
+TARGET_SR  = params['data']['target_sr']   # 16000
+WINDOW_SEC = params['data']['window_sec']  # 4.0
+STRIDE_SEC = params['data']['stride_sec']  # 2.0
+MIN_RMS    = params['data']['min_rms']     # 0.001
+
+LPE_OUTPUT_DIM     = params['model']['lpe_dim']  # 32
+LPE_INPUT_FEATURES = 6                           # 6
+
+DEFAULT_CKPT = os.environ.get(
+    'DISFLUENCY_CHECKPOINT',
+    str(ROOT / 'stage2_fyp_best.pt')
+)
+
+# ── Gemini & Groq ─────────────────────────────────────────────────────────
+GEMINI_API_KEY = os.environ.get('GEMINI_API_KEY')
+GROK_API_KEY   = os.environ.get('GROK_API_KEY')
+gemini_model = None
+
+if GEMINI_API_KEY:
+    try:
+        import google.generativeai as genai
+        genai.configure(api_key=GEMINI_API_KEY)
+        gemini_model = genai.GenerativeModel('gemini-2.0-flash')
+        print("[startup] Gemini API initialised.")
+    except Exception as e:
+        print(f"[startup] Warning: Gemini init failed: {e}")
+else:
+    print("[startup] GEMINI_API_KEY not set — will rely on Groq fallback.")
+
+if GROK_API_KEY:
+    print("[startup] Groq API key found — available as fallback.")
+else:
+    print("[startup] GROK_API_KEY not set — Groq fallback disabled.")
+
+# ── Load Model at Startup ──────────────────────────────────────────────────
+print("[startup] Loading model...")
+device = 'cuda' if torch.cuda.is_available() else 'cpu'
+print(f"[startup] Device: {device}")
+
+# Download model from HF Hub
+print("[startup] Downloading model from HuggingFace Hub...")
+try:
+    model_path = hf_hub_download(
+        repo_id="yourusername/developmental-stuttering-detection",
+        filename="stage2_fyp_best.pt",
+        cache_dir="./models"
+    )
+except Exception as e:
+    print(f"[startup] Warning: HF Hub download failed: {e}")
+    model_path = Path(DEFAULT_CKPT)
+
+print(f"[startup] Model path: {model_path}")
+
+# Initialize and load model
+model = ChildFluencyNet(
+    wavlm_name   = params['model'].get('wavlm_name', 'microsoft/wavlm-large'),
+    acoustic_dim = params['model']['acoustic_dim'],
+    lpe_dim      = LPE_OUTPUT_DIM,
+    hidden_dim   = params['model']['hidden_dim'],
+    dropout      = params['model']['dropout']
+).to(device)
+
+if Path(model_path).exists():
+    ckpt = torch.load(str(model_path), map_location=device)
+    model.load_state_dict(ckpt['model'])
+    epoch = ckpt.get('epoch', '?')
+    val_f1 = round(float(ckpt.get('val_f1', 0)), 4)
+    print(f"[startup] Loaded → epoch {epoch}, val_f1={val_f1}")
+else:
+    print(f"[startup] WARNING: checkpoint not found at {model_path}")
+
+model.eval()
+print("[startup] Model ready for inference.")
+
+
+# ── Inference Logic (kept from FastAPI version) ────────────────────────────
+
+def segment_audio(audio: np.ndarray):
+    window_samples = int(WINDOW_SEC * TARGET_SR)
+    stride_samples = int(STRIDE_SEC * TARGET_SR)
+    windows, start = [], 0
+    while start + window_samples <= len(audio):
+        seg = audio[start: start + window_samples]
+        if float(np.sqrt(np.mean(seg ** 2))) >= MIN_RMS:
+            windows.append((
+                start,
+                start + window_samples,
+                round(start / TARGET_SR, 3),
+                round((start + window_samples) / TARGET_SR, 3),
+            ))
+        start += stride_samples
+    return windows
+
+
+def predict_windows(audio: np.ndarray, threshold: float = 0.5, batch_size: int = 16) -> List[dict]:
+    """Segment audio and run SLD inference."""
+    windows = segment_audio(audio)
+    if not windows:
+        return []
+
+    results = []
+    for b in range(0, len(windows), batch_size):
+        batch = windows[b: b + batch_size]
+        audio_t = torch.from_numpy(
+            np.stack([audio[s:e].astype(np.float32) for s, e, _, _ in batch])
+        ).to(device)
+
+        lpe_t = torch.full(
+            (len(batch), LPE_INPUT_FEATURES), 0.5, dtype=torch.float32
+        ).to(device)
+
+        with torch.no_grad():
+            if device == 'cuda':
+                with torch.cuda.amp.autocast():
+                    out = model(audio_t, lpe_t)
+            else:
+                out = model(audio_t, lpe_t)
+
+        probs = torch.sigmoid(out['sld_logit']).cpu().numpy().flatten()
+
+        for i, (_, _, s_sec, e_sec) in enumerate(batch):
+            p = float(probs[i])
+            results.append({
+                'window_idx': b + i,
+                'start_sec' : s_sec,
+                'end_sec'   : e_sec,
+                'sld_prob'  : round(p, 4),
+                'prediction': 'SLD' if p >= threshold else 'fluent',
+            })
+
+    return results
+
+
+def compute_confidence(positive_rate: float, n_windows: int) -> str:
+    if n_windows < 5:
+        return 'low'
+    consensus = max(positive_rate, 1 - positive_rate)
+    if consensus >= 0.80:
+        return 'high'
+    elif consensus >= 0.60:
+        return 'medium'
+    return 'low'
+
+
+# ── Gradio Main Function ───────────────────────────────────────────────────
+
+def analyze_speech(audio_file, threshold: float = 0.5) -> Tuple[str, str]:
+    """
+    Main Gradio function: analyze audio and return results + optional analysis.
+    
+    Args:
+        audio_file: Path to uploaded audio file (Gradio provides this)
+        threshold: SLD probability threshold (default 0.5)
+    
+    Returns:
+        (results_json, analysis_text)
+    """
+    try:
+        t_start = time.time()
+        
+        # Load audio
+        audio, _ = librosa.load(audio_file, sr=TARGET_SR, mono=True)
+        duration_sec = round(len(audio) / TARGET_SR, 2)
+        
+        if duration_sec < WINDOW_SEC:
+            error_msg = f"Audio too short ({duration_sec}s). Need ≥{WINDOW_SEC}s."
+            return (
+                json.dumps({"error": error_msg, "status": "failed"}, indent=2),
+                f"❌ {error_msg}"
+            )
+        
+        # Run inference
+        window_scores = predict_windows(audio, threshold=threshold)
+        
+        if not window_scores:
+            error_msg = "No valid speech windows found."
+            return (
+                json.dumps({"error": error_msg, "status": "failed"}, indent=2),
+                f"❌ {error_msg}"
+            )
+        
+        # Compute metrics
+        n_total = len(window_scores)
+        n_positive = sum(1 for w in window_scores if w['prediction'] == 'SLD')
+        pos_rate = round(n_positive / n_total, 4)
+        
+        pos_probs = [w['sld_prob'] for w in window_scores if w['prediction'] == 'SLD']
+        sld_prob = round(
+            float(np.mean(pos_probs)) if pos_probs
+            else float(np.mean([w['sld_prob'] for w in window_scores])),
+            4
+        )
+        
+        elapsed = round(time.time() - t_start, 2)
+        
+        # Build results dict
+        results = {
+            'filename': Path(audio_file).name,
+            'duration_sec': duration_sec,
+            'sld_probability': sld_prob,
+            'prediction': 'stuttering_detected' if n_positive > 0 else 'fluent',
+            'confidence': compute_confidence(pos_rate, n_total),
+            'n_windows_total': n_total,
+            'n_windows_positive': n_positive,
+            'positive_rate': pos_rate,
+            'processing_time_sec': elapsed,
+            'note': "LPE features set to neutral (0.5). Ablation study: ΔF1=+0.002 (negligible)."
+        }
+        
+        results_json = json.dumps(results, indent=2)
+        
+        # Generate text summary
+        severity = (
+            'Severe'   if pos_rate >= 0.50 else
+            'Moderate' if pos_rate >= 0.25 else
+            'Mild'     if pos_rate >= 0.10 else
+            'Minimal'
+        )
+        
+        summary = f"""
+✅ ANALYSIS COMPLETE
+
+📊 Results:
+- File: {Path(audio_file).name}
+- Duration: {duration_sec}s
+- SLD Rate: {pos_rate * 100:.1f}%
+- Severity: {severity}
+- Confidence: {results['confidence'].upper()}
+- Windows Analyzed: {n_total}
+- Windows with SLD: {n_positive}
+
+🔍 Prediction: {'🚨 STUTTERING DETECTED' if results['prediction'] == 'stuttering_detected' else '✅ FLUENT (No significant stuttering)'}
+
+⏱️ Processing time: {elapsed}s
+
+📝 Note: This is a screening aid only. Consult a Speech-Language Pathologist for formal evaluation.
+"""
+        
+        return (results_json, summary)
+    
+    except Exception as e:
+        error_msg = f"Error: {str(e)}"
+        return (
+            json.dumps({"error": error_msg, "status": "failed"}, indent=2),
+            f"❌ {error_msg}"
+        )
+
+
+# ── Optional: Generate Gemini Analysis ─────────────────────────────────────
+
+def generate_analysis(
+    audio_file,
+    child_name: str = "Child",
+    child_age: str = "?",
+    threshold: float = 0.5
+) -> str:
+    """Generate a detailed Gemini-powered analysis of the results."""
+    
+    try:
+        # First, run the prediction
+        audio, _ = librosa.load(audio_file, sr=TARGET_SR, mono=True)
+        duration_sec = round(len(audio) / TARGET_SR, 2)
+        
+        window_scores = predict_windows(audio, threshold=threshold)
+        if not window_scores:
+            return "❌ No valid speech windows found."
+        
+        n_total = len(window_scores)
+        n_positive = sum(1 for w in window_scores if w['prediction'] == 'SLD')
+        pos_rate = round(n_positive / n_total, 4)
+        sld_pct = round(pos_rate * 100, 1)
+        
+        detected = n_positive > 0
+        severity = (
+            'Severe'   if pos_rate >= 0.50 else
+            'Moderate' if pos_rate >= 0.25 else
+            'Mild'     if pos_rate >= 0.10 else
+            'Minimal'
+        )
+        
+        confidence = compute_confidence(pos_rate, n_total)
+        
+        # Fallback text
+        fallback = (
+            f"OVERALL ASSESSMENT\n"
+            f"{child_name} ({child_age} years) showed {severity.lower()} stutter-like disfluencies "
+            f"in {sld_pct}% of the recorded speech "
+            f"({'stuttering detected' if detected else 'no significant stuttering detected'}).\n\n"
+            f"ACOUSTIC DETAILS\n"
+            f"Total windows analysed: {n_total}\n"
+            f"Windows with SLD: {n_positive}\n"
+            f"SLD rate: {sld_pct}%\n"
+            f"Confidence: {confidence}\n"
+            f"Recording duration: {duration_sec:.1f}s\n\n"
+            f"RECOMMENDATIONS\n"
+            f"1. {'Consult a Speech-Language Pathologist for a formal evaluation.' if detected else 'Continue monitoring speech fluency over time.'}\n"
+            f"2. Use slow, relaxed speech when speaking with the child.\n"
+            f"3. Create a low-pressure environment to reduce communication anxiety.\n\n"
+            f"IMPORTANT DISCLAIMER\n"
+            f"This is an automated screening aid and does not constitute a clinical diagnosis. "
+            f"A qualified Speech-Language Pathologist must evaluate the child formally."
+        )
+        
+        prompt = f"""You are a Speech-Language Pathologist writing a screening summary for a parent.
+
+YOUR INSTRUCTIONS:
+1. DO NOT recalculate any scores.
+2. DO NOT change the classification.
+3. Your ONLY job is to format the provided data into a warm, empathetic, easy-to-read report.
+
+CHILD INFORMATION:
+- Name: {child_name}
+- Age: {child_age} years
+
+ACOUSTIC SCREENING RESULTS (Do not alter):
+- SLD (Stutter-Like Disfluency) Rate: {sld_pct}%
+- Severity Category: {severity}
+- System Classification: {'STUTTERING DETECTED' if detected else 'FLUENT (No significant stuttering)'}
+- Confidence: {confidence}
+- Recording Duration: {duration_sec:.1f} seconds
+- Windows Analysed: {n_total}, Windows with SLD: {n_positive}
+
+Please write a comprehensive, plain-text summary (no markdown, no **, no ##, no special characters).
+
+Structure your response with these exact section headings on their own lines:
+
+OVERALL ASSESSMENT
+2-3 sentences summarising the result in plain, reassuring language.
+
+WHAT THIS MEANS
+Explain what stutter-like disfluencies are and how this result relates to typical speech development for this age.
+
+RECOMMENDATIONS
+Simple numbered points (1., 2., 3.) — at least three actionable suggestions for parents.
+
+IMPORTANT DISCLAIMER
+State clearly this is a screening tool, not a clinical diagnosis, and that a Speech-Language Pathologist evaluation is required.
+
+Keep the tone warm, empathetic, and parent-friendly. Plain text only."""
+
+        # Tier 1: Gemini
+        if gemini_model is not None:
+            try:
+                response = gemini_model.generate_content(prompt)
+                return response.text
+            except Exception as e:
+                print(f"[generate-analysis] Gemini error: {e}. Falling back to Groq.")
+
+        # Tier 2: Groq
+        if GROK_API_KEY:
+            try:
+                from groq import Groq
+                groq_client = Groq(api_key=GROK_API_KEY)
+                groq_response = groq_client.chat.completions.create(
+                    model="llama-3.3-70b-versatile",
+                    messages=[{"role": "user", "content": prompt}],
+                    temperature=0.65,
+                    max_tokens=1024,
+                )
+                return groq_response.choices[0].message.content
+            except Exception as e:
+                print(f"[generate-analysis] Groq error: {e}. Using fallback.")
+
+        # Tier 3: Fallback
+        return fallback
+    
+    except Exception as e:
+        return f"❌ Error generating analysis: {str(e)}"
+
+
+# ── Gradio Interface ───────────────────────────────────────────────────────
+
+with gr.Blocks(title="Developmental Stuttering Detection") as demo:
+    gr.Markdown("# 🎤 Developmental Stuttering Detection")
+    gr.Markdown("Upload a recording of a child speaking to screen for signs of stutter-like disfluencies.")
+    
+    with gr.Tabs():
+        # TAB 1: Prediction
+        with gr.Tab("Analysis"):
+            with gr.Row():
+                audio_input = gr.Audio(
+                    type="filepath",
+                    label="Upload Audio (WAV recommended)",
+                    info="Minimum 4 seconds"
+                )
+                threshold_slider = gr.Slider(
+                    minimum=0.1,
+                    maximum=0.9,
+                    value=0.5,
+                    step=0.05,
+                    label="SLD Threshold",
+                    info="Lower = more sensitive, Higher = more conservative"
+                )
+            
+            analyze_btn = gr.Button("🔍 Analyze", variant="primary")
+            
+            with gr.Row():
+                results_output = gr.JSON(label="Raw Results")
+                summary_output = gr.Textbox(label="Summary", lines=10)
+            
+            analyze_btn.click(
+                fn=analyze_speech,
+                inputs=[audio_input, threshold_slider],
+                outputs=[results_output, summary_output]
+            )
+        
+        # TAB 2: Detailed Analysis (Gemini-powered)
+        with gr.Tab("Detailed Report"):
+            gr.Markdown("Generate a detailed clinical-style report powered by AI.")
+            
+            audio_input_2 = gr.Audio(type="filepath", label="Upload Audio")
+            child_name_input = gr.Textbox(value="Child", label="Child's Name")
+            child_age_input = gr.Textbox(value="?", label="Child's Age")
+            threshold_slider_2 = gr.Slider(
+                minimum=0.1, maximum=0.9, value=0.5, step=0.05, label="SLD Threshold"
+            )
+            
+            generate_btn = gr.Button("📄 Generate Report", variant="primary")
+            report_output = gr.Textbox(label="Generated Report", lines=15)
+            
+            generate_btn.click(
+                fn=generate_analysis,
+                inputs=[audio_input_2, child_name_input, child_age_input, threshold_slider_2],
+                outputs=report_output
+            )
+    
+    gr.Markdown("""
+---
+⚠️ **IMPORTANT DISCLAIMER**
+
+This tool is a **screening aid only** and does not constitute a clinical diagnosis.
+All results must be reviewed by a qualified Speech-Language Pathologist.
+
+For more information, consult a healthcare professional.
+""")
+
+
+# ── Launch ─────────────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    demo.launch(share=False)