modules/cif_model.py

"""CONTRASTIVE INTERFERENCE FIELD (CIF) MODEL

A novel mathematical framework that treats L1 and L2 (English) as
competing attractor fields in acoustic feature space.

The child's actual speech production is modeled as a state vector S in
high-dimensional acoustic space. Two attractor points -- A_L1 (L1 target)
and A_L2 (English target) -- exert competing pulls. The Contrastive Interference
Index (CII) measures where S falls between these attractors.

Supports multiple L1 languages: Bhojpuri, Hindi, Bangla, Odia.

Definitions:
    CII(S) = d(S, A_L2) / (d(S, A_L1) + d(S, A_L2))
    where d is weighted Mahalanobis distance.

    CII = 0  -> perfect L2 (English) production
    CII = 1  -> pure L1 transfer
    CII = 0.5 -> equal pull from both languages

Per-dimension decomposition:
    CII_phoneme, CII_prosody, CII_rhythm, CII_voice, CII_fluency

Trajectory prediction:
    CII(t) = CII_0 * exp(-lambda * t) + CII_inf

Author: Neil Shankar Ray, IIT Patna
"""

from __future__ import annotations

import math
from dataclasses import dataclass, field
from typing import Any

import numpy as np
from scipy.optimize import curve_fit


# ── Attractor definitions ────────────────────────────────────────────────
# Each attractor is a dict of feature_name -> (expected_value, weight, natural_std)
# natural_std is used to normalize deviations (Mahalanobis-style)

# Per-L1 attractors: what L1-dominant speech sounds like for each language
L1_ATTRACTORS: dict[str, dict[str, tuple[float, float, float]]] = {
    "bho": {  # Bhojpuri: 6 vowels, strong retroflex, syllable-timed
        "f1_mean": (620, 1.0, 80), "f2_mean": (1350, 1.0, 150),
        "vowel_space_area": (130000, 1.2, 30000), "phoneme_accuracy": (0.45, 1.5, 0.15),
        "interference_score": (70, 1.0, 15),
        "npvi_v": (32, 1.3, 8), "percent_v": (58, 1.0, 5), "prosodic_score": (35, 1.0, 12),
        "speech_rate_syl": (3.2, 0.8, 0.8), "pause_to_speech": (0.35, 0.8, 0.1),
        "delta_c": (25, 1.0, 8),
        "hnr": (11, 0.7, 4), "jitter": (0.025, 0.7, 0.01), "voice_quality_score": (55, 0.8, 12),
        "fluency_score": (25, 1.2, 10), "coarticulation_index": (0.35, 1.0, 0.15),
        "cognitive_load": (60, 0.8, 15),
    },
    "hin": {  # Hindi: 10 vowels, retroflex, better English exposure typically
        "f1_mean": (590, 1.0, 80), "f2_mean": (1400, 1.0, 150),
        "vowel_space_area": (170000, 1.2, 35000), "phoneme_accuracy": (0.55, 1.5, 0.15),
        "interference_score": (55, 1.0, 15),
        "npvi_v": (35, 1.3, 8), "percent_v": (55, 1.0, 5), "prosodic_score": (42, 1.0, 12),
        "speech_rate_syl": (3.5, 0.8, 0.8), "pause_to_speech": (0.30, 0.8, 0.1),
        "delta_c": (30, 1.0, 8),
        "hnr": (12, 0.7, 4), "jitter": (0.022, 0.7, 0.01), "voice_quality_score": (58, 0.8, 12),
        "fluency_score": (32, 1.2, 10), "coarticulation_index": (0.40, 1.0, 0.15),
        "cognitive_load": (52, 0.8, 15),
    },
    "ben": {  # Bangla: 7 vowels, retroflex, strong nasalization
        "f1_mean": (610, 1.0, 80), "f2_mean": (1380, 1.0, 150),
        "vowel_space_area": (145000, 1.2, 30000), "phoneme_accuracy": (0.48, 1.5, 0.15),
        "interference_score": (65, 1.0, 15),
        "npvi_v": (30, 1.3, 8), "percent_v": (57, 1.0, 5), "prosodic_score": (38, 1.0, 12),
        "speech_rate_syl": (3.3, 0.8, 0.8), "pause_to_speech": (0.33, 0.8, 0.1),
        "delta_c": (27, 1.0, 8),
        "hnr": (11, 0.7, 4), "jitter": (0.024, 0.7, 0.01), "voice_quality_score": (56, 0.8, 12),
        "fluency_score": (28, 1.2, 10), "coarticulation_index": (0.37, 1.0, 0.15),
        "cognitive_load": (58, 0.8, 15),
    },
    "ori": {  # Odia: 6 vowels, strong retroflex, similar to Bhojpuri
        "f1_mean": (615, 1.0, 80), "f2_mean": (1340, 1.0, 150),
        "vowel_space_area": (135000, 1.2, 30000), "phoneme_accuracy": (0.47, 1.5, 0.15),
        "interference_score": (68, 1.0, 15),
        "npvi_v": (33, 1.3, 8), "percent_v": (57, 1.0, 5), "prosodic_score": (36, 1.0, 12),
        "speech_rate_syl": (3.2, 0.8, 0.8), "pause_to_speech": (0.34, 0.8, 0.1),
        "delta_c": (26, 1.0, 8),
        "hnr": (11, 0.7, 4), "jitter": (0.024, 0.7, 0.01), "voice_quality_score": (55, 0.8, 12),
        "fluency_score": (26, 1.2, 10), "coarticulation_index": (0.36, 1.0, 0.15),
        "cognitive_load": (59, 0.8, 15),
    },
}

