Upload ml/model/fusion.py with huggingface_hub

ml/model/fusion.py

@@ -1,72 +1,102 @@
 """
-ml/model/fusion.py - Multi-modal fusion + per-user self-calibration
-====================================================================
-The heart of the "not over-strict" requirement. It fuses every modality into
-ONE diagnosis, graded into coarse buckets, and calibrates those bucket
-thresholds against the user's own "my normal" recording so a natural-speech
-baseline is not flagged as disordered.
-
-Honesty & Clinical Realism:
-  - Silence Guard: Pure silence or missing speech returns 'silent' status
-    rather than penalizing the user with a severe disease rating.
-  - Stutter probability P(stutter) is mapped into clinical buckets:
-      P < 0.60 -> fluent (healthy speech cadence)
-      0.60 <= P < 0.78 -> mild (occasional repetition or micro-hesitation)
-      0.78 <= P < 0.90 -> moderate (frequent disfluent events)
-      P >= 0.90 -> severe (prolonged blocks or continuous repetitions)
-  - Pronunciation GOP directly evaluates phonetic accuracy and alignment.
+ml/model/fusion.py - Multi-modal fusion + per-user self-calibration + uncertainty
+==================================================================================
+Fuses neural disfluency classification, CTC phonetic alignment, and Praat acoustic
+correlates of phonation into a unified, transparent screening report.
+
+Design & Scientific Transparency Notes:
+  - Heuristic Weighting: The Fluency Index (0..100) utilizes empirically tuned
+    weights (40% disfluency, 45% pronunciation, 15% phonation correlates) designed
+    for educational screening & practice feedback.
+  - Uncertainty Estimation: Flags borderline classification regions and noisy
+    recordings with explicit confidence scoring rather than overconfident assertions.
+  - Clinical Grounding: This prototype measures acoustic correlates of phonation
+    and phonetic alignment; it is not an FDA-cleared diagnostic device.
 """
 from __future__ import annotations
-from typing import Optional
+from typing import Optional, Dict, Any, Tuple
 
 import numpy as np
 
-BUCKETS = ["fluent", "mild", "moderate", "severe", "silent"]
+BUCKETS = ["fluent", "mild", "moderate", "severe", "silent", "uncertain"]
+
+# Configurable empirical thresholds
+THRESHOLD_MILD = 0.60
+THRESHOLD_MODERATE = 0.78
+THRESHOLD_SEVERE = 0.90
+BORDERLINE_BAND = (0.42, 0.58)
 
 
 def bucket_of(sev: int) -> str:
     if sev == 4:
         return "silent"
+    if sev == 5:
+        return "uncertain"
     return BUCKETS[min(max(sev, 0), 3)]
 
 
-def severity_from_softmax(probs) -> int | None:
-    """0..3 clinical severity from a stutter softmax."""
+def severity_from_softmax(probs) -> Tuple[Optional[int], float, str]:
+    """
+    0..3 clinical severity from a stutter softmax with uncertainty tracking.
+    Returns: (severity_bucket, p_stutter, confidence_rating)
+    """
     if probs is None:
-        return None
+        return None, 0.0, "unknown"
     p = np.asarray(probs, dtype=float)
     if p.sum() > 0:
         p = p / p.sum()
     if len(p) == 0:
-        return None
+        return None, 0.0, "unknown"
     
     p_stutter = float(p[1]) if len(p) == 2 else float(np.sum(p[1:]))
-    if p_stutter < 0.60:
-        return 0
-    elif p_stutter < 0.78:
-        return 1
-    elif p_stutter < 0.90:
-        return 2
+    
+    # Uncertainty / Confidence estimation
+    margin = abs(p_stutter - 0.50) * 2.0  # 0.0 at 50/50 tossup, 1.0 at pure certainty
+    if BORDERLINE_BAND[0] <= p_stutter <= BORDERLINE_BAND[1]:
+        confidence = "low (borderline transition)"
+    elif margin < 0.35:
+        confidence = "moderate"
+    else:
+        confidence = "high"
+
+    if p_stutter < THRESHOLD_MILD:
+        return 0, p_stutter, confidence
+    elif p_stutter < THRESHOLD_MODERATE:
+        return 1, p_stutter, confidence
+    elif p_stutter < THRESHOLD_SEVERE:
+        return 2, p_stutter, confidence
     else:
-        return 3
+        return 3, p_stutter, confidence
 
 
-def articulation_severity(articulatory: dict) -> int:
-    """0..3 roughness from real Praat values."""
+def articulation_severity(articulatory: dict) -> Tuple[int, str]:
+    """
+    0..3 roughness from acoustic correlates of phonation (Praat PointProcess).
+    Returns: (severity_index, summary_string)
+    """
     if articulatory.get("is_silent"):
-        return 4
+        return 4, "silent"
     s = 0
-    if articulatory.get("hnr_db", 20) < 9.0: s += 1
-    if articulatory.get("jitter", 0.0) > 0.05: s += 1
-    if articulatory.get("voiced_ratio", 1.0) < 0.25: s += 1
-    return min(s, 3)
+    flags = []
+    if articulatory.get("hnr_db", 20) < 9.0:
+        s += 1
+        flags.append("Low HNR (<9 dB)")
+    if articulatory.get("jitter", 0.0) > 0.05:
+        s += 1
+        flags.append("High Jitter (>5%)")
+    if articulatory.get("voiced_ratio", 1.0) < 0.25:
+        s += 1
+        flags.append("Low Voicing (<25%)")
+    
+    desc = ", ".join(flags) if flags else "Normal acoustic range"
+    return min(s, 3), desc
 
 
 # ---------------------------------------------------------------------------
 # Per-user self-calibration
 # ---------------------------------------------------------------------------
 class CalibrationProfile:
