app.py

import gradio as gr
import os
import io
import re
import random
import librosa
import soundfile as sf
import pandas as pd
from gradio_client import Client, handle_file
from transformers import pipeline
from datasets import load_dataset, Audio
from gradio_client import Client
from stats_data import get_indomain_breakdown, get_experimental_summary, SPEAKER_META

# 1. Initialize Baseline ASR (Strict English, Repetition Penalty 3.0)
print("Initializing Whisper Tiny Baseline...")
whisper_asr = pipeline(
    "automatic-speech-recognition", 
    model="openai/whisper-tiny",
    generate_kwargs={
        "language": "en", 
        "task": "transcribe", 
        "repetition_penalty": 3.0
    }
)

# Configuration from Environment Variables
HF_TOKEN = os.getenv("HF_TOKEN")
PRIVATE_BACKEND_URL = "st192011/Torgo-DSR-Private"

def normalize_text(text):
    if not text: return ""
    # Remove punctuation and lowercase
    text = re.sub(r'[^\w\s]', '', text).lower().strip()
    return " ".join(text.split())

def format_audio(audio_path):
    """Ensures audio is 16kHz mono to match ASR training conditions."""
    y, sr = librosa.load(audio_path, sr=16000)
    out_path = "formatted_input.wav"
    sf.write(out_path, y, sr)
    return out_path

# --- Logic: Data Loading ---
def get_sample_logic(speaker_id):
    try:
        if "UA" in speaker_id:
            # UA-Speech Access (Direct pull for F02)
            dataset = load_dataset("resproj007/uaspeech_female", split="train", streaming=True)
            dataset = dataset.cast_column("audio", Audio(decode=False))
            # UA is small, skip slightly for variety
            sample = next(iter(dataset.skip(random.randint(0, 30))))
            gt_text = sample.get('text') or sample.get('transcription') or sample.get('sentence')
        else:
            # Torgo Access (Manual filtering as per Colab fix)
            dataset = load_dataset("abnerh/TORGO-database", split="train", streaming=True)
            dataset = dataset.cast_column("audio", Audio(decode=False))
            
            def filter_spk(x):
                sid = str(x.get('speaker_id', '')).upper()
                if not sid or sid == "NONE":
                    sid = os.path.basename(x['audio']['path']).split('_')[0].upper()
                return sid == speaker_id
            
            speaker_ds = dataset.filter(filter_spk)
            sample = next(iter(speaker_ds.shuffle(buffer_size=10)))
            gt_text = sample.get('transcription') or sample.get('text')

        # Decode Bytes manually to bypass torchcodec errors
        audio_bytes = sample['audio']['bytes']
        audio_data, sr = librosa.load(io.BytesIO(audio_bytes), sr=16000)
        temp_path = "dataset_sample.wav"
        sf.write(temp_path, audio_data, sr)
        
        return temp_path, gt_text.lower().strip(), SPEAKER_META[speaker_id]
    except Exception as e:
        return None, f"Dataset Error: {e}", {}

# --- Logic: Model Processing ---
def process_audio_step_1(audio_path):
    """Runs Whisper Baseline and returns normalized text."""
    if not audio_path: return "No audio loaded", ""
    
    # Pre-process audio format to 16k
    formatted_path = format_audio(audio_path)
    
    # Run Whisper
    result = whisper_asr(formatted_path)
    raw_w = result["text"]
    norm_w = normalize_text(raw_w)
    return raw_w, norm_w

def process_audio_step_2(audio_path, norm_whisper):
    """Sends audio + normalized whisper to the Private Model API."""
    if not audio_path or not norm_whisper: 
        return "Please load data and run Whisper (Step 1) first."
    
    try:
        # Connect to the private API
        client = Client(PRIVATE_BACKEND_URL, token=HF_TOKEN)
        
        # FIX: Wrap audio_path with handle_file()
        # This sends the metadata required by Pydantic ('gradio.FileData')
        prediction = client.predict(
            audio_path=handle_file(audio_path), 
            whisper_norm=norm_whisper, 
            api_name="/predict_dsr"
        )
        return prediction
    except Exception as e:
        return f"Backend Connection Required. Details: {e}"