DEFAULT_L1_CODE = "bho"

# English L2 attractor (what native-like English sounds like)
ATTRACTOR_L2: dict[str, tuple[float, float, float]] = {
    # Phoneme features
    "f1_mean":              (480, 1.0, 80),       # wider vowel space
    "f2_mean":              (1650, 1.0, 150),     # broader F2 range
    "vowel_space_area":     (280000, 1.2, 40000), # larger VSA (12+ vowels)
    "phoneme_accuracy":     (0.88, 1.5, 0.08),   # high accuracy
    "interference_score":   (10, 1.0, 8),          # low interference

    # Prosody features
    "npvi_v":               (62, 1.3, 8),          # stress-timed
    "percent_v":            (42, 1.0, 5),          # lower vocalic proportion
    "prosodic_score":       (82, 1.0, 10),         # high nativeness

    # Rhythm features
    "speech_rate_syl":      (5.0, 0.8, 0.8),      # natural rate
    "pause_to_speech":      (0.18, 0.8, 0.06),    # less pausing
    "delta_c":              (55, 1.0, 10),         # high consonantal variability

    # Voice quality features
    "hnr":                  (18, 0.7, 4),          # clear voice
    "jitter":               (0.012, 0.7, 0.005),  # low jitter
    "voice_quality_score":  (80, 0.8, 8),          # good quality

    # Fluency features
    "fluency_score":        (70, 1.2, 12),         # strong connected speech
    "coarticulation_index": (0.75, 1.0, 0.12),    # smooth coarticulation
    "cognitive_load":       (20, 0.8, 10),         # low effort
}


# ── Feature groupings for per-dimension CII ──────────────────────────────

DIMENSION_FEATURES: dict[str, list[str]] = {
    "phoneme": ["f1_mean", "f2_mean", "vowel_space_area", "phoneme_accuracy", "interference_score"],
    "prosody": ["npvi_v", "percent_v", "prosodic_score"],
    "rhythm":  ["speech_rate_syl", "pause_to_speech", "delta_c"],
    "voice":   ["hnr", "jitter", "voice_quality_score"],
    "fluency": ["fluency_score", "coarticulation_index", "cognitive_load"],
}


# ── Result dataclasses ───────────────────────────────────────────────────

@dataclass
class DimensionCII:
    name: str
    cii: float               # 0-1
    severity: str            # "Low", "Mild", "Moderate", "High", "Critical"
    distance_to_l1: float
    distance_to_l2: float
    features_used: list[str]
    bar: str                 # visual bar e.g. "████████░░"


@dataclass
class TrajectoryPrediction:
    cii_initial: float       # CII_0
    cii_residual: float      # CII_inf (asymptotic floor)
    decay_rate: float        # lambda
    predicted_cii_8w: float  # predicted CII at 8 weeks
    weeks_to_moderate: float | None   # weeks to reach CII < 0.5
    weeks_to_mild: float | None       # weeks to reach CII < 0.3
    confidence: str          # "high", "moderate", "low"


