Upload ml/model/fusion.py with huggingface_hub

ml/model/fusion.py

@@ -1,31 +1,29 @@
 """
-ml/model/fusion.py - Multi-modal fusion + per-user self-calibration + uncertainty
-==================================================================================
+ml/model/fusion.py - Multi-modal fusion + per-user self-calibration + margin estimation
+========================================================================================
 Fuses neural disfluency classification, CTC phonetic alignment, and Praat acoustic
-correlates of phonation into a unified, transparent screening report.
+correlates of phonation into a 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.
-  - Single-Shift Calibration: The user's baseline offset is applied exactly ONCE
-    at the overall decision level to prevent compounding discount errors.
-  - 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.
+  - Heuristic Weighting: The Composite Screening Index (0..100) utilizes transparent
+    heuristic weights (40% disfluency, 45% pronunciation, 15% phonation correlates).
+  - Single-Shift Calibration: The user's baseline offset is applied consistently at the
+    composite level to prevent compounding discount errors.
+  - Prediction Margin: Quantifies distance from the decision boundary (|P - 0.5| * 200%).
+  - Non-Diagnostic Screening: Categorizes into 'typical', 'mild', 'moderate', or 'significant'
+    concern bands rather than definitive clinical severity diagnoses.
 """
 from __future__ import annotations
 from typing import Optional, Dict, Any, Tuple
 
 import numpy as np
 
-BUCKETS = ["fluent", "mild", "moderate", "severe", "silent", "uncertain"]
+CONCERN_BANDS = ["typical", "mild", "moderate", "significant", "silent", "uncertain"]
 
-# Configurable empirical thresholds
+# Empirical screening thresholds
 THRESHOLD_MILD = 0.60
 THRESHOLD_MODERATE = 0.78
-THRESHOLD_SEVERE = 0.90
+THRESHOLD_SIGNIFICANT = 0.90
 BORDERLINE_BAND = (0.42, 0.58)
 
 
@@ -34,59 +32,65 @@ def bucket_of(sev: int) -> str:
         return "silent"
     if sev == 5:
         return "uncertain"
-    return BUCKETS[min(max(sev, 0), 3)]
+    return CONCERN_BANDS[min(max(sev, 0), 3)]
 
 
-def severity_from_softmax(probs) -> Tuple[Optional[int], float, str]:
+def severity_from_softmax(probs) -> Tuple[Optional[int], float, str, float]:
     """
-    0..3 clinical severity from a stutter softmax with uncertainty tracking.
-    Returns: (severity_bucket, p_stutter, confidence_rating)
+    0..3 concern band from a stutter softmax with prediction margin estimation.
+    Returns: (concern_band_index, p_stutter, margin_label, margin_pct)
     """
     if probs is None:
-        return None, 0.0, "unknown"
+        return None, 0.0, "unknown", 0.0
     p = np.asarray(probs, dtype=float)
     if p.sum() > 0:
         p = p / p.sum()
     if len(p) == 0:
-        return None, 0.0, "unknown"
+        return None, 0.0, "unknown", 0.0
     
     p_stutter = float(p[1]) if len(p) == 2 else float(np.sum(p[1:]))
-    
-    # Uncertainty / Confidence estimation
-    margin = abs(p_stutter - 0.50) * 2.0  # 0.0 at 50/50 tossup, 1.0 at pure certainty
+    margin_pct = round(abs(p_stutter - 0.50) * 200.0, 1)  # 0% at 0.50 tossup, 100% at 0.0 or 1.0
+
     if BORDERLINE_BAND[0] <= p_stutter <= BORDERLINE_BAND[1]:
-        confidence = "low (borderline transition)"
-    elif margin < 0.35:
-        confidence = "moderate"
+        margin_label = f"Borderline Margin ({margin_pct:.0f}%)"
+    elif margin_pct < 40.0:
+        margin_label = f"Moderate Margin ({margin_pct:.0f}%)"
     else:
-        confidence = "high"
+        margin_label = f"High Margin ({margin_pct:.0f}%)"
 
     if p_stutter < THRESHOLD_MILD:
-        return 0, p_stutter, confidence
+        return 0, p_stutter, margin_label, margin_pct
     elif p_stutter < THRESHOLD_MODERATE:
-        return 1, p_stutter, confidence
-    elif p_stutter < THRESHOLD_SEVERE:
-        return 2, p_stutter, confidence
+        return 1, p_stutter, margin_label, margin_pct
+    elif p_stutter < THRESHOLD_SIGNIFICANT:
+        return 2, p_stutter, margin_label, margin_pct
     else:
-        return 3, p_stutter, confidence
+        return 3, p_stutter, margin_label, margin_pct
 
 
 def articulation_severity(articulatory: dict) -> Tuple[int, str]:
     """
-    0..3 roughness from acoustic correlates of phonation (Praat PointProcess).
-    Returns: (severity_index, summary_string)
+    0..3 roughness index from acoustic correlates of phonation (Praat PointProcess).
+    Returns: (concern_index, summary_string)
     """
     if articulatory.get("is_silent"):
         return 4, "silent"
+    if not articulatory.get("is_valid", True):
+        return 0, "Acoustic features unmeasured"
+
     s = 0
     flags = []
-    if articulatory.get("hnr_db", 20) < 9.0:
+    hnr = articulatory.get("hnr_db")
+    jitter = articulatory.get("jitter")
+    voiced_ratio = articulatory.get("voiced_ratio")
+
+    if hnr is not None and hnr < 9.0:
         s += 1
-        flags.append("Low HNR (<9 dB)")
-    if articulatory.get("jitter", 0.0) > 0.05:
+        flags.append(f"Low HNR ({hnr:.1f} dB)")
+    if jitter is not None and jitter > 0.05:
         s += 1
-        flags.append("High Jitter (>5%)")
-    if articulatory.get("voiced_ratio", 1.0) < 0.25:
+        flags.append(f"High Jitter ({jitter*100:.1f}%)")
+    if voiced_ratio is not None and voiced_ratio < 0.25:
         s += 1
         flags.append("Low Voicing (<25%)")
     
@@ -98,7 +102,7 @@ def articulation_severity(articulatory: dict) -> Tuple[int, str]:
 # Per-user self-calibration (Single Shift Guarded)
 # ---------------------------------------------------------------------------
 class CalibrationProfile:
-    """Per-user bucket-offset derived from their own healthy baseline recording."""
+    """Per-user baseline offset derived from their own healthy baseline recording."""
 
     def __init__(self, normal_fluent: Optional[float] = None):
         self.normal_fluent = normal_fluent
@@ -108,10 +112,10 @@ class CalibrationProfile:
         return self.normal_fluent is not None
 
     def shift(self, raw: int) -> int:
-        """Single bucket offset so the user's natural baseline == fluent."""
+        """Single bucket offset so the user's natural baseline == typical."""
         if not self.active or raw in (4, 5):
             return raw
-        offset = 1 if self.normal_fluent < 0.60 else 0
+        offset = 1 if (self.normal_fluent is not None and self.normal_fluent < 0.60) else 0
         return max(0, raw - offset)
 
 
@@ -136,8 +140,8 @@ def diag_statistics(
     weights: Tuple[float, float, float] = (0.40, 0.45, 0.15),
 ) -> dict:
     """
-    Fuse all modalities into an auditable diagnostic screening dictionary.
-    Single-shift calibration applied strictly at the composite level.
+    Fuse all modalities into an auditable screening assessment dictionary.
+    Single-shift calibration applied consistently at the composite level.
     """
     cal = calibration or CalibrationProfile()
     w_stut, w_pron, w_art = weights
@@ -158,7 +162,8 @@ def diag_statistics(
                 "overall": "silent",
             },
             "fluency_100": None,
-            "confidence": "high (silence confirmed)",
+            "confidence": "High (Silence Confirmed)",
+            "prediction_margin": 100.0,
             "is_silent": True,
             "self_calibrated": False,
             "evidence": {
@@ -166,30 +171,29 @@ def diag_statistics(
             },
         }
 
