""" webapp.py - Clinical-Grade Multi-Modal Speech Diagnostics Interface =================================================================== A high-end, research-grade Streamlit application for clinical speech assessment: - Real-Time Live Streaming Diagnostics HUD & Subsystem Telemetry - Neural Disfluency Detection (wav2vec2 + LoRA) - Phonetic Word-Level Alignment & Pronunciation Accuracy (wav2vec2-CTC) - Explicit Sound Disorder Rules (Rhotacism 'r', Sigmatism 's') - Biomechanical Vocal Fold Phonation (Parselmouth / Praat PointProcess) - Multi-Modal Decision Fusion with Speaker Self-Calibration """ from __future__ import annotations import io import os import tempfile import time from pathlib import Path from typing import Optional, Dict, Any, List, Tuple import numpy as np import soundfile as sf import streamlit as st from ml.model.engine import SpeechDiagnosticEngine # Page Configuration st.set_page_config( page_title="Anvaya | Clinical Speech Diagnostics", layout="wide", initial_sidebar_state="expanded", ) # Benchmark Sample Library (Real Physical Audio on Local Disk) BENCHMARK_SAMPLES = { "Fluent Control (Normal Cadence)": ( "data/synthetic_lattice/audio/synth_000000_fluent_control.wav", "o that like here in the states to beco" ), "Syllable Repetition (Disfluency Test)": ( "data/synthetic_lattice/audio/synth_001000_stutter_repetition.wav", "it too am" ), "Sound Prolongation (Disfluency Test)": ( "data/synthetic_lattice/audio/synth_002000_stutter_prolongation.wav", "im going to revert back to you" ), "Glottal Block (Laryngeal Arrest)": ( "data/synthetic_lattice/audio/synth_003000_stutter_block.wav", "hopes" ), } # Single Word Practice Presets WORD_PRESETS = [ "rabbit", "red", "sun", "sweet", "three", "water", "kitten", "spot", ] # Full Sentence Practice Presets SENTENCE_PRESETS = { "The Red Rabbit (Rhotacism & Rhotic 'R' Evaluation)": "the red rabbit ran around the green yard", "The Sweet Sun (Sigmatism & Sibilant 'S' Evaluation)": "the sweet sun shines softly in the sky", "The Blue Spot (Phonetic Balance & Plosives)": "the blue spot is on the key", "The Rainbow Passage (Standardized Clinical Protocol)": "the rainbow is a division of white light into many beautiful colors", } # High-End Design Engineering CSS (Zero Emojis, Minimalist Modern Health-Tech) st.markdown(""" """, unsafe_allow_html=True) @st.cache_resource(show_spinner="Initializing diagnostic neural engine...") def _get_engine(): return SpeechDiagnosticEngine.get_instance() def _process_audio_bytes(raw_bytes: bytes, filename: str = "recording.wav") -> Tuple[str, bytes, float]: """Resample, condition, and normalize any audio buffer to standard 16kHz PCM WAV.""" from ml.model.pron_eval import _load_wave, SR arr = _load_wave(raw_bytes) tmp_path = Path(tempfile.gettempdir()) / f"anvaya_std_{Path(filename).stem}.wav" sf.write(str(tmp_path), arr, SR, subtype="PCM_16") buf = io.BytesIO() sf.write(buf, arr, SR, format="WAV", subtype="PCM_16") dur = float(len(arr) / SR) return str(tmp_path), buf.getvalue(), dur def _read_file_to_bytes(file_path: str) -> Tuple[str, bytes, float]: """Read a local audio file and ensure standard 16kHz WAV format for browser playback.""" from ml.model.pron_eval import _load_wave, SR arr = _load_wave(file_path) buf = io.BytesIO() sf.write(buf, arr, SR, format="WAV", subtype="PCM_16") dur = float(len(arr) / SR) return file_path, buf.getvalue(), dur def _plot_waveform(audio_path: str): """Plot acoustic waveform & energy envelope.""" try: import matplotlib.pyplot as plt arr, sr = sf.read(audio_path, dtype="float32") if arr.ndim > 1: arr = arr.mean(axis=1) if len(arr) == 0: return time_axis = np.linspace(0, len(arr) / sr, num=len(arr)) fig, ax = plt.subplots(figsize=(10, 1.8), dpi=100) fig.patch.set_facecolor("none") ax.set_facecolor("none") ax.plot(time_axis, arr, color="#38BDF8", alpha=0.9, linewidth=0.8) ax.fill_between(time_axis, arr, -arr, color="#0284C7", alpha=0.18) ax.set_xlabel("Time (seconds)", fontsize=8, color="#64748B") ax.set_ylabel("Amplitude", fontsize=8, color="#64748B") ax.tick_params(colors="#64748B", labelsize=8) for spine in ax.spines.values(): spine.set_color("rgba(255,255,255,0.08)") plt.tight_layout(pad=0.4) st.pyplot(fig) plt.close(fig) except Exception: pass def main(): engine = _get_engine() # Session State Initialization if "target_phrase" not in st.session_state: st.session_state["target_phrase"] = "the red rabbit ran around the green yard" # Top Brand Navigation Bar import torch device_label = "CUDA (GPU Accelerated)" if torch.cuda.is_available() else "CPU" st.markdown(f"""