@dataclass
class CIFResult:
    overall_cii: float
    overall_severity: str
    dimensions: list[DimensionCII]
    state_vector: dict[str, float]
    trajectory: TrajectoryPrediction | None
    methodology: str


# ── Core math ────────────────────────────────────────────────────────────

def _weighted_mahalanobis(
    state: dict[str, float],
    attractor: dict[str, tuple[float, float, float]],
    features: list[str],
) -> float:
    """Compute weighted Mahalanobis-style distance from state to attractor.

    For each feature:
        contribution = weight * ((state_val - attractor_val) / sigma)^2

    Distance = sqrt(sum of contributions)

    This normalizes each feature by its natural standard deviation (sigma)
    so that all features contribute proportionally regardless of unit scale.
    The weight further controls clinical importance.
    """
    total = 0.0
    n_used = 0

    for feat in features:
        if feat not in state or feat not in attractor:
            continue
        s_val = state[feat]
        a_val, weight, sigma = attractor[feat]

        if sigma <= 0:
            continue

        normalized_diff = (s_val - a_val) / sigma
        total += weight * (normalized_diff ** 2)
        n_used += 1

    if n_used == 0:
        return 0.0

    # Normalize by number of features so dimensions with more features
    # don't dominate the overall CII
    return math.sqrt(total / n_used)


def _compute_cii(d_l1: float, d_l2: float) -> float:
    """CII = d(S, A_L2) / (d(S, A_L1) + d(S, A_L2))

    Returns 0 when speech is at the English attractor,
    1 when at the L1 attractor.
    """
    denominator = d_l1 + d_l2
    if denominator < 1e-10:
        return 0.5  # equidistant / no signal
    return d_l2 / denominator


def _classify_severity(cii: float) -> str:
    if cii >= 0.75:
        return "Critical"
    elif cii >= 0.60:
        return "High"
    elif cii >= 0.45:
        return "Moderate"
    elif cii >= 0.25:
        return "Mild"
    else:
        return "Low"


def _make_bar(cii: float, width: int = 10) -> str:
    """Generate visual bar: ████████░░"""
    filled = max(0, min(width, round(cii * width)))
    empty = width - filled
    return "\u2588" * filled + "\u2591" * empty


# ── Trajectory prediction ────────────────────────────────────────────────

def _exponential_decay(t: np.ndarray, cii_0: float, lam: float, cii_inf: float) -> np.ndarray:
    """CII(t) = CII_0 * exp(-lambda * t) + CII_inf"""
    return cii_0 * np.exp(-lam * t) + cii_inf


def predict_trajectory(
    historical_ciis: list[tuple[float, float]],  # [(week_number, cii_value), ...]
    current_cii: float,
) -> TrajectoryPrediction:
    """Fit exponential decay to historical CII values and predict future.

    If insufficient history, uses population-level defaults for lambda.
    """
    confidence = "low"

    if len(historical_ciis) >= 4:
        # Enough data to fit the curve
        try:
            t_data = np.array([h[0] for h in historical_ciis])
            cii_data = np.array([h[1] for h in historical_ciis])

            # Initial guesses
            p0 = [cii_data[0], 0.05, 0.15]
            bounds = ([0, 0.001, 0], [1.5, 0.5, 0.8])

            popt, pcov = curve_fit(_exponential_decay, t_data, cii_data, p0=p0, bounds=bounds, maxfev=5000)
            cii_0, lam, cii_inf = popt
            confidence = "high" if len(historical_ciis) >= 8 else "moderate"
        except (RuntimeError, ValueError):
            # Fit failed, use defaults
            cii_0 = current_cii
            lam = 0.034  # population average
            cii_inf = 0.12
            confidence = "low"
    elif len(historical_ciis) >= 2:
        # Minimal data — estimate lambda from two-point slope
        t1, c1 = historical_ciis[0]
        t2, c2 = historical_ciis[-1]
        dt = t2 - t1
        if dt > 0 and c1 > c2 and c1 > 0.15:
            lam = -math.log(max(0.01, (c2 - 0.12) / (c1 - 0.12))) / dt
            lam = max(0.005, min(0.3, lam))
        else:
            lam = 0.034
        cii_0 = current_cii
        cii_inf = 0.12
        confidence = "low"
    else:
        # No history — use population defaults
        cii_0 = current_cii
        lam = 0.034
        cii_inf = 0.12
        confidence = "low"

    # Predict at 8 weeks from now
    predicted_8w = cii_0 * math.exp(-lam * 8) + cii_inf

    # Weeks to reach moderate (CII < 0.5)
    weeks_to_moderate = None
    if current_cii >= 0.5 and cii_inf < 0.5:
        target = 0.5 - cii_inf
        if cii_0 > 0 and target > 0 and target < cii_0:
            weeks_to_moderate = round(-math.log(target / cii_0) / lam, 1)

    # Weeks to reach mild (CII < 0.3)
    weeks_to_mild = None
    if current_cii >= 0.3 and cii_inf < 0.3:
        target = 0.3 - cii_inf
        if cii_0 > 0 and target > 0 and target < cii_0:
            weeks_to_mild = round(-math.log(target / cii_0) / lam, 1)

    return TrajectoryPrediction(
        cii_initial=round(cii_0, 4),
        cii_residual=round(cii_inf, 4),
        decay_rate=round(lam, 4),
        predicted_cii_8w=round(max(cii_inf, predicted_8w), 4),
        weeks_to_moderate=weeks_to_moderate,
        weeks_to_mild=weeks_to_mild,
        confidence=confidence,
    )