-    # 2. Raw Subsystem Severities (Unshifted at modality level)
-    sev, stut_prob_val, confidence_rating = severity_from_softmax(probs)
+    # 2. Raw Subsystem Concern Bands
+    sev, stut_prob_val, margin_label, margin_pct = 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 = sev  # Raw unshifted severity
+        stut_bucket = sev
         p = np.asarray(probs, dtype=float)
         stut_probs_disp = p.tolist()
-        # Continuous stutter penalty
         stut_loss = max(0.0, (stut_prob_val - 0.45) / 0.55)
 
-    # Pronunciation penalty from GOP
+    # Pronunciation penalty from phonetic alignment
     pr = pronunciation.get("pron_score") if isinstance(pronunciation, dict) else None
     if pr is not None:
         pron_loss = 1.0 - float(pr)
         if pron_loss < 0.15:
-            pron_bucket = 0  # fluent
+            pron_bucket = 0  # typical
         elif pron_loss < 0.40:
             pron_bucket = 1  # mild
         elif pron_loss < 0.70:
             pron_bucket = 2  # moderate
         else:
-            pron_bucket = 3  # severe
+            pron_bucket = 3  # significant
     else:
         pron_loss = 0.0
         pron_bucket = None
@@ -198,19 +202,23 @@ def diag_statistics(
     art_bucket, art_desc = articulation_severity(articulatory)
     art_loss = art_bucket / 3.0 if art_bucket != 4 else 0.0
 
-    # Composite Fluency Index (0..100)
+    # Composite Screening Index (0..100)
     if sev is not None and pr is not None:
-        fluency = int(100.0 * max(0.0, 1.0 - (w_stut * stut_loss + w_pron * pron_loss + w_art * art_loss)))
+        composite_score = 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)))
+        composite_score = int(100.0 * max(0.0, 1.0 - (0.75 * stut_loss + 0.25 * art_loss)))
     elif pr is not None:
-        fluency = int(100.0 * max(0.0, 1.0 - (0.75 * pron_loss + 0.25 * art_loss)))
+        composite_score = int(100.0 * max(0.0, 1.0 - (0.75 * pron_loss + 0.25 * art_loss)))
     else:
-        fluency = int(100.0 * max(0.0, 1.0 - art_loss))
+        composite_score = int(100.0 * max(0.0, 1.0 - art_loss))
     
-    fluency = max(0, min(100, fluency))
+    # Adjust score if calibrated
+    if cal.active and cal.normal_fluent is not None and cal.normal_fluent < 0.60:
+        composite_score = min(100, composite_score + 10)
+
+    composite_score = max(0, min(100, composite_score))
 
-    # Single shift applied exactly once at the overall rating
+    # Single shift applied consistently at the composite overall rating
     present = [b for b in (stut_bucket, pron_bucket, art_bucket) if b is not None and b != 4]
     overall_raw = max(present) if present else 0
     overall = cal.shift(overall_raw) if cal.active else overall_raw
@@ -222,16 +230,17 @@ def diag_statistics(
             "articulation": bucket_of(art_bucket),
             "overall": bucket_of(overall),
         },
-        "fluency_100": fluency,
-        "confidence": confidence_rating,
+        "fluency_100": composite_score,
+        "confidence": margin_label,
+        "prediction_margin": margin_pct,
         "is_silent": False,
         "self_calibrated": cal.active,
         "evidence": {
             "stutter_probs": stut_probs_disp,
-            "stutter_severity_raw": sev,
-            "overall_severity_raw": overall_raw,
-            "overall_severity_calibrated": overall,
-            "pron_goodness": pr,
+            "stutter_concern_raw": sev,
+            "overall_concern_raw": overall_raw,
+            "overall_concern_calibrated": overall,
+            "pron_score": pr,
             "articulatory_correlates": articulatory,
             "articulatory_summary": art_desc,
             "heuristic_weights_applied": {