ANVAYA v2.0 Assistant

Multi-Modal Speech Screening, Phonetic Alignment & Acoustic Phonation Analysis

Engine Online · {device_label}

""", unsafe_allow_html=True) st.info( "**Clinical Practice & Screening Disclaimer**: Anvaya is an exploratory research prototype for assistive speech practice and acoustic analysis. It is not an FDA-cleared diagnostic medical device and does not substitute for a clinical evaluation by a licensed Speech-Language Pathologist (SLP)." ) selected_audio_path: Optional[str] = None playable_wav_bytes: Optional[bytes] = None audio_dur: float = 0.0 # Sidebar: Audio Source Setup with st.sidebar: st.markdown("### Audio Ingestion") input_mode = st.radio( "Acquisition Method:", ["Record Microphone Audio", "Upload Audio File (.wav, .mp3, .webm, .m4a)", "Benchmark Sample Library"], ) if input_mode == "Record Microphone Audio": st.caption("Press record to capture speech. If browser errors occur, use direct File Upload.") try: mic_audio = st.audio_input("Microphone Input") if mic_audio: try: selected_audio_path, playable_wav_bytes, audio_dur = _process_audio_bytes(mic_audio.getvalue(), "mic_recording.wav") except Exception as ex: st.error(f"Could not decode audio: {ex}. Please try again or drop a WAV file.") except AttributeError: st.info("Direct audio recording requires Streamlit 1.40+.") elif input_mode == "Upload Audio File (.wav, .mp3, .webm, .m4a)": uploaded = st.file_uploader("Upload audio recording", type=["wav", "mp3", "webm", "m4a", "ogg", "flac"]) if uploaded: try: selected_audio_path, playable_wav_bytes, audio_dur = _process_audio_bytes(uploaded.getvalue(), uploaded.name) except Exception as ex: st.error(f"Error reading uploaded file: {ex}") elif input_mode == "Benchmark Sample Library": available_samples = {k: v for k, v in BENCHMARK_SAMPLES.items() if Path(v[0]).exists()} if available_samples: chosen_sample = st.selectbox("Select Ground-Truth Sample:", list(available_samples.keys())) file_path, matched_text = available_samples[chosen_sample] selected_audio_path, playable_wav_bytes, audio_dur = _read_file_to_bytes(file_path) if st.button("Sync Target Phrase to Sample"): st.session_state["target_phrase"] = matched_text st.rerun() st.caption(f"Expected Transcription: \"{matched_text}\"") else: st.info("Generate benchmark samples via: python -m ml.cli synth-data") st.markdown("---") st.markdown("### Baseline Calibration") st.caption("Upload a 5-second sample of your healthy voice to calibrate diagnostic thresholds:") normal_file = st.file_uploader("Healthy Baseline Sample (optional)", type=["wav", "mp3", "webm"], key="normal_uploader") normal_audio_path = None if normal_file: try: normal_audio_path, _, _ = _process_audio_bytes(normal_file.getvalue(), "normal_baseline.wav") except Exception: pass # Front-and-Center Target Sentence Input Card with st.container(border=True): st.markdown("### Target Phrase Configuration") st.caption("Select a single-word target or choose a standardized clinical reading passage:") # Quick Word Practice Presets st.markdown("

Single Word Evaluation Targets:

", unsafe_allow_html=True) w_cols = st.columns(len(WORD_PRESETS)) for idx, word in enumerate(WORD_PRESETS): with w_cols[idx]: if st.button(word, key=f"btn_word_{word}", use_container_width=True): st.session_state["target_phrase"] = word st.rerun() # Full Practice Sentences col_preset, col_btn = st.columns([3.5, 1]) with col_preset: preset_choice = st.selectbox("Clinical Reading Protocols:", ["(Custom Phrase)"] + list(SENTENCE_PRESETS.keys())) with col_btn: if preset_choice != "(Custom Phrase)": if st.button("Apply Protocol", use_container_width=True): st.session_state["target_phrase"] = SENTENCE_PRESETS[preset_choice] st.rerun() target_sentence = st.text_area( "Target Phrase (Expected Speech):", value=st.session_state["target_phrase"], key="target_phrase_input", height=60, ) st.session_state["target_phrase"] = target_sentence # Guard: Mandatory Target Phrase if not target_sentence.strip(): st.warning("Please configure a Target Phrase above to evaluate speech articulation.") return # Direct In-Page Audio Dropzone if nothing selected yet if not selected_audio_path or not playable_wav_bytes: with st.container(border=True): st.markdown("### Audio Acquisition Direct Upload") st.caption("Record using the left sidebar or drag-and-drop any audio clip (.wav, .mp3, .m4a, voice memo) right here:") direct_upload = st.file_uploader("Drop audio file here to diagnose", type=["wav", "mp3", "webm", "m4a", "ogg", "flac"], key="main_direct_upload") if direct_upload: try: selected_audio_path, playable_wav_bytes, audio_dur = _process_audio_bytes(direct_upload.getvalue(), direct_upload.name) st.rerun() except Exception as ex: st.error(f"Error loading file: {ex}") else: st.info("Record speech in the sidebar, drop an audio file above, or select a Benchmark Sample to run diagnostics.") return # Audio Playback and Acoustic Waveform Card with st.container(border=True): col_t1, col_t2 = st.columns([1.5, 1.5]) with col_t1: st.markdown("##### Reference Target") st.markdown(f"

\"{target_sentence.strip()}\"

", unsafe_allow_html=True) st.caption(f"Acoustic Duration: {audio_dur:.2f}s | Sample Rate: 16,000 Hz PCM") with col_t2: st.markdown("##### Audio Stream Playback") st.audio(playable_wav_bytes, format="audio/wav") st.markdown("##### Acoustic Energy & Waveform Envelope") _plot_waveform(selected_audio_path) # -------------------------------------------------------------------------- # LIVE STREAMING TELEMETRY HUD & DIAGNOSTICS EXECUTION # -------------------------------------------------------------------------- st.markdown("---") # Real-Time Telemetry Container telemetry_placeholder = st.empty() logs: List[str] = [] diag_res: Optional[Dict[str, Any]] = None if hasattr(engine, "diagnose_audio_stream"): try: stream_gen = engine.diagnose_audio_stream( audio_input=selected_audio_path, target_phrase=target_sentence, normal_calibration_audio=normal_audio_path, ) for item in stream_gen: step_num = item["step"] total_steps = item["total"] label = item["label"] detail = item["detail"] progress = item["progress"] elapsed = item["elapsed_ms"] log_entry = f"

[{elapsed:06.1f}ms] [Step {step_num}/{total_steps}] {label}: {detail}

" logs.append(log_entry) telemetry_html = f"""