# ── Feature extraction from pipeline results ─────────────────────────────

def _extract_state_vector(profile: dict[str, Any]) -> dict[str, float]:
    """Extract the acoustic state vector S from pipeline results.

    Maps every relevant score/measurement from the 10-layer pipeline
    into the feature space defined by the attractors.
    """
    state: dict[str, float] = {}

    # ── From phoneme_analysis ──
    pa = profile.get("phoneme_analysis", {})
    state["phoneme_accuracy"] = float(pa.get("overall_accuracy", 0))
    state["interference_score"] = float(pa.get("interference_score", 50))
    state["vowel_space_area"] = float(pa.get("vowel_space_area", 0))

    formant_means = pa.get("formant_means", {})
    state["f1_mean"] = float(formant_means.get("f1", 0))
    state["f2_mean"] = float(formant_means.get("f2", 0))

    # If phoneme_analysis formants are zero, try feature_extraction
    if state["f1_mean"] == 0:
        fe = profile.get("feature_extraction", {})
        praat = fe.get("parselmouth", {})
        formants = praat.get("formants", {})
        state["f1_mean"] = float(formants.get("f1_mean", 0))
        state["f2_mean"] = float(formants.get("f2_mean", 0))
        if state["vowel_space_area"] == 0:
            state["vowel_space_area"] = float(formants.get("vowel_space_area", 0))

    # ── From prosodic_profile ──
    pp = profile.get("prosodic_profile", {})
    rhythm = pp.get("rhythm", {})
    state["npvi_v"] = float(rhythm.get("npvi_v", 0))
    state["percent_v"] = float(rhythm.get("percent_v", 50))
    state["delta_c"] = float(rhythm.get("delta_c", 0))
    state["prosodic_score"] = float(pp.get("prosodic_score", 50))
    state["speech_rate_syl"] = float(pp.get("speech_rate_syl_per_sec", 0))
    state["pause_to_speech"] = float(pp.get("pause_to_speech_ratio", 0))

    # ── From voice_quality ──
    vq = profile.get("voice_quality", {})
    breathiness = vq.get("breathiness", {})
    creakiness = vq.get("creakiness", {})
    state["hnr"] = float(breathiness.get("hnr", 0))
    state["jitter"] = float(creakiness.get("jitter_local", 0))
    state["voice_quality_score"] = float(vq.get("overall_quality_score", 50))

    # ── From connected_speech ──
    cs = profile.get("connected_speech", {})
    state["fluency_score"] = float(cs.get("fluency_score", 0))
    state["coarticulation_index"] = float(cs.get("coarticulation_index", 0))

    # ── From morpheme_boundary ──
    mb = profile.get("morpheme_boundary", {})
    cog = mb.get("cognitive_load", {})
    state["cognitive_load"] = float(cog.get("score", 50))

    return state


# ── Main entry point ─────────────────────────────────────────────────────

def compute_cif(
    profile: dict[str, Any],
    l1_code: str = DEFAULT_L1_CODE,
    historical_ciis: list[tuple[float, float]] | None = None,
) -> dict[str, Any]:
    """Compute the Contrastive Interference Field analysis.

    Args:
        profile: Full pipeline result dict from _run_full_pipeline
        l1_code: L1 language code ("bho", "hin", "ben", "ori")
        historical_ciis: Optional list of (week, cii) tuples for trajectory fitting

    Returns:
        Dict suitable for JSON serialization containing full CIF analysis.
    """
    state = _extract_state_vector(profile)
    attractor_l1 = L1_ATTRACTORS.get(l1_code, L1_ATTRACTORS[DEFAULT_L1_CODE])

    from modules.l1_targets import get_l1_profile
    l1_profile = get_l1_profile(l1_code)
    l1_display_name = l1_profile.display_name

    # ── Per-dimension CII ──
    dimensions: list[DimensionCII] = []
    dimension_ciis: list[float] = []
    dimension_weights: list[float] = []

    for dim_name, features in DIMENSION_FEATURES.items():
        d_l1 = _weighted_mahalanobis(state, attractor_l1, features)
        d_l2 = _weighted_mahalanobis(state, ATTRACTOR_L2, features)
        cii = _compute_cii(d_l1, d_l2)

        # Weight dimensions by their clinical importance for overall CII
        # Phoneme and rhythm deviations are the most perceptually salient
        dim_weight = {
            "phoneme": 1.5,
            "prosody": 1.3,
            "rhythm":  1.2,
            "voice":   0.7,
            "fluency": 1.3,
        }.get(dim_name, 1.0)

        dimensions.append(DimensionCII(
            name=dim_name,
            cii=round(cii, 4),
            severity=_classify_severity(cii),
            distance_to_l1=round(d_l1, 4),
            distance_to_l2=round(d_l2, 4),
            features_used=features,
            bar=_make_bar(cii),
        ))
        dimension_ciis.append(cii)
        dimension_weights.append(dim_weight)

    # ── Overall CII (weighted average of dimension CIIs) ──
    w_arr = np.array(dimension_weights)
    c_arr = np.array(dimension_ciis)
    overall_cii = float(np.average(c_arr, weights=w_arr))

    # ── Trajectory prediction ──
    trajectory = None
    if historical_ciis is not None:
        trajectory = predict_trajectory(historical_ciis, overall_cii)
    else:
        # Still provide a default prediction using population lambda
        trajectory = predict_trajectory([], overall_cii)

    result = CIFResult(
        overall_cii=round(overall_cii, 4),
        overall_severity=_classify_severity(overall_cii),
        dimensions=dimensions,
        state_vector={k: round(v, 4) for k, v in state.items()},
        trajectory=trajectory,
        methodology=(
            f"Contrastive Interference Field (CIF) Model v2.0 -- "
            f"Weighted Mahalanobis distance from L1 ({l1_display_name}) and L2 (English) "
            f"attractor points in 17-dimensional acoustic feature space. "
            f"CII = d(S,A_L2) / (d(S,A_L1) + d(S,A_L2)). "
            f"Trajectory: CII(t) = CII_0 * exp(-lambda*t) + CII_inf."
        ),
    )

    # Serialize to dict
    return _serialize_cif(result)


def _serialize_cif(result: CIFResult) -> dict[str, Any]:
    """Convert CIFResult to a JSON-serializable dict."""
    return {
        "overall_cii": result.overall_cii,
        "overall_severity": result.overall_severity,
        "dimensions": [
            {
                "name": d.name.capitalize(),
                "cii": d.cii,
                "severity": d.severity,
                "distance_to_l1": d.distance_to_l1,
                "distance_to_l2": d.distance_to_l2,
                "features_used": d.features_used,
                "bar": d.bar,
            }
            for d in result.dimensions
        ],
        "state_vector": result.state_vector,
        "trajectory": {
            "cii_initial": result.trajectory.cii_initial,
            "cii_residual": result.trajectory.cii_residual,
            "decay_rate": result.trajectory.decay_rate,
            "predicted_cii_8w": result.trajectory.predicted_cii_8w,
            "weeks_to_moderate": result.trajectory.weeks_to_moderate,
            "weeks_to_mild": result.trajectory.weeks_to_mild,
            "confidence": result.trajectory.confidence,
        } if result.trajectory else None,
        "methodology": result.methodology,
    }