-    """Per-user bucket-offset derived from their own 'normal' recording."""
+    """Per-user bucket-offset derived from their own healthy baseline recording."""
 
     def __init__(self, normal_fluent: Optional[float] = None):
         self.normal_fluent = normal_fluent
@@ -77,7 +107,7 @@ class CalibrationProfile:
 
     def shift(self, raw: int) -> int:
         """Bucket offset so the user's natural baseline == fluent."""
-        if not self.active or raw == 4:
+        if not self.active or raw in (4, 5):
             return raw
         offset = 1 if self.normal_fluent < 0.60 else 0
         return max(0, raw - offset)
@@ -94,16 +124,21 @@ def calibrate_from_normal(normal_clip_probs) -> CalibrationProfile:
 
 
 # ---------------------------------------------------------------------------
-# Top-level fusion
+# Top-level multi-modal fusion
 # ---------------------------------------------------------------------------
 def diag_statistics(
     probs,
     pronunciation: dict,
     articulatory: dict,
     calibration: Optional[CalibrationProfile] = None,
+    weights: Tuple[float, float, float] = (0.40, 0.45, 0.15),
 ) -> dict:
-    """Fuse all modalities into a single human-facing diagnosis dict."""
+    """
+    Fuse all modalities into an auditable diagnostic screening dictionary.
+    Weights: (w_stutter, w_pronunciation, w_acoustics)
+    """
     cal = calibration or CalibrationProfile()
+    w_stut, w_pron, w_art = weights
 
     # 1. Silence Guard
     is_silent = False
@@ -121,21 +156,21 @@ def diag_statistics(
                 "overall": "silent",
             },
             "fluency_100": None,
+            "confidence": "high (silence confirmed)",
             "is_silent": True,
             "self_calibrated": False,
             "evidence": {
-                "note": "No speech detected in audio. Please speak clearly into your microphone.",
+                "note": "No active speech signal detected. Ambient energy below threshold.",
             },
         }
 
-    sev = severity_from_softmax(probs)
+    sev, stut_prob_val, confidence_rating = severity_from_softmax(probs)
     if sev is None:
         stut_bucket, stut_prob_val, stut_probs_disp = None, 0.0, None
         stut_loss = 0.0
     else:
         stut_bucket = cal.shift(sev) if cal.active else sev
         p = np.asarray(probs, dtype=float)
-        stut_prob_val = float(p[1]) if len(p) > 1 else float(np.sum(p[1:]))
         stut_probs_disp = p.tolist()
         
         # Continuous stutter penalty
@@ -143,7 +178,7 @@ def diag_statistics(
         if cal.active and cal.normal_fluent is not None and cal.normal_fluent < 0.60:
             stut_loss = max(0.0, stut_loss - 0.25)
 
-    # Pronunciation penalty
+    # Pronunciation penalty from GOP
     pr = pronunciation.get("pron_score") if isinstance(pronunciation, dict) else None
     if pr is not None:
         pron_loss = 1.0 - float(pr)
@@ -159,13 +194,13 @@ def diag_statistics(
         pron_loss = 0.0
         pron_bucket = None
 
-    # Articulation penalty
-    art_bucket = articulation_severity(articulatory)
+    # Phonation acoustic penalty
+    art_bucket, art_desc = articulation_severity(articulatory)
     art_loss = art_bucket / 3.0 if art_bucket != 4 else 0.0
 
-    # Clinically realistic 0..100 Fluency Index
+    # Composite Fluency Index (0..100)
     if sev is not None and pr is not None:
-        fluency = int(100.0 * max(0.0, 1.0 - (0.40 * stut_loss + 0.45 * pron_loss + 0.15 * art_loss)))
+        fluency = int(100.0 * max(0.0, 1.0 - (w_stut * stut_loss + w_pron * pron_loss + w_art * art_loss)))
     elif sev is not None:
         fluency = int(100.0 * max(0.0, 1.0 - (0.75 * stut_loss + 0.25 * art_loss)))
     elif pr is not None:
@@ -187,14 +222,21 @@ def diag_statistics(
             "overall": bucket_of(overall),
         },
         "fluency_100": fluency,
+        "confidence": confidence_rating,
         "is_silent": False,
         "self_calibrated": cal.active,
         "evidence": {
             "stutter_probs": stut_probs_disp,
-            "stutter_severity": sev,
+            "stutter_severity_raw": sev,
             "pron_goodness": pr,
-            "articulatory": articulatory,
-            "normal_fluent": cal.normal_fluent,
-            "shift_applied": cal.active and cal.normal_fluent is not None and cal.normal_fluent < 0.60,
+            "articulatory_correlates": articulatory,
+            "articulatory_summary": art_desc,
+            "heuristic_weights_applied": {
+                "stutter_weight": w_stut,
+                "pronunciation_weight": w_pron,
+                "acoustics_weight": w_art,
+            },
+            "normal_fluent_baseline": cal.normal_fluent,
+            "calibration_shift_applied": cal.active and cal.normal_fluent is not None and cal.normal_fluent < 0.60,
         },
     }