# --- UI Construction ---
with gr.Blocks(theme=gr.themes.Soft(), title="Torgo DSR Lab") as demo:
    gr.Markdown("# ⚗️ Torgo DSR Lab")
    gr.Markdown("Neural Reconstruction Layer for Torgo and UA-Speech Zero-Shot.")
    
    # Hidden state to store the path of the currently active audio
    active_audio_path = gr.State("")

    with gr.Tab("🔬 Laboratory"):
        with gr.Row():
            # LEFT COLUMN: Data Input
            with gr.Column(scale=1):
                with gr.Group():
                    gr.Markdown("### Channel A: Research Datasets")
                    speaker_input = gr.Dropdown(sorted(list(SPEAKER_META.keys())), label="Select Speaker Profile", value="F01")
                    load_btn = gr.Button("Load Sample from Dataset")
                    gt_box = gr.Textbox(label="Ground Truth (Reference)", interactive=False)
                    meta_display = gr.JSON(label="Speaker Metadata")
                
                gr.Markdown("---")
                
                with gr.Group():
                    gr.Markdown("### Channel B: Personal Input")
                    user_audio = gr.Audio(sources=["microphone", "upload"], type="filepath", label="Record or Upload Audio")
                    user_load_btn = gr.Button("Use This Audio")

            # RIGHT COLUMN: Transcripts
            with gr.Column(scale=2):
                gr.Markdown("### Analysis & Reconstruction")
                
                with gr.Group():
                    gr.Markdown("#### Step 1: ASR Baseline")
                    whisper_btn = gr.Button("Run Whisper Tiny")
                    w_raw = gr.Textbox(label="Whisper Raw Transcript")
                    w_norm = gr.Textbox(label="Whisper Normalized")
                
                gr.Markdown("---")
                
                with gr.Group():
                    gr.Markdown("#### Step 2: Neural Reconstruction")
                    model_btn = gr.Button("Run Our Correction Model", variant="primary")
                    final_out = gr.Textbox(label="DSR Lab Prediction (5K Model)")

    with gr.Tab("📊 Research Statistics"):
        gr.Markdown("# 🔬 Performance Evaluation")
        
        with gr.Row():
            with gr.Column():
                gr.Markdown("""
                ### 📏 Metric: Exact Match Accuracy
                Accuracy is the percentage of samples where the **normalized prediction** (lowercase, no punctuation) exactly matches the **normalized ground truth**.
                """)
            
            with gr.Column():
                gr.Markdown("""
                ### 🧪 Model Definitions
                * **5K Pure Model:** Trained on real-world Torgo articulatory distortions. Optimized for phonetic fidelity.
                * **10K Triple-Mix Model:** Includes synthetic data and anchors; utilized for generalization (LOSO) testing.
                """)

        gr.Markdown("---")
        gr.Markdown("## 1. Torgo In-Domain Analysis (By Speaker)")
        gr.DataFrame(get_indomain_breakdown())
        
        gr.Markdown("## 2. Experimental Milestone Summary")
        gr.DataFrame(get_experimental_summary())

        gr.Markdown("""
        ### 🔍 Key Discovery: The Acoustic Floor
        Our research found that the **5K Pure Model** achieved higher accuracy in both in-domain and zero-shot tasks. This suggests an **'Acoustic Floor'** exists where real-world phonetic distortions are more valuable for model grounding than synthetic linguistic diversity.
        """)

    # --- Event Handlers ---

    # Dataset Channel: Load -> Update State -> Update UI Text/Meta
    load_btn.click(
        get_sample_logic, 
        inputs=speaker_input, 
        outputs=[active_audio_path, gt_box, meta_display]
    )

    # Personal Channel: Use Audio -> Update State -> Clear Reference
    user_load_btn.click(
        lambda x: (x, "User Recorded (No Ground Truth)", {"Dataset": "Custom", "Severity": "N/A"}), 
        inputs=user_audio, 
        outputs=[active_audio_path, gt_box, meta_display]
    )

    # Step 1: Whisper (Uses State)
    whisper_btn.click(
        process_audio_step_1, 
        inputs=active_audio_path, 
        outputs=[w_raw, w_norm]
    )

    # Step 2: Model (Uses State + Whisper result)
    model_btn.click(
        process_audio_step_2, 
        inputs=[active_audio_path, w_norm], 
        outputs=final_out
    )

demo.launch()