LIVE DIAGNOSTIC SUBSYSTEM PIPELINE (STREAMING)

                                {step_num}/{total_steps} Expert Modules Executed
                            

{''.join(logs)}

""" telemetry_placeholder.markdown(telemetry_html, unsafe_allow_html=True) if "final_result" in item: diag_res = item["final_result"] time.sleep(0.04) except Exception: diag_res = engine.diagnose_audio( audio_input=selected_audio_path, target_phrase=target_sentence, normal_calibration_audio=normal_audio_path, ) else: with st.spinner("Executing neural diagnostic pipeline..."): diag_res = engine.diagnose_audio( audio_input=selected_audio_path, target_phrase=target_sentence, normal_calibration_audio=normal_audio_path, ) if not diag_res: diag_res = engine.diagnose_audio(selected_audio_path, target_sentence, normal_audio_path) # Handle Silence Guard if diag_res["is_silent"] or diag_res["decision"].get("is_silent"): st.warning("No Speech Detected: The audio recording is silent or below acoustic energy thresholds. Please speak clearly into your microphone.") return result = diag_res["decision"] pron = diag_res["pronunciation"] flaws_report = diag_res["flaws"] artic = diag_res["articulation"] p_stut = float(diag_res["stutter_probs"][1]) if (diag_res["stutter_probs"] and len(diag_res["stutter_probs"]) > 1) else 0.0 # Accuracy and Severity Calculations pron_acc = max(0.0, min(100.0, (1.0 - pron.get("wer", 0.0)) * 100.0)) overall_bucket = result["buckets"]["overall"].lower() fluency_score = int(result["fluency_100"]) if result.get("fluency_100") is not None else 100 # Length Mismatch Warning Notice if pron.get("length_warning"): st.markdown(f""" """, unsafe_allow_html=True) # Executive Diagnostic KPI Grid st.markdown(f"### Diagnostic Assessment Summary (Total Inference Latency: {diag_res['latency_ms']} ms)", unsafe_allow_html=True) c1, c2, c3, c4 = st.columns(4) sev_class_map = { "fluent": "sev-fluent", "mild": "sev-mild", "moderate": "sev-moderate", "severe": "sev-severe", "silent": "sev-silent", } badge_style = sev_class_map.get(overall_bucket, "sev-fluent") with c1: st.markdown(f"""

