""" Speech Analysis — AI Inference Score Engine Pure Wav2Vec2 pipeline: no reference audio files needed. Receives a child's recording + target word, runs phoneme extraction via forced alignment, then diagnoses pronunciation quality using acoustic feature thresholds. Pipeline: 1. phoneme_extractor.extract_shin() → isolate the ש segment 2. extract_phoneme_features() → pitch-agnostic acoustic features 3. diagnose_phoneme() → rule-based expert diagnosis """ import os import json import argparse import numpy as np import librosa from scipy.signal import butter, sosfilt from scipy.stats import skew # ============================================================================= # Configuration # ============================================================================= # Scoring backend toggle: # "stt" (default) — use the STT "pocket" (stt_judge.py) for basic ש/ס & ק/ת # detection, falling back to the math pipeline when STT # can't decide. This is the MVP / data-collection mode. # "math" — skip STT entirely; use only the Wav2Vec2 + acoustic # pipeline (the long-term, nuance-capable system). # The math pipeline is ALWAYS preserved; this switch only chooses what runs. SCORING_BACKEND = os.environ.get("SCORING_BACKEND", "stt").strip().lower() # Frequency bands SH_BAND_LOW = 2000 SH_BAND_HIGH = 5000 S_BAND_LOW = 6000 S_BAND_HIGH = 10000 FRICATIVE_LOW_HZ = 3000 FRICATIVE_HIGH_HZ = 8000 LATERAL_SH_LOW = 1000 LATERAL_SH_HIGH = 3000 # ============================================================================= # Audio Processing Helpers # ============================================================================= def apply_bandpass_filter(audio: np.ndarray, sr: int, low_hz: int, high_hz: int) -> np.ndarray: """Apply a bandpass Butterworth filter.""" nyquist = sr / 2 low_normalized = max(0.01, min(low_hz / nyquist, 0.99)) high_normalized = max(low_normalized + 0.01, min(high_hz / nyquist, 0.99)) sos = butter(N=4, Wn=[low_normalized, high_normalized], btype='bandpass', output='sos') return sosfilt(sos, audio) def calculate_band_energy(audio: np.ndarray, sr: int, low_hz: int, high_hz: int) -> float: """Calculate total energy in a specific frequency band.""" stft = np.abs(librosa.stft(audio)) freqs = librosa.fft_frequencies(sr=sr) band_mask = (freqs >= low_hz) & (freqs <= high_hz) return np.sum(stft[band_mask, :] ** 2) def calculate_s_band_ratio(audio: np.ndarray, sr: int) -> float: """Calculate fraction of energy in S-band (6-10 kHz) vs Sh-band.""" sh_energy = calculate_band_energy(audio, sr, SH_BAND_LOW, SH_BAND_HIGH) s_energy = calculate_band_energy(audio, sr, S_BAND_LOW, S_BAND_HIGH) return s_energy / (sh_energy + s_energy + 1e-10) def calculate_lateral_energy_ratio(audio: np.ndarray, sr: int) -> float: """Calculate fraction of energy in lateral band (1-3 kHz).""" lateral_energy = calculate_band_energy(audio, sr, LATERAL_SH_LOW, LATERAL_SH_HIGH) total_energy = calculate_band_energy(audio, sr, 1000, 10000) return lateral_energy / (total_energy + 1e-10) # ============================================================================= # Phoneme-Level Feature Extraction (Phase 2) # ============================================================================= def extract_phoneme_features(audio: np.ndarray, sr: int) -> dict: """ Compute pitch-agnostic acoustic features on an isolated phoneme snippet. Designed to run on the raw ש audio returned by extract_shin(). Args: audio: 1-D numpy array (raw phoneme audio, expected 16 kHz). sr: Sample rate. Returns: Dict of acoustic features for diagnosis. """ # --- Bandpassed audio (3-8 kHz) --- audio_bp = apply_bandpass_filter(audio, sr, FRICATIVE_LOW_HZ, FRICATIVE_HIGH_HZ) if len(audio_bp) < 2048: audio_bp = np.pad(audio_bp, (0, 2048 - len(audio_bp))) centroid = librosa.feature.spectral_centroid(y=audio_bp, sr=sr)[0] centroid_mean = float(np.mean(centroid)) centroid_median = float(np.median(centroid)) bandwidth = librosa.feature.spectral_bandwidth(y=audio_bp, sr=sr)[0] bandwidth_mean = float(np.mean(bandwidth)) S = np.abs(librosa.stft(audio_bp)) frame_skewness = skew(S, axis=0) spectral_skewness = float(np.nan_to_num( np.mean(np.nan_to_num(frame_skewness, nan=0.0)), nan=0.0 )) # Energy ratio: high band (4-8 kHz) vs mid band (2-4 kHz) high_energy = calculate_band_energy(audio, sr, 4000, 8000) mid_energy = calculate_band_energy(audio, sr, 2000, 4000) high_mid_ratio = float(high_energy / (mid_energy + 1e-10)) s_band_ratio = calculate_s_band_ratio(audio_bp, sr) # --- Pre-bandpass features --- lateral_ratio = calculate_lateral_energy_ratio(audio, sr) sub3k_energy = calculate_band_energy(audio, sr, 500, 3000) broad_energy = calculate_band_energy(audio, sr, 500, 10000) sub3k_ratio = float(sub3k_energy / (broad_energy + 1e-10)) return { "centroid_mean": float(round(centroid_mean, 1)), "centroid_median": float(round(centroid_median, 1)), "bandwidth_mean": float(round(bandwidth_mean, 1)), "spectral_skewness": float(round(spectral_skewness, 4)), "high_mid_ratio": float(round(high_mid_ratio, 4)), "lateral_ratio": float(round(float(lateral_ratio), 4)), "s_band_ratio": float(round(float(s_band_ratio), 4)), "sub3k_ratio": float(round(sub3k_ratio, 4)), } # ============================================================================= # Binary Diagnosis — Goldilocks Zone (Phase 3) # ============================================================================= # AI gate — Wav2Vec2 alignment confidence. Catches lisps and substitutions on # full words because the language model has word-level context to lock onto. # Bypassed in _score_shin_word for sh_sound (isolated CTC scores are unreliable). GOLDILOCKS_AI_SCORE_MIN = 0.70 # Shin-phoneme thresholds — applied to all ש words and sh_sound alike. # Re-calibrated on 20+ production recordings (multiple shin words, varied speakers): # - Centroid 2700-6500 Hz: lower catches wet/lateral, upper catches S. # - s_band_ratio < 45%: catches S substitution (S energy concentrates 6-10kHz). # Raised from 0.40 because correct ש in words with following plosive can hit # 40.1% (e.g. nachash), while S substitutions start at 60%+ — clean buffer. # - sub3k_ratio gate REMOVED as a hard gate (kept as info-only in logs). # Reason: the 120ms extraction window after ש_onset catches articulatory # transition (breath, tongue movement) when ש is followed by another phoneme. # Of 8 production recordings with correctly-pronounced ש, only 1 (dvash, with # ש at absolute word-end) passed sub3k < 12%; the rest scored 29-99% on a # gate calibrated against simulator-quiet audio. Strict sub3k gate caused # ~80% false-negative rate on real-world recordings. # Trade-off: lose detection of subtle wet lisp where centroid + s_band stay normal. # Clear wet/lateral cases (centroid < 2700 or s_band > 45%) are still caught. SHIN_CENTROID_LOW = 2700 SHIN_CENTROID_HIGH = 6500 SHIN_S_BAND_MAX = 0.45 # Isolated ש (stage 1, sh_sound) gets a more lenient s_band ceiling. A clean but # emphatic/sustained ש hiss legitimately pushes s_band higher than a mid-word ש # (51-66% observed on correct isolated productions, incl. "sha"), and stage 1 is # the foundational, confidence-building level where failing a correct production # is the worst outcome. A clear ס substitution sits well above this (a sustained # ס runs ~80%+). Trade-off: a mild/borderline ס may pass in stage 1. SH_SOUND_S_BAND_MAX = 0.72 # NOTE: s_band gate (energy in 6-10kHz) was removed because correct SH has nearly # all energy in 2-5kHz, NOT 6-10kHz. The gate measured S-like energy, not fricative # presence — correct SH scored 0.5% and always failed. # Omission is already caught by AI score (no SH = low confidence) and centroid # (silence/vowel has centroid far below 3000 Hz). # s_band gate REMOVED for isolated SH — does not generalize across devices. # Simulator recordings: correct SH = 0.7%, wet CH = 22%, S = 29% (looked great). # Browser recordings: correct SH = 58% (!!!) — browser noise suppression and AGC # boost high frequencies, completely distorting the s_band ratio. # Same lesson as full words: energy ratio gates are device-dependent. # The frontend MUST disable browser audio processing for accurate spectral analysis: # getUserMedia({ audio: { echoCancellation:false, noiseSuppression:false, autoGainControl:false } }) # Segment extraction settings SEGMENT_EXTRACT_DURATION = 0.12 # extract 120ms FORWARD from Wav2Vec2 onset # Syllable pipeline uses a longer window because wetness/CH artifacts in # children's productions often manifest in the transition out of ש into # the following vowel, not in the first 120ms of the fricative itself. # A 200ms window captures both the fricative core and the transition. SYLLABLE_SEGMENT_EXTRACT_DURATION = 0.20 # ש vs ס competitive discrimination (model-driven, replaces the s_band heuristic). # The Hebrew Wav2Vec2 model natively separates ש (/ʃ/) from ס (/s/) — they are # contrastive phonemes (שיר vs סיר). We ask it directly via frame-level posteriors # (phoneme_extractor.shin_vs_samekh) instead of the brittle energy-ratio gate that # mislabeled emphatic-but-correct ש as ס. # # is_samekh : samekh posterior strictly beats shin posterior → ס substitution # PRESENCE : max(shin, samekh) must clear this floor, else the clip has no # clear sibilant at all (vowel / ח / omission / noise) → reject. # The decision is fundamentally DIRECTIONAL (is the sibilant ש or ס?), not about # absolute magnitude. A sustained isolated "שששש" produces low absolute Wav2Vec # posteriors (CTC spreads probability across many frames) — a perfectly clean ש # measured only 0.005-0.03 yet beat ס by 3-6x. The old 0.02 floor wrongly # rejected those as "absent". The floor is now tiny: it only rejects genuine # no-sibilant clips, while real silence is already caught by the global signal # gate upstream. Lenient toward passing a real ש (pedagogy: never fail a correct # production); a clear ס still wins decisively (samekh posterior ~0.20). SHIN_SAMEKH_PRESENCE_FLOOR = 0.003 SHIN_SAMEKH_MARGIN = 0.0 # samekh must beat shin by more than this to call it ס # Low-confidence band guard: background noise can squeak past the presence # floor (observed: noise strength 0.0030 with a coin-flip gap of +0.0019 → # wrongly PASSed). Real weak-but-clean ש always shows a DECISIVE direction # (observed gaps +0.0036…+0.20). So when strength is below the confident level, # ש must also win by at least this absolute gap — otherwise the clip is treated # as no-sibilant. Above the confident level, direction alone decides (as before). SHIN_SAMEKH_CONFIDENT_STRENGTH = 0.02 SHIN_SAMEKH_LOWBAND_MIN_GAP = 0.0025 # Binary feedback FEEDBACK_CORRECT = "מצוין! הש׳ נשמעת ברורה ונכונה." FEEDBACK_INCORRECT = ( "הש׳ לא נשמעה ברורה. נסה לעגל את השפתיים, לשמור את הלשון באמצע " "ולנשוף אוויר בעדינות. הקשב לדוגמה ונסה שוב." ) # Specific feedback when the model heard ס instead of ש. FEEDBACK_SAMEKH_SUB = ( "נשמע כמו ס׳. כדי לומר ש׳, הרם/י מעט את הלשון לאחור ועגל/י את השפתיים. " "הקשב/י לדוגמה ונסה/י שוב." ) def diagnose_phoneme(features: dict, alignment_score: float, duration: float, used_fallback: bool = False, shin_audio: np.ndarray = None, sr: int = 16000, word: str = None) -> dict: """ Data-driven binary diagnosis for ש pronunciation. Three gates — ALL must pass: 1. AI score ≥ 0.70 — Wav2Vec2 confidence (bypassed for sh_sound) 2. Centroid 2700-6500 Hz — correct spectral shape 3. s_band_ratio < 45% — catches S substitution sub3k_ratio is logged for diagnostics but is NO LONGER a gate — see comment near SHIN_S_BAND_MAX for the calibration history. Args: features: Dict from extract_phoneme_features(). alignment_score: Wav2Vec2 forced-alignment confidence for ש. duration: Duration of the ש segment in seconds. used_fallback: Whether the segment came from fallback extraction. shin_audio: Raw audio of the extracted segment (unused, kept for API compat). sr: Sample rate of shin_audio. Returns: Dict with diagnosis (CORRECT/INCORRECT), feedback, and evidence. """ centroid = features["centroid_mean"] s_band = features["s_band_ratio"] sub3k = features["sub3k_ratio"] # 3 gates for all shin words (isolated and full-word). The caller bypasses # the AI gate for sh_sound by setting alignment_score=1.0 because CTC # alignment of repeated single characters gives near-zero scores. is_valid_ai = alignment_score >= GOLDILOCKS_AI_SCORE_MIN is_valid_pitch = SHIN_CENTROID_LOW <= centroid <= SHIN_CENTROID_HIGH # Lenient s_band ceiling for the isolated stage-1 sound (see SH_SOUND_S_BAND_MAX). s_band_max = SH_SOUND_S_BAND_MAX if word == "sh_sound" else SHIN_S_BAND_MAX is_valid_s_band = s_band < s_band_max is_pass = is_valid_ai and is_valid_pitch and is_valid_s_band evidence = { "centroid_mean": float(centroid), "s_band_ratio": float(s_band), "sub3k_ratio": float(sub3k), # logged for diagnostics, not a gate "alignment_score": float(alignment_score), "is_valid_ai": bool(is_valid_ai), "is_valid_pitch": bool(is_valid_pitch), "is_valid_s_band": bool(is_valid_s_band), } if is_pass: return { "diagnosis": "CORRECT", "feedback": FEEDBACK_CORRECT, "evidence": evidence, } return { "diagnosis": "INCORRECT", "feedback": FEEDBACK_INCORRECT, "evidence": evidence, } # ============================================================================= # Segment Extraction # ============================================================================= def _extract_shin_segment(audio: np.ndarray, sr: int, hint_start: float, hint_end: float, hint_score: float, duration_sec: float = None) -> tuple: """ Extract a segment of audio starting FORWARD from the Wav2Vec2 onset hint. Wav2Vec2 reliably identifies the onset of the 'SH' sound. We simply take hint_start as the beginning and grab `duration_sec` forward. No centering, no energy scanning, no backward look into pre-speech silence. Only falls back to a full-recording energy scan if Wav2Vec2 completely failed (no hint / score essentially zero). Args: audio: Full recording audio (1-D numpy array). sr: Sample rate. hint_start: Wav2Vec2 start boundary (seconds). hint_end: Wav2Vec2 end boundary (seconds). hint_score: Wav2Vec2 alignment confidence for the ש segment (0-1). duration_sec: How many seconds to extract forward from the hint. Defaults to SEGMENT_EXTRACT_DURATION (120ms) used by full-word and isolated-sound pipelines. The syllable pipeline passes a longer value (200ms) to capture wetness/CH artifacts in the transition. Returns: (segment_audio, used_fallback): Extracted audio and whether fallback was used. """ if duration_sec is None: duration_sec = SEGMENT_EXTRACT_DURATION extract_samples = int(duration_sec * sr) # ----------------------------------------------------------------- # Primary path: extract FORWARD from Wav2Vec2 onset # ----------------------------------------------------------------- if hint_score > 0.01: start_sample = max(0, int(hint_start * sr)) end_sample = min(len(audio), start_sample + extract_samples) segment = audio[start_sample:end_sample] start_sec = start_sample / sr end_sec = end_sample / sr print(f"[SEGMENT] Forward extraction from Wav2Vec2 onset. " f"Hint: {hint_start:.3f}-{hint_end:.3f}s (score={hint_score:.4f}). " f"Extracting {start_sec:.3f}-{end_sec:.3f}s ({(end_sample-start_sample)/sr*1000:.0f}ms).") return segment, False # ----------------------------------------------------------------- # Fallback: Wav2Vec2 completely failed — scan full recording for # the loudest 3-8kHz frame and extract 120ms forward from there. # ----------------------------------------------------------------- print(f"[SEGMENT] Wav2Vec2 failed (score={hint_score:.4f}). " f"Falling back to full-recording energy scan.") if len(audio) < 2048: return audio, True hop_length = 512 stft = np.abs(librosa.stft(audio, hop_length=hop_length)) freqs = librosa.fft_frequencies(sr=sr) fric_mask = (freqs >= FRICATIVE_LOW_HZ) & (freqs <= FRICATIVE_HIGH_HZ) fric_energy_per_frame = np.sum(stft[fric_mask, :] ** 2, axis=0) peak_frame = int(np.argmax(fric_energy_per_frame)) peak_sample = peak_frame * hop_length start_sample = max(0, peak_sample) end_sample = min(len(audio), start_sample + extract_samples) segment = audio[start_sample:end_sample] print(f"[SEGMENT] Energy peak at {peak_sample/sr:.3f}s. " f"Extracting {start_sample/sr:.3f}-{end_sample/sr:.3f}s ({(end_sample-start_sample)/sr*1000:.0f}ms).") return segment, True # ============================================================================= # Main Scoring API # ============================================================================= # ============================================================================= # Feedback strings for K/T competitive alignment # ============================================================================= FEEDBACK_KUF_CORRECT = "מצוין! הק׳ נשמעת ברורה ונכונה." FEEDBACK_KUF_INCORRECT = ( "הק׳ לא נשמעה ברורה. נסה להוציא את הצליל מעמוק בגרון, " "ולא מקצה הלשון או מהשפתיים. הקשב לדוגמה ונסה שוב." ) FEEDBACK_TAV_CORRECT = "מצוין! הת׳ נשמעת ברורה ונכונה." FEEDBACK_TAV_INCORRECT = ( "נשמע שאמרת ק׳ במקום ת׳. נסה להוציא את הצליל מקצה הלשון, " "לא מעמוק בגרון. הקשב לדוגמה ונסה שוב." ) # Burst spectral centroid threshold separating velar ק from alveolar ת. # Raised 2500 -> 3400 for the CHILD target population. Children have shorter # vocal tracts, so all their burst frequencies sit ~25% higher than an adult's: # a child's correct velar ק burst reaches 3000-3500 Hz (verified on a new child # whose correct ק scored 3008-3546 and was wrongly flagged as ת at the old 2500 # threshold). A child's alveolar ת sits higher still (~4000+ Hz), so 3400 keeps # them separable. NOTE: calibrated on limited child data — the robust long-term # fix is multi-child calibration from live usage. Raising this also makes # t_sound stricter (an adult ת < 3400 would read as ק), an acceptable trade # since the product targets children and ק practice is primary. PLOSIVE_CENTROID_THRESHOLD_HZ = 3400 # ============================================================================= # Silence/no-signal gate — runs before every scoring pipeline # ============================================================================= # Without this gate, a recording with no actual speech (pure silence + room # noise) still produces a score: the silence-trim helper forces a 0.5s minimum # slice, then Wav2Vec2 alignment fails, the energy-scan fallback picks the # loudest noise burst, and that noise burst can pass the spectral gates by # accident — producing a spurious "CORRECT" diagnosis on absolute silence. # Real speech amplitudes (even quiet child speech) are 0.05+. Room-noise # floor on consumer mics is ~0.005-0.015. Threshold of 0.02 sits in the # clear gap between them. SIGNAL_MIN_PEAK_AMPLITUDE = 0.02 SIGNAL_MIN_RMS = 0.003 FEEDBACK_NO_SIGNAL = "לא נשמע צליל ברור. נסה לדבר חזק יותר ובברור." # ============================================================================= # CV-Syllable Thresholds (e.g. שא, קה) — strict binary gates # ============================================================================= # Syllables (2-char transcripts) cannot use the full-word pipelines: # CTC alignment is unreliable on short transcripts, and the 120ms extraction # window contains both consonant AND vowel — so gates that assume a pure # fricative/plosive snippet (sub3k, AI score) fail systematically. # # Calibration (8 SH + 8 K recordings, half correct / half substitution): # - Correct SH: centroid 4009-5139 Hz, s_band 8-32% # - S sub: centroid 4649-5139 Hz, s_band 40-88% → s_band separates cleanly # - Correct K: burst centroid 1500-2500 Hz (per isolated-plosive physics) # - T sub: burst centroid 2500-5000 Hz # Thresholds chosen strict: prefer false-negative over false-positive (per pedagogy). SYLLABLE_SHIN_CENTROID_LOW = 2700 SYLLABLE_SHIN_CENTROID_HIGH = 5500 SYLLABLE_SHIN_S_BAND_MAX = 0.45 # matched to full-word threshold (see calibration above) # Wetness / lateral-lisp detector (WavLM one-class OOD). Wired ONLY into the # ש-syllable pipeline. The model is an optional pickle built offline by # wetness_detector.py from clean-ש recordings. Until that file exists AND has # calibrated thresholds, the gate is inert and the syllable pipeline behaves # exactly as before. See wetness_detector.py for the method. WETNESS_MODEL_PATH = os.path.join( os.path.dirname(os.path.abspath(__file__)), "wetness_sh_syllable.pkl" ) _wetness_model_cache = "UNLOADED" # sentinel distinct from None (= "tried, absent") def _get_wetness_model(): """ Load the wetness model once and cache it. Returns None if no model file exists (caller then skips wetness scoring — pipeline unchanged). """ global _wetness_model_cache if _wetness_model_cache != "UNLOADED": return _wetness_model_cache if not os.path.exists(WETNESS_MODEL_PATH): print(f"[WETNESS] No model at {WETNESS_MODEL_PATH} — wetness gate inactive.") _wetness_model_cache = None return None try: from wetness_detector import WetnessModel _wetness_model_cache = WetnessModel.load(WETNESS_MODEL_PATH) print(f"[WETNESS] Loaded model: {_wetness_model_cache.metadata}") except Exception as e: print(f"[WETNESS] Failed to load model, gate inactive: {e}") _wetness_model_cache = None return _wetness_model_cache # CH-substitution detection. The acoustic features (centroid, s_band, bandwidth, # skewness, sub3k, lateral) do NOT separate a CH-substituted ש from a clean ש # in a 200ms window — calibration on 8 recordings (3 dry, 4 wet, 1 deliberate # CH) showed every feature overlapped. CTC competitive alignment is the only # remaining signal. We run a שX vs חX alignment for the syllable; only reject # when ח wins by a decisive margin (calibrated against the K-syllable ח gate). # Map syllable key → (shin_transcript, het_transcript). SYLLABLE_SHIN_CH_MAP = { "sh_syllable_sha": ("שא", "חא"), "sh_syllable_she": ("שה", "חה"), "sh_syllable_shi": ("שי", "חי"), "sh_syllable_shu": ("שו", "חו"), } # Minimum (het_score - shin_score) required to reject as CH. Set to match the # K-syllable ח gate threshold so behavior is consistent across phonemes. SYLLABLE_SHIN_CH_REJECT_GAP = 0.30 # K-syllable scoring is now driven primarily by multi-way CTC competitive # alignment (see SYLLABLE_KUF_ERROR_MAP below). Per-vowel burst-centroid # ranges proved unstable across the speaker population — correct K burst # centroid spread is too wide for sharp thresholds without admitting # substitutions. CTC, given the right set of competing transcripts, is # far more discriminative because the model has been trained on millions # of hours of Hebrew speech and "knows" what each phoneme sounds like. # Map each K syllable to its correct transcript and the most common # substitutions a child might produce. Multi-way competitive_align scores # the audio against all of them; the BEST competitor must lose to ק by # at least SYLLABLE_KUF_MIN_GAP for the recording to be diagnosed CORRECT. # # Common substitutions covered: # ת — velar fronting (most common ק error) # ג — voicing (ק→ג, same place, voiced) # ד — voicing + fronting # ב — bilabial place error # פ — bilabial place + manner # ח — manner error (stop → fricative) SYLLABLE_KUF_ERROR_MAP = { "k_syllable_ka": { "correct": "קא", "errors": ["תא", "גא", "דא", "בא", "פא", "חא"], }, "k_syllable_ke": { "correct": "קה", "errors": ["תה", "גה", "דה", "בה", "פה", "חה"], }, "k_syllable_ki": { "correct": "קי", "errors": ["תי", "גי", "די", "בי", "פי", "חי"], }, "k_syllable_ku": { "correct": "קו", "errors": ["תו", "גו", "דו", "בו", "פו", "חו"], }, } # Phonemes that count as a CORRECT realization of ק for pedagogy purposes. # Per consultation with a speech-language pathologist: ג is a legitimate # developmental substitution that should be praised rather than corrected. # ק itself and ג both get PASS; anything else (ת ד ב פ ח / omission) is FAIL. SYLLABLE_KUF_ACCEPTED_LEADERS = {"ק", "ג"} # When ח leads the competitive alignment, it usually means the child's ק # release had a fricative quality (very common for ק before /u/). Only reject # when ח wins by a very large margin — soft frication should still pass. # Calibration: real ק with light frication had ח_score - ק_score ≤ 0.027, # while a deliberately-said ח had ח_score - ק_score = 0.498. A 0.30 # threshold sits squarely in the gap with comfortable buffer on both sides. SYLLABLE_KUF_HET_GAP_MAX = 0.30 # When the model is "confused" (every transcript scores within this band of # each other), it can't actually discriminate the phoneme. We default to # CORRECT in that case rather than reject a likely-fine ק — substitutions # tend to produce a clear winner, not a tie. SYLLABLE_KUF_CONFUSION_BAND = 0.005 # Acoustic safety net: reject if no real burst landed in the first 250ms. # Catches edge cases where the audio survived the global silence gate but # contains only late vowel energy with no actual K release at the start. SYLLABLE_KUF_MAX_BURST_TIME_MS = 250 def _score_isolated_plosive(trimmed_path: str, word: str) -> dict: """ Score an isolated plosive sound (k_sound / t_sound) using burst spectral centroid. The AI model (Wav2Vec2) has a systematic bias making it useless for isolated single-character discrimination. Instead we use physics: - Velar stop ק: tongue contacts soft palate → burst centroid ~1.5-2.5 kHz - Alveolar stop ת: tongue at alveolar ridge → burst centroid ~2.5-4.0 kHz Key insight: children say "kuh" / "tuh" — the vowel "uh" is much louder than the brief consonant burst. Using max-energy finds the vowel (centroid ~300 Hz), not the burst. We use ONSET DETECTION to find the first significant energy rise in the plosive band — that's the burst, not the vowel. Algorithm: 1. Compute STFT (n_fft=512, hop=128 → 8ms per frame) 2. Energy in 500-5000 Hz per frame 3. Onset detection: first frame crossing 15% of (noise_floor → peak) range 4. Narrow 2-frame window (~16ms) at onset — avoids vowel contamination 5. Centroid computed ONLY in 500-5000 Hz — excludes voiced fundamental 6. k_sound: PASS if centroid < threshold; t_sound: PASS if centroid ≥ threshold """ import soundfile as sf_plosive audio, sr = sf_plosive.read(trimmed_path, dtype="float32") if audio.ndim > 1: audio = audio.mean(axis=1) # STFT: n_fft=512 (32ms at 16kHz), hop=128 (8ms per frame) n_fft = 512 hop_length = 128 stft = np.abs(librosa.stft(audio, n_fft=n_fft, hop_length=hop_length)) freqs = librosa.fft_frequencies(sr=sr, n_fft=n_fft) # Plosive-relevant frequency mask (500-5000 Hz) plosive_mask = (freqs >= 500) & (freqs <= 5000) plosive_freqs = freqs[plosive_mask] frame_energies = np.sum(stft[plosive_mask, :] ** 2, axis=0) # --- Onset detection: find the BURST, not the vowel --- # The burst is the first significant energy rise. The vowel that follows # is louder but comes later. Using max-energy would always find the vowel. noise_floor = float(np.median(frame_energies)) peak_energy = float(np.max(frame_energies)) onset_threshold = noise_floor + (peak_energy - noise_floor) * 0.15 onset_candidates = np.where(frame_energies > onset_threshold)[0] if len(onset_candidates) > 0: burst_frame = int(onset_candidates[0]) else: burst_frame = int(np.argmax(frame_energies)) # Narrow window: burst onset + 1 frame only (~16ms). # Wider windows bleed into the vowel and pull centroid down. start_f = burst_frame end_f = min(stft.shape[1], burst_frame + 2) # Centroid ONLY in plosive range (500-5000 Hz). # Including 0-500 Hz lets the voiced fundamental (~300 Hz) dominate. burst_spectrum = np.sum(stft[plosive_mask, start_f:end_f] ** 2, axis=1) burst_centroid = float( np.sum(plosive_freqs * burst_spectrum) / (np.sum(burst_spectrum) + 1e-10) ) # k_sound expects velar (low centroid); t_sound expects alveolar (high centroid) expecting_velar = (word == "k_sound") if expecting_velar: is_correct = burst_centroid < PLOSIVE_CENTROID_THRESHOLD_HZ correct_label, error_label = "ק (velar)", "ת (alveolar)" else: # t_sound is_correct = burst_centroid >= PLOSIVE_CENTROID_THRESHOLD_HZ correct_label, error_label = "ת (alveolar)", "ק (velar)" print(f"\n{'='*60}") print(f"[SCORE ENGINE] Word: {word} | File: {os.path.basename(trimmed_path)}") print(f"[PLOSIVE] Burst frame: {burst_frame} " f"(t={burst_frame * hop_length / sr * 1000:.0f}ms)") print(f" Noise floor: {noise_floor:.4f} | " f"Peak: {peak_energy:.4f} | Onset threshold: {onset_threshold:.4f}") print(f" Burst centroid: {burst_centroid:.0f} Hz | " f"threshold={PLOSIVE_CENTROID_THRESHOLD_HZ} Hz") print(f" Expecting: {correct_label}") print(f"[RESULT] {'CORRECT → PASS' if is_correct else f'SUBSTITUTION ({error_label}) → FAIL'}") print(f"{'='*60}\n") if expecting_velar: feedback = FEEDBACK_KUF_CORRECT if is_correct else FEEDBACK_KUF_INCORRECT else: feedback = FEEDBACK_TAV_CORRECT if is_correct else FEEDBACK_TAV_INCORRECT diag = "CORRECT" if is_correct else "INCORRECT" score = 100 if is_correct else 0 status = "PASS" if is_correct else "FAIL" return { "score": int(score), "status": status, "diagnosis": diag, "feedback": feedback, "details": { "burst_centroid_hz": int(round(burst_centroid)), "threshold_hz": PLOSIVE_CENTROID_THRESHOLD_HZ, }, "alignment": { "segments": [], }, "evidence": { "burst_centroid_hz": int(round(burst_centroid)), "threshold_hz": PLOSIVE_CENTROID_THRESHOLD_HZ, "expecting_velar": expecting_velar, "is_correct": is_correct, }, } def _shin_samekh_decision(cmp: dict) -> tuple: """ Apply the ש-vs-ס rule to a phoneme_extractor.shin_vs_samekh() result. Returns (is_correct, label, feedback): 'shin' — clear ש → CORRECT 'samekh' — ס beat ש by more than the margin → INCORRECT 'absent' — no clear sibilant at all (vowel / ח / omission / noise) below the presence floor → INCORRECT """ strength = cmp["sibilant_strength"] if strength < SHIN_SAMEKH_PRESENCE_FLOOR: return False, "absent", FEEDBACK_INCORRECT if (cmp["samekh_score"] - cmp["shin_score"]) > SHIN_SAMEKH_MARGIN: return False, "samekh", FEEDBACK_SAMEKH_SUB # Low-confidence band: ש "won", but so weakly that background noise could # produce the same coin-flip. Require a decisive gap before passing. if (strength < SHIN_SAMEKH_CONFIDENT_STRENGTH and (cmp["shin_score"] - cmp["samekh_score"]) < SHIN_SAMEKH_LOWBAND_MIN_GAP): return False, "absent", FEEDBACK_INCORRECT return True, "shin", FEEDBACK_CORRECT def _score_isolated_shin(trimmed_path: str, word: str) -> dict: """ Score the isolated ש sound (sh_sound) by asking the model "ש or ס?" directly. Replaces the old centroid + s_band heuristic, which wrongly failed an emphatic-but-correct ש (a hot ש bleeds energy into 6-10 kHz and the s_band ratio crossed the ס threshold). The model has no such confusion — it learned the actual phoneme. The global silence gate in score_pronunciation already rejects true no-audio before we reach here. """ from phoneme_extractor import shin_vs_samekh cmp = shin_vs_samekh(trimmed_path) is_correct, label, feedback = _shin_samekh_decision(cmp) print(f"\n{'='*60}") print(f"[SCORE ENGINE] Word: {word} | File: {os.path.basename(trimmed_path)} (ISOLATED SH)") print(f"[SH-vs-S] ש={cmp['shin_score']:.4f} | ס={cmp['samekh_score']:.4f} " f"| gap={cmp['gap']:+.4f} | strength={cmp['sibilant_strength']:.4f}") print(f"[DECISION] {label} → {'CORRECT' if is_correct else 'INCORRECT'} " f"(presence≥{SHIN_SAMEKH_PRESENCE_FLOOR}, margin>{SHIN_SAMEKH_MARGIN})") print(f"{'='*60}\n") return { "score": 100 if is_correct else 0, "status": "PASS" if is_correct else "FAIL", "diagnosis": "CORRECT" if is_correct else "INCORRECT", "feedback": feedback, "details": { "shin_score": round(cmp["shin_score"], 4), "samekh_score": round(cmp["samekh_score"], 4), "shin_samekh_gap": cmp["gap"], }, "alignment": {"segments": []}, "evidence": { "shin_score": cmp["shin_score"], "samekh_score": cmp["samekh_score"], "sibilant_strength": cmp["sibilant_strength"], "is_samekh": cmp["is_samekh"], "decision": label, }, } def _extract_temporal_features(audio: np.ndarray, sr: int) -> dict: """ Compute time-domain (not spectral) features of a fricative segment. Spectral features (centroid, s_band) describe WHICH frequencies are present. Temporal features describe HOW the sound flows over time. A clean ש is a steady turbulent stream; a wet/lateral ש has an unstable, bubbling airflow that shows up as fluctuation in the amplitude envelope — even when the average spectrum looks identical. These features are diagnostic-only for now. Returns: env_cov: coefficient of variation of the amplitude envelope (std/mean). Higher = more amplitude fluctuation (bubbling). mod_ratio: fraction of envelope-modulation energy in the 4-16 Hz band (the rate at which saliva bubbling modulates the sound). zcr_cov: coefficient of variation of the zero-crossing rate across frames. Higher = less stable noise character. env_kurtosis: kurtosis of the amplitude envelope. Spiky (bursty) envelope has high kurtosis; steady stream is flatter. """ if len(audio) < 256: return {"env_cov": 0.0, "mod_ratio": 0.0, "zcr_cov": 0.0, "env_kurtosis": 0.0} frame_length = 256 hop_length = 128 # Amplitude envelope via per-frame RMS. At 16 kHz, hop=128 → ~125 Hz # envelope sample rate, enough to resolve modulation up to ~60 Hz. rms = librosa.feature.rms(y=audio, frame_length=frame_length, hop_length=hop_length)[0] env_sr = sr / hop_length mean_env = float(np.mean(rms)) std_env = float(np.std(rms)) env_cov = std_env / (mean_env + 1e-10) # Kurtosis of the envelope (Fisher; 0 = Gaussian-flat, high = spiky). if std_env > 1e-10: env_kurtosis = float(np.mean(((rms - mean_env) / std_env) ** 4) - 3.0) else: env_kurtosis = 0.0 # Modulation spectrum: FFT the (mean-removed) envelope, measure fraction of # energy in the 4-16 Hz band where saliva bubbling tends to live. env_centered = rms - mean_env if len(env_centered) >= 4: spectrum = np.abs(np.fft.rfft(env_centered)) ** 2 mod_freqs = np.fft.rfftfreq(len(env_centered), d=1.0 / env_sr) band = (mod_freqs >= 4.0) & (mod_freqs <= 16.0) total = float(np.sum(spectrum)) + 1e-10 mod_ratio = float(np.sum(spectrum[band]) / total) else: mod_ratio = 0.0 # Zero-crossing-rate stability across frames. zcr = librosa.feature.zero_crossing_rate(audio, frame_length=frame_length, hop_length=hop_length)[0] mean_zcr = float(np.mean(zcr)) zcr_cov = float(np.std(zcr)) / (mean_zcr + 1e-10) return { "env_cov": float(round(env_cov, 4)), "mod_ratio": float(round(mod_ratio, 4)), "zcr_cov": float(round(zcr_cov, 4)), "env_kurtosis": float(round(env_kurtosis, 4)), } def _score_shin_syllable(recording_path: str, word: str, trimmed_path: str) -> dict: """ Score a שׁ-CV syllable (שא/שה/שי/שו) by asking the model "ש or ס?" directly. History: the old centroid + s_band gates were brittle — vowel context (/i/, /u/) pushed a clean ש above the s_band threshold while some ס fell below it, so the bands overlapped and clean ש was wrongly failed. We now decide with phoneme_extractor.shin_vs_samekh (frame-level posteriors), exactly as the ק pipeline asks ק-vs-ת. Spectral/temporal features are still computed and logged for the dataset, but they no longer gate the result. The wetness gate (WavLM) stays inert until a calibrated model ships. """ from phoneme_extractor import extract_shin shin_result = extract_shin(trimmed_path, word) shin_audio = shin_result["shin_audio"] shin_sr = shin_result["sample_rate"] shin_meta = shin_result["shin"] used_fallback = False try: import soundfile as sf_engine full_audio, full_sr = sf_engine.read(recording_path, dtype="float32") if full_audio.ndim > 1: full_audio = full_audio.mean(axis=1) if full_sr != shin_sr: import torchaudio.functional as F_resample import torch as torch_engine waveform = torch_engine.from_numpy(full_audio).unsqueeze(0) waveform = F_resample.resample(waveform, full_sr, shin_sr) full_audio = waveform.squeeze(0).numpy() shin_audio, used_fallback = _extract_shin_segment( full_audio, shin_sr, shin_meta["start_sec"], shin_meta["end_sec"], shin_meta["score"], duration_sec=SYLLABLE_SEGMENT_EXTRACT_DURATION, ) except Exception as e: print(f"[FALLBACK] Could not run fallback detector: {e}") # (No-speech guard removed: the ש-vs-ס presence floor below now handles # "no clear sibilant" via the model's own posteriors, instead of fragile # forced-alignment confidence that wrongly killed valid weak ש.) features = extract_phoneme_features(shin_audio, shin_sr) centroid = features["centroid_mean"] s_band = features["s_band_ratio"] # Temporal (time-domain) features — diagnostic only, no gate yet. # Probing whether wet/lateral ש shows up as envelope instability where the # spectral features (centroid/s_band) failed to separate it from clean ש. temporal = _extract_temporal_features(shin_audio, shin_sr) is_valid_pitch = SYLLABLE_SHIN_CENTROID_LOW <= centroid <= SYLLABLE_SHIN_CENTROID_HIGH is_valid_s_band = s_band < SYLLABLE_SHIN_S_BAND_MAX # CH-substitution gate (CTC-based). Acoustic features can't distinguish # CH from clean ש in a 200ms syllable window, so we ask the model directly: # does the audio look more like שX or חX? Reject only on a decisive ח win. ch_map = SYLLABLE_SHIN_CH_MAP.get(word) is_valid_ch = True ctc_evidence = {} if ch_map is not None: shin_transcript, het_transcript = ch_map try: from phoneme_extractor import _get_word_score shin_ctc_score = _get_word_score(trimmed_path, shin_transcript) het_ctc_score = _get_word_score(trimmed_path, het_transcript) ch_gap = het_ctc_score - shin_ctc_score is_valid_ch = ch_gap < SYLLABLE_SHIN_CH_REJECT_GAP ctc_evidence = { "shin_ctc_score": float(round(shin_ctc_score, 4)), "het_ctc_score": float(round(het_ctc_score, 4)), "ch_gap": float(round(ch_gap, 4)), } except Exception as e: print(f"[CH GATE] CTC scoring failed, skipping gate: {e}") # Wetness / lateral-lisp gate (WavLM one-class OOD). Only the ש-syllable # pipeline uses this. Inert unless a calibrated model pickle is present: # an uncalibrated or absent model yields is_valid_wetness=True (no change). is_valid_wetness = True wetness_evidence = {} wetness_model = _get_wetness_model() if wetness_model is not None: try: from wetness_detector import extract_embedding import soundfile as sf_wet # Feed the WHOLE trimmed syllable to WavLM — this matches how the # wetness model was trained (whole-clip embeddings), NOT the 200ms # spectral-analysis segment. Train/inference inputs must match. wet_audio, wet_sr = sf_wet.read(trimmed_path, dtype="float32") if wet_audio.ndim > 1: wet_audio = wet_audio.mean(axis=1) emb = extract_embedding(wet_audio, wet_sr) verdict = wetness_model.verdict(emb) is_valid_wetness = verdict["label"] != "wet" wetness_evidence = { "wetness_distance": float(round(verdict["distance"], 3)), "wetness_label": verdict["label"], } except Exception as e: print(f"[WETNESS] Scoring failed, skipping wetness gate: {e}") # --- PRIMARY decision: ask the model "ש or ס?" via frame-level posteriors --- # This replaces the brittle centroid + s_band gates (kept below as info-only # logging). Mirrors the ק pipeline's competitive approach. The presence floor # inside _shin_samekh_decision rejects "no clear sibilant" (ח / vowel / noise). from phoneme_extractor import shin_vs_samekh cmp = shin_vs_samekh(trimmed_path) shin_ok, sh_label, sh_feedback = _shin_samekh_decision(cmp) # centroid / s_band / CH are now INFO-ONLY (logged, not gating). is_pass = shin_ok and is_valid_wetness # Full feature dump for wetness-detection calibration. The current 2-gate # syllable pipeline (centroid + s_band) misses lateral lisp / wet ש because # neither feature moves much on subtle wetness. We log every available # feature here so a future calibration round can identify which one # actually separates wet from dry syllables. print(f"\n{'='*60}") print(f"[SCORE ENGINE] Word: {word} | File: {os.path.basename(recording_path)} (SH SYLLABLE)") print(f"[FEATURES] centroid_mean={features['centroid_mean']:.1f} Hz") print(f" centroid_median={features['centroid_median']:.1f} Hz") print(f" bandwidth_mean={features['bandwidth_mean']:.1f} Hz") print(f" spectral_skewness={features['spectral_skewness']:.4f}") print(f" high_mid_ratio={features['high_mid_ratio']:.4f}") print(f" s_band_ratio={features['s_band_ratio']:.4f} " f"({features['s_band_ratio']*100:.1f}%)") print(f" sub3k_ratio={features['sub3k_ratio']:.4f} " f"({features['sub3k_ratio']*100:.1f}%)") print(f" lateral_ratio={features['lateral_ratio']:.4f}") print(f"[TEMPORAL] env_cov={temporal['env_cov']:.4f} " f"(envelope fluctuation — higher = bubbling)") print(f" mod_ratio={temporal['mod_ratio']:.4f} " f"(4-16Hz modulation energy fraction)") print(f" zcr_cov={temporal['zcr_cov']:.4f} " f"(zero-crossing instability)") print(f" env_kurtosis={temporal['env_kurtosis']:.4f} " f"(envelope spikiness)") print(f"[SH-vs-S] ש={cmp['shin_score']:.4f} | ס={cmp['samekh_score']:.4f} " f"| gap={cmp['gap']:+.4f} | strength={cmp['sibilant_strength']:.4f} → {sh_label} " f"(presence≥{SHIN_SAMEKH_PRESENCE_FLOOR}, margin>{SHIN_SAMEKH_MARGIN})") print(f"[INFO-ONLY] centroid={centroid:.1f}Hz, s_band={s_band*100:.1f}%" + (f", CH gap={ctc_evidence['ch_gap']:+.4f}" if ctc_evidence else "") + (f", wetness={wetness_evidence['wetness_label']}" if wetness_evidence else "")) print(f"[RESULT] {'CORRECT → PASS' if is_pass else 'INCORRECT → FAIL'}") print(f"{'='*60}\n") diag = "CORRECT" if is_pass else "INCORRECT" if is_pass: feedback = FEEDBACK_CORRECT elif not is_valid_wetness: feedback = FEEDBACK_INCORRECT else: feedback = sh_feedback return { "score": 100 if is_pass else 0, "status": "PASS" if is_pass else "FAIL", "diagnosis": diag, "feedback": feedback, "details": { "shin_score": round(cmp["shin_score"], 4), "samekh_score": round(cmp["samekh_score"], 4), "shin_samekh_gap": cmp["gap"], "centroid_hz": int(round(centroid)), "s_band_ratio": float(round(s_band, 4)), **ctc_evidence, }, "alignment": { "segments": shin_result["segments"], "shin": shin_result["shin"], "used_fallback_detector": used_fallback, }, "evidence": { "shin_score": cmp["shin_score"], "samekh_score": cmp["samekh_score"], "sibilant_strength": cmp["sibilant_strength"], "is_samekh": cmp["is_samekh"], "decision": sh_label, "centroid_mean": float(centroid), "s_band_ratio": float(s_band), "is_valid_wetness": bool(is_valid_wetness), **{f"temporal_{k}": v for k, v in temporal.items()}, **ctc_evidence, **wetness_evidence, }, } def _score_kuf_syllable(trimmed_path: str, word: str) -> dict: """ Score a ק-CV syllable (קא, קה, קי, קו) using top-choice CTC selection with clinical-pedagogy-aware accept/reject rules. Decision logic (per SLP consultation): 1. If ק or ג is the top-scoring transcript among the 7 candidates (ק, ת, ג, ד, ב, פ, ח), the recording PASSES — ג is a legitimate developmental substitution that should be praised. 2. If ח is the top scorer: - If ח beats ק by > 0.30, the child clearly produced a fricative, not ק → FAIL. - Otherwise (ח-ק gap ≤ 0.30) it's a slightly fricated ק → PASS. 3. If ת / ד / ב / פ is the top scorer → FAIL (real substitution). 4. If every transcript scores within 0.005 of each other, the CTC model is "confused" (typical of very short 2-char alignments) — we default to PASS rather than reject a probably-fine ק arbitrarily. Acoustic safety net: regardless of CTC outcome, a real burst must land within 250ms post-trim. Late "bursts" are usually misclassified vowel onsets, indicating the ק itself was omitted. All-pass criterion: CTC verdict is "ק-acceptable" AND burst is timely. """ from phoneme_extractor import multi_competitive_align import soundfile as sf_plosive # ─── Acoustic safety net: detect a burst, log its position ────────── audio, sr = sf_plosive.read(trimmed_path, dtype="float32") if audio.ndim > 1: audio = audio.mean(axis=1) n_fft = 512 hop_length = 128 stft = np.abs(librosa.stft(audio, n_fft=n_fft, hop_length=hop_length)) freqs = librosa.fft_frequencies(sr=sr, n_fft=n_fft) plosive_mask = (freqs >= 500) & (freqs <= 5000) frame_energies = np.sum(stft[plosive_mask, :] ** 2, axis=0) noise_floor = float(np.median(frame_energies)) peak_energy = float(np.max(frame_energies)) onset_threshold = noise_floor + (peak_energy - noise_floor) * 0.15 onset_candidates = np.where(frame_energies > onset_threshold)[0] if len(onset_candidates) > 0: burst_frame = int(onset_candidates[0]) else: burst_frame = int(np.argmax(frame_energies)) burst_time_ms = burst_frame * hop_length / sr * 1000 gate_position = burst_time_ms < SYLLABLE_KUF_MAX_BURST_TIME_MS # ─── Primary gate: multi-way CTC + top-choice rule ───────────────── error_map = SYLLABLE_KUF_ERROR_MAP.get(word) if error_map is None: print(f"[K SYLLABLE] Unknown word '{word}' — no error map; " f"falling back to position gate only") gate_ctc = True ctc_evidence = {} ctc_summary = "no error map configured" verdict = "unknown-word fallback" else: ctc = multi_competitive_align( trimmed_path, error_map["correct"], error_map["errors"], ) correct_score = ctc["correct_score"] all_scores = dict(ctc["all_error_scores"]) all_scores[error_map["correct"]] = correct_score # Identify the top-scoring transcript among all 7 candidates. top_transcript = max(all_scores, key=all_scores.get) top_score = all_scores[top_transcript] # First character of the top transcript identifies the phoneme: # the syllable structure is always [consonant][vowel]. top_phoneme = top_transcript[0] if top_transcript else "" # Confusion detection: every candidate scores within a tiny band. score_range = max(all_scores.values()) - min(all_scores.values()) is_confused = score_range < SYLLABLE_KUF_CONFUSION_BAND # Apply the SLP-informed acceptance rules. het_transcript = "ח" + error_map["correct"][1] # e.g. "חא" for ka het_score = all_scores.get(het_transcript, 0.0) het_minus_k_gap = het_score - correct_score if is_confused: gate_ctc = True verdict = (f"confused (range={score_range:.4f} < " f"{SYLLABLE_KUF_CONFUSION_BAND}) — accept by default") elif top_phoneme in SYLLABLE_KUF_ACCEPTED_LEADERS: gate_ctc = True verdict = f"top='{top_transcript}' ({top_phoneme} is accepted)" elif top_phoneme == "ח": if het_minus_k_gap > SYLLABLE_KUF_HET_GAP_MAX: gate_ctc = False verdict = (f"top='{top_transcript}' (ח-ק gap {het_minus_k_gap:+.4f} " f"> {SYLLABLE_KUF_HET_GAP_MAX} — strong fricative, reject)") else: gate_ctc = True verdict = (f"top='{top_transcript}' but ח-ק gap " f"{het_minus_k_gap:+.4f} ≤ {SYLLABLE_KUF_HET_GAP_MAX} " f"(light frication, accept)") else: gate_ctc = False verdict = (f"top='{top_transcript}' ({top_phoneme} is a real " f"substitution, reject)") ctc_evidence = { "correct_score": correct_score, "top_transcript": top_transcript, "top_score": float(round(top_score, 4)), "top_phoneme": top_phoneme, "all_scores": {k: float(round(v, 4)) for k, v in all_scores.items()}, "score_range": float(round(score_range, 4)), "het_minus_k_gap": float(round(het_minus_k_gap, 4)), "is_confused": bool(is_confused), "gate_ctc": bool(gate_ctc), } ctc_summary = (f"correct '{error_map['correct']}'={correct_score:.4f}, " f"top='{top_transcript}'={top_score:.4f}, " f"ח-ק gap={het_minus_k_gap:+.4f}") is_pass = gate_ctc and gate_position fail_reasons = [] if error_map is not None and not gate_ctc: fail_reasons.append(verdict) if not gate_position: fail_reasons.append( f"burst at {burst_time_ms:.0f}ms ≥ {SYLLABLE_KUF_MAX_BURST_TIME_MS}ms " f"(no real burst — likely omission)" ) fail_reason = "; ".join(fail_reasons) print(f"\n{'='*60}") print(f"[SCORE ENGINE] Word: {word} | File: {os.path.basename(trimmed_path)} (K SYLLABLE)") print(f"[BURST] Frame: {burst_frame} (t={burst_time_ms:.0f}ms) | " f"Peak: {peak_energy:.4f}") print(f"[CTC] {ctc_summary}") print(f"[VERDICT] {verdict}") print(f"[GATES] 1) CTC verdict: " f"{'PASS' if gate_ctc else 'FAIL'}") print(f" 2) Burst < {SYLLABLE_KUF_MAX_BURST_TIME_MS}ms: " f"{'PASS' if gate_position else 'FAIL'} ({burst_time_ms:.0f}ms)") print(f"[RESULT] {'CORRECT → PASS' if is_pass else f'INCORRECT → FAIL ({fail_reason})'}") print(f"{'='*60}\n") diag = "CORRECT" if is_pass else "INCORRECT" return { "score": 100 if is_pass else 0, "status": "PASS" if is_pass else "FAIL", "diagnosis": diag, "feedback": FEEDBACK_KUF_CORRECT if is_pass else FEEDBACK_KUF_INCORRECT, "details": { "burst_time_ms": int(round(burst_time_ms)), **ctc_evidence, }, "alignment": {"segments": []}, "evidence": { "burst_time_ms": int(round(burst_time_ms)), "gate_position": bool(gate_position), "verdict": verdict, "fail_reason": fail_reason, **ctc_evidence, }, } def _check_audio_has_signal(audio_path: str) -> tuple: """ Return (has_signal, peak_amplitude, rms) for the audio at the given path. A recording is considered to have signal if either the peak amplitude exceeds SIGNAL_MIN_PEAK_AMPLITUDE or the RMS exceeds SIGNAL_MIN_RMS. Both conditions are checked because some recordings have brief clicks (high peak, low RMS) while quiet sustained speech has the opposite. """ try: import soundfile as sf_check audio, _sr = sf_check.read(audio_path, dtype="float32") if audio.ndim > 1: audio = audio.mean(axis=1) peak = float(np.max(np.abs(audio))) if len(audio) > 0 else 0.0 rms = float(np.sqrt(np.mean(audio ** 2))) if len(audio) > 0 else 0.0 has_signal = peak >= SIGNAL_MIN_PEAK_AMPLITUDE or rms >= SIGNAL_MIN_RMS return has_signal, peak, rms except Exception as e: # If we can't read the audio, assume signal is present and let the # downstream pipeline produce its own error. Better to over-process # than to wrongly reject a recording due to a soundfile glitch. print(f"[SIGNAL CHECK] Could not read audio for signal check: {e}") return True, 0.0, 0.0 def _reduce_noise(recording_path: str) -> str: """ Apply non-stationary spectral gating noise reduction. Uses noisereduce's non-stationary mode, which adapts to variable backgrounds (classrooms, living rooms, wind) rather than assuming constant noise. Safe to run even if noisereduce is missing — returns the original path on any error. """ try: import noisereduce as nr import soundfile as sf_nr import tempfile audio, sr = sf_nr.read(recording_path, dtype="float32") if audio.ndim > 1: audio = audio.mean(axis=1) # Non-stationary mode handles variable room noise better than stationary. # prop_decrease=0.8 — aggressive enough for noisy rooms, safe for speech. reduced = nr.reduce_noise(y=audio, sr=sr, stationary=False, prop_decrease=0.8) denoised_path = os.path.join( tempfile.gettempdir(), f"denoised_{os.path.basename(recording_path)}" ) sf_nr.write(denoised_path, reduced, sr) print(f"[DENOISE] Non-stationary spectral gating applied " f"({len(audio)} samples, {sr} Hz)") return denoised_path except Exception as e: print(f"[DENOISE] Noise reduction failed (continuing with original): {e}") return recording_path def _trim_silence(recording_path: str) -> str: """ Trim leading silence from a recording, return path to trimmed file. Returns the original path if no significant silence found. Mobile recordings often have 1-2s of silence before the user speaks, which causes Wav2Vec2 alignment to fail (score ~0.0001). """ try: import soundfile as sf_trim audio_raw, sr_raw = sf_trim.read(recording_path, dtype="float32") if audio_raw.ndim > 1: audio_raw = audio_raw.mean(axis=1) trimmed_audio, trim_index = librosa.effects.trim(audio_raw, top_db=25) trim_offset_sec = trim_index[0] / sr_raw # Guarantee at least 0.5s of audio after trimming. # Isolated plosives (K/T) leave only ~0.3s after aggressive trim. # Anchor on the speech ONSET (trim_index[0]) so we always capture the # actual sound — keeping the "last 0.5s" would grab trailing silence # when the child spoke at the start of the recording. min_samples = int(0.5 * sr_raw) if len(trimmed_audio) < min_samples: keep_start = trim_index[0] keep_end = min(len(audio_raw), keep_start + min_samples) trimmed_audio = audio_raw[keep_start:keep_end] trim_offset_sec = keep_start / sr_raw print(f"[TRIM] Trim would leave <0.5s — keeping {len(trimmed_audio)/sr_raw:.3f}s from speech onset instead") if trim_offset_sec > 0.05: import tempfile trimmed_path = os.path.join( tempfile.gettempdir(), f"trimmed_{os.path.basename(recording_path)}" ) sf_trim.write(trimmed_path, trimmed_audio, sr_raw) print(f"[TRIM] Removed {trim_offset_sec:.3f}s of leading silence " f"({len(audio_raw)} → {len(trimmed_audio)} samples)") return trimmed_path except Exception as e: print(f"[TRIM] Silence trimming failed (continuing with original): {e}") return recording_path def _score_shin_word(recording_path: str, word: str, trimmed_path: str) -> dict: """ Score a ש (Shin) word with two gates: 1. AI-confidence omission gate (forced-align confidence of the ש ≥ 0.70). 2. Model-driven ש-vs-ס discrimination (shin_vs_samekh, frame posteriors), replacing the old s_band energy ratio that failed emphatic-but-correct ש. centroid / s_band are computed for logging only and no longer gate. """ # Step 1: Phoneme isolation via Wav2Vec2 forced alignment from phoneme_extractor import extract_shin shin_result = extract_shin(trimmed_path, word) # Step 1.5: Fallback fricative refinement shin_audio = shin_result["shin_audio"] shin_sr = shin_result["sample_rate"] shin_meta = shin_result["shin"] used_fallback = False try: import soundfile as sf_engine full_audio, full_sr = sf_engine.read(recording_path, dtype="float32") if full_audio.ndim > 1: full_audio = full_audio.mean(axis=1) if full_sr != shin_sr: import torchaudio.functional as F_resample import torch as torch_engine waveform = torch_engine.from_numpy(full_audio).unsqueeze(0) waveform = F_resample.resample(waveform, full_sr, shin_sr) full_audio = waveform.squeeze(0).numpy() shin_audio, used_fallback = _extract_shin_segment( full_audio, shin_sr, shin_meta["start_sec"], shin_meta["end_sec"], shin_meta["score"] ) except Exception as e: print(f"[FALLBACK] Could not run fallback detector: {e}") # (sh_sound is routed to _score_isolated_shin upstream; this path is ש WORDS.) # Step 2: Acoustic features (for the AI omission gate + dataset logging) features = extract_phoneme_features(shin_audio, shin_sr) alignment_score = shin_meta["score"] duration = shin_meta["duration"] if used_fallback: duration = max(duration, len(shin_audio) / shin_sr) # Step 3a: AI-confidence gate (omission catch). For a real ש word the # forced-alignment confidence of the ש is high (~0.90-0.9999 in practice); # a word that wasn't actually produced drops well below 0.70. is_valid_ai = alignment_score >= GOLDILOCKS_AI_SCORE_MIN # Step 3b: PRIMARY ס detector — ask the model "ש or ס?" on the whole word # via frame-level posteriors (phoneme_extractor.shin_vs_samekh), replacing # the brittle s_band energy ratio that mislabeled emphatic/hot ש as ס. from phoneme_extractor import shin_vs_samekh cmp = shin_vs_samekh(trimmed_path) shin_ok, sh_label, _sh_feedback = _shin_samekh_decision(cmp) is_correct = is_valid_ai and shin_ok score = 100 if is_correct else 0 status = "PASS" if is_correct else "FAIL" diag = "CORRECT" if is_correct else "INCORRECT" if is_correct: feedback = FEEDBACK_CORRECT elif not is_valid_ai: feedback = FEEDBACK_INCORRECT # the ש word wasn't clearly produced elif sh_label == "samekh": feedback = FEEDBACK_SAMEKH_SUB else: feedback = FEEDBACK_INCORRECT # Debug logging print(f"\n{'='*60}") print(f"[SCORE ENGINE] Word: {word} | File: {os.path.basename(recording_path)}") print(f"[ALIGNMENT] ש score={alignment_score:.4f}, duration={duration:.4f}s, " f"fallback={'YES' if used_fallback else 'no'}") print(f"[SH-vs-S] ש={cmp['shin_score']:.4f} | ס={cmp['samekh_score']:.4f} " f"| gap={cmp['gap']:+.4f} | strength={cmp['sibilant_strength']:.4f} → {sh_label}") print(f"[GATES] AI(omission) {alignment_score:.4f} ≥ {GOLDILOCKS_AI_SCORE_MIN}: " f"{'PASS' if is_valid_ai else 'FAIL'} | ש-vs-ס: " f"{'PASS' if shin_ok else 'FAIL'}") print(f"[INFO-ONLY] centroid={features['centroid_mean']:.1f}Hz, " f"s_band={features['s_band_ratio']*100:.1f}%, " f"sub3k={features['sub3k_ratio']*100:.1f}%") print(f"[RESULT] {diag} → {status}") print(f"{'='*60}\n") return { "score": int(score), "status": status, "diagnosis": diag, "feedback": feedback, "details": { "shin_score": round(cmp["shin_score"], 4), "samekh_score": round(cmp["samekh_score"], 4), "shin_samekh_gap": cmp["gap"], "alignment_score": float(round(alignment_score, 4)), "centroid_hz": int(round(features["centroid_mean"])), }, "alignment": { "segments": shin_result["segments"], "shin": shin_result["shin"], "used_fallback_detector": used_fallback, }, "evidence": { "shin_score": cmp["shin_score"], "samekh_score": cmp["samekh_score"], "sibilant_strength": cmp["sibilant_strength"], "is_samekh": cmp["is_samekh"], "decision": sh_label, "alignment_score": float(alignment_score), "is_valid_ai": bool(is_valid_ai), "centroid_mean": float(features["centroid_mean"]), "s_band_ratio": float(features["s_band_ratio"]), }, } def _score_kuf_word(recording_path: str, word: str, trimmed_path: str) -> dict: """ Score a ק (Kuf) word using Competitive Alignment: Align against CORRECT transcript (ק) and ERROR transcript (ת), compare confidence — if error scores higher, child substituted T for K. Isolated sounds (k_sound / t_sound) are routed to _score_isolated_plosive() which uses burst spectral centroid — the Wav2Vec2 model has a systematic bias that makes it unable to discriminate isolated single-character plosives. """ # Isolated plosive: bypass AI model, use burst spectral centroid instead if word in ("k_sound", "t_sound"): return _score_isolated_plosive(trimmed_path, word) from phoneme_extractor import competitive_align result = competitive_align(trimmed_path, word) correct_score = result["correct_score"] error_score = result["error_score"] gap = result["confidence_gap"] is_substitution = result["is_substitution"] # Debug logging print(f"\n{'='*60}") print(f"[SCORE ENGINE] Word: {word} | File: {os.path.basename(recording_path)}") print(f"[COMPETITIVE] Correct '{result['correct_transcript']}': {correct_score:.4f}") print(f" Error '{result['error_transcript']}': {error_score:.4f}") print(f" Gap: {gap:+.4f}") print(f"[RESULT] {'SUBSTITUTION → FAIL' if is_substitution else 'CORRECT → PASS'}") print(f"{'='*60}\n") diag = "INCORRECT" if is_substitution else "CORRECT" is_correct = not is_substitution score = 100 if is_correct else 0 status = "PASS" if is_correct else "FAIL" feedback = FEEDBACK_KUF_CORRECT if is_correct else FEEDBACK_KUF_INCORRECT return { "score": int(score), "status": status, "diagnosis": diag, "feedback": feedback, "details": { "correct_score": float(round(correct_score, 4)), "error_score": float(round(error_score, 4)), "confidence_gap": float(round(gap, 4)), }, "alignment": { "segments": result["segments"], }, "evidence": { "correct_transcript": result["correct_transcript"], "error_transcript": result["error_transcript"], "correct_score": float(round(correct_score, 4)), "error_score": float(round(error_score, 4)), "confidence_gap": float(round(gap, 4)), "is_substitution": bool(is_substitution), }, } def score_pronunciation(recording_path: str, word: str) -> dict: """ Score a child's pronunciation using pure AI inference. Routes to the appropriate scoring pipeline based on the word's phoneme type: - ש (Shin) words → 2-gate system (AI score + centroid) - ק (Kuf) words → Competitive Alignment (correct vs error transcript) Args: recording_path: Path to the user's WAV recording. word: Target word key (e.g. 'shalom', 'kof'). Returns: JSON-compatible dict with score, status, diagnosis, feedback, acoustic details, and alignment metadata. """ if not os.path.exists(recording_path): return { "score": 0, "status": "ERROR", "error_type": "file_not_found", "feedback": f"Recording file not found: {recording_path}", "details": {}, } # Step 0a: Noise reduction (non-stationary spectral gating). # SKIP for isolated plosives (k_sound / t_sound): aggressive spectral gating # destroys the brief burst transient that is the entire diagnostic signal. # A 3-second recording with 2.5s of silence causes the algorithm to learn # "this is a quiet recording" and suppress the K/T burst as noise. from phoneme_extractor import WORD_PHONEME_TYPE phoneme_type = WORD_PHONEME_TYPE.get(word, "shin") if word in ("k_sound", "t_sound", "sh_sound"): denoised_path = recording_path print(f"[DENOISE] Skipping noise reduction for isolated sound ({word}) — preserving spectral profile") else: denoised_path = _reduce_noise(recording_path) # Step 0b: Trim leading silence on the denoised audio trimmed_path = _trim_silence(denoised_path) # Clean up intermediate denoised file if trimming produced a new file if denoised_path != recording_path and denoised_path != trimmed_path: try: os.remove(denoised_path) except OSError: pass # Step 0c: Global silence/no-signal gate. # If the user recorded absolute silence, the Wav2Vec2 alignment fails, # the energy-scan fallback picks up random room noise, and the spectral # gates can accidentally pass that noise as CORRECT. Reject before any # pipeline runs. has_signal, peak_amp, rms = _check_audio_has_signal(trimmed_path) if not has_signal: print(f"[SIGNAL CHECK] Recording rejected as silence " f"(peak={peak_amp:.4f} < {SIGNAL_MIN_PEAK_AMPLITUDE}, " f"rms={rms:.4f} < {SIGNAL_MIN_RMS})") if trimmed_path != recording_path and os.path.exists(trimmed_path): try: os.remove(trimmed_path) except OSError: pass return { "score": 0, "status": "FAIL", "diagnosis": "INCORRECT", "error_type": "silence_or_no_signal", "feedback": FEEDBACK_NO_SIGNAL, "details": { "peak_amplitude": peak_amp, "rms": rms, }, "evidence": {"reason": "audio_too_quiet"}, } try: # ── HYBRID routing (the recommended plan) ──────────────────────────── # STT shines for the isolated SOUND and CV SYLLABLE stages: there is no # real word for a transcription model to "autocorrect" a non-word # substitution back into, so what it writes is what it heard. # For full WORDS, STT has a language-model prior and will quietly fix a # non-word slip ("דבס"→"דבש"), hiding the error — so words are scored by # the autocorrect-immune MATH pipeline (competitive ש-vs-ס / ק-vs-ת, # reading phoneme posteriors directly). # SCORING_BACKEND=="stt" enables STT for the sound/syllable stages only; # SCORING_BACKEND=="math" disables STT everywhere. The math pipeline is # always the fallback when STT can't decide. _is_sound_or_syllable = ( word in ("sh_sound", "k_sound", "t_sound") or word.startswith("sh_syllable_") or word.startswith("k_syllable_") ) if SCORING_BACKEND == "stt" and _is_sound_or_syllable: try: from stt_judge import judge as _stt_judge stt_result = _stt_judge(trimmed_path, word) except Exception as e: print(f"[STT] judge crashed — falling back to math: {e}") stt_result = None if stt_result is not None: return stt_result print("[STT] no STT decision — falling back to math pipeline") elif SCORING_BACKEND == "stt": print(f"[HYBRID] '{word}' is a full word → math pipeline (autocorrect-immune)") # Route to the correct pipeline. phoneme_type already resolved above. # CV syllables (sh_syllable_* / k_syllable_*) bypass the full-word # pipelines: their 2-char transcripts make CTC confidence and sub3k # gates unreliable. They use stripped-down acoustic-only gates. if word == "sh_sound": # Isolated ש: model-driven ש-vs-ס (raw posteriors), no fragile # forced-alignment or s_band heuristic. return _score_isolated_shin(trimmed_path, word) if word.startswith("k_syllable_"): return _score_kuf_syllable(trimmed_path, word) if word.startswith("sh_syllable_"): return _score_shin_syllable(trimmed_path, word, trimmed_path) if phoneme_type == "kuf": return _score_kuf_word(trimmed_path, word, trimmed_path) else: return _score_shin_word(trimmed_path, word, trimmed_path) except Exception as e: # NEVER return the raw exception to the client. Log the full traceback # server-side (captured into server_logs for debugging), and return a # valid, complete result with Hebrew "try again" feedback. status=ERROR # (not FAIL) so the game treats it as a retry, not a lost heart — the # child shouldn't be penalized for a processing error. import traceback print(f"[SCORE ENGINE] Scoring crashed for word '{word}': {e}") print(traceback.format_exc()) return { "score": 0, "status": "ERROR", "diagnosis": "ERROR", "error_type": "scoring_failed", "feedback": "אופס, לא הצלחנו לעבד את ההקלטה. נסו שוב 🙂", "details": {}, "alignment": {"segments": []}, "evidence": {"reason": "scoring_exception"}, } finally: if trimmed_path != recording_path and os.path.exists(trimmed_path): os.remove(trimmed_path) # ============================================================================= # CLI Interface # ============================================================================= def main(): """Command-line interface for the score engine.""" parser = argparse.ArgumentParser( description="Score a pronunciation recording using AI inference" ) parser.add_argument( "recording_path", help="Path to the user's audio recording" ) parser.add_argument( "word", help="Target word (e.g. 'shalom', 'shemesh', 'shir')" ) parser.add_argument( "--pretty", action="store_true", help="Pretty-print the JSON output" ) args = parser.parse_args() result = score_pronunciation(args.recording_path, args.word) indent = 2 if args.pretty else None print(json.dumps(result, indent=indent, ensure_ascii=False)) if __name__ == "__main__": main()