Clinical Stratification

{overall_bucket.upper()}

Fused Multi-Modal Assessment

""", unsafe_allow_html=True) with c2: st.markdown(f"""

Fluency Index

{fluency_score}/100

Continuous Cadence Rating

""", unsafe_allow_html=True) with c3: st.markdown(f"""

Pronunciation Accuracy

{pron_acc:.1f}%

Word Alignment Precision

""", unsafe_allow_html=True) with c4: st.markdown(f"""

Phonetic Goodness (GOP)

{pron.get('pron_score', 0) * 100:.1f}%

Sub-Word Acoustic Metric

""", unsafe_allow_html=True) # Transcription Display if pron.get("asr_hypothesis"): st.markdown(f""" """, unsafe_allow_html=True) # Comprehensive Pathology Diagnostic Grid st.markdown("---") st.markdown("### Specific Pathology Diagnostic Analysis") f_col1, f_col2 = st.columns(2) with f_col1: with st.container(border=True): st.markdown("#### Articulatory Sound Disorders ('R' & 'S' Checks)") # Rhotacism Check if flaws_report["has_r_flaw"]: for r_err in flaws_report["r_sound_issues"]: st.markdown(f"

Rhotacism Flaw: {r_err['message']}

", unsafe_allow_html=True) else: st.markdown("

'R' Sound Articulation: Accurate (No R->W/L substitution detected).

", unsafe_allow_html=True) # Sigmatism Check if flaws_report["has_s_flaw"]: for s_err in flaws_report["s_sound_issues"]: st.markdown(f"

Sigmatism Flaw: {s_err['message']}

", unsafe_allow_html=True) else: st.markdown("

'S' Sound Articulation: Accurate (No sibilant lisp or 'th' substitution detected).

", unsafe_allow_html=True) with f_col2: with st.container(border=True): st.markdown("#### Disfluency Flow & Phonation Acoustics") # Disfluency / Stutter Check if p_stut >= 0.78: st.markdown(f"

Disfluency Detected: Elevated probability of repetition/block ({p_stut*100:.1f}%)

", unsafe_allow_html=True) elif p_stut >= 0.60: st.markdown(f"

Mild Hesitation: Minor syllable repetition observed ({p_stut*100:.1f}%)

", unsafe_allow_html=True) else: st.markdown("

Fluency Cadence: Continuous speech flow (No disfluent events detected).

", unsafe_allow_html=True) # Vocal Phonation Check if flaws_report["voice_quality_issues"]: for v_err in flaws_report["voice_quality_issues"]: st.markdown(f"

Vocal Perturbation: {v_err}

", unsafe_allow_html=True) else: st.markdown(f"

Voice Quality: Healthy phonation (HNR: {artic.get('hnr_db', 0):.1f} dB, Jitter: {artic.get('jitter', 0)*100:.2f}%).

", unsafe_allow_html=True) # Word-by-Word Granular Phonetic Alignment Card st.markdown("---") st.markdown("### Granular Word-Level Pronunciation Alignment") with st.container(border=True): alignment = pron.get("alignment", []) # Build visual chips chips_html = '

' for item in alignment: status = item["status"] exp = item["expected"] spk = item["spoken"] if status == "correct": chips_html += f'[MATCH] {exp}' elif status == "substitution": chips_html += f'[DIFF] {exp} (heard: "{spk}")' elif status == "omission": chips_html += f'[UNSPOKEN] {exp}' elif status == "insertion": chips_html += f'[EXTRA] {spk}' chips_html += '

' st.markdown(chips_html, unsafe_allow_html=True) # Metrics Sub-Row m1, m2, m3, m4 = st.columns(4) with m1: st.metric("Pronunciation Accuracy", f"{pron_acc:.1f}%") with m2: st.metric("Phonetic Goodness (GOP)", f"{pron.get('pron_score', 0) * 100:.1f}%") with m3: st.metric("Word Error Rate (WER)", f"{pron.get('wer', 0) * 100:.1f}%") with m4: matched_words = max(0, pron.get("n_reference_words", 1) - pron.get("word_error", 0)) st.metric("Words Matched", f"{matched_words} / {pron.get('n_reference_words', 1)} target words") # Auditable Evidence Drawer with st.expander("Auditable Telemetry & Acoustic Evidence Trace"): st.json({ "clinical_decision": result, "pathology_flaws": flaws_report, "praat_phonation_metrics": artic, "pronunciation_accuracy_pct": pron_acc, "step_latencies_ms": diag_res.get("step_timings_ms", {}), "total_latency_ms": diag_res["latency_ms"], }) if __name__ == "__main__": main()