app.py

"""Language Attrition: derive linguistic factors, then model them.

Two tabs, both point-and-click (no Spanish, no Praat, no code needed):
  1. Derive   — upload a recording, get its ~25 linguistic factors, stack them.
  2. Model    — pick two variables, see the scatter, correlation and regression.

The numbers are identical to the course pipeline (same code under the hood).
Phase-2 markers (VOT, vowel space, rhythm) need forced alignment and are not
produced here.
"""
import os
import tempfile

# Workaround for a gradio_client 1.3.0 bug (shipped with gradio 4.44.1): its
# JSON-schema walker assumes every node is a dict, but components can emit a
# bare boolean node (e.g. `additionalProperties: true`). That crashes API-info
# generation on page load with "TypeError: argument of type 'bool' is not
# iterable". Guard the walker so boolean nodes resolve to "Any".
import gradio_client.utils as _gcu

_orig_json_schema_to_python_type = _gcu._json_schema_to_python_type


def _safe_json_schema_to_python_type(schema, defs=None):
    if isinstance(schema, bool):
        return "Any"
    return _orig_json_schema_to_python_type(schema, defs)


_gcu._json_schema_to_python_type = _safe_json_schema_to_python_type

import gradio as gr
import librosa
import numpy as np
import pandas as pd
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from scipy import stats

from features import derive_features_with_text, FEATURE_COLUMNS

MAX_AUDIO_SECONDS = 600
PRIMARY, ACCENT = "#004782", "#d1495b"
ALL_COLS = ["Speaker"] + FEATURE_COLUMNS


# ----------------------------------------------------------------------------
# Tab 1: Derive
# ----------------------------------------------------------------------------
def _to_df(rows):
    if not rows:
        return pd.DataFrame(columns=ALL_COLS)
    return pd.DataFrame(rows)[ALL_COLS]


def _save_csv(df, prefix="features"):
    f = tempfile.NamedTemporaryFile(mode="w", suffix=".csv", prefix=f"{prefix}_", delete=False)
    df.to_csv(f.name, index=False)
    f.close()
    return f.name


def add_recording(audio_path, transcript_path, label, rows):
    rows = rows or []
    if audio_path is None:
        return rows, _to_df(rows), None, "", "Please upload an audio file first."
    if transcript_path is None:
        return rows, _to_df(rows), None, "", "Please upload the transcript (.json) for this speaker."
    dur = librosa.get_duration(path=audio_path)
    if dur > MAX_AUDIO_SECONDS:
        return rows, _to_df(rows), None, "", f"Audio too long ({dur:.0f}s). Max is {MAX_AUDIO_SECONDS}s."
    try:
        row, text = derive_features_with_text(audio_path, transcript_path, (label or "").strip() or None)
    except Exception as e:  # noqa: BLE001
        return rows, _to_df(rows), None, "", f"Error while processing: {e}"
    rows = [r for r in rows if r["Speaker"] != row["Speaker"]] + [row]
    df = _to_df(rows)
    return rows, df, _save_csv(df), text, (
        f"Added **{row['Speaker']}**. Table now has {len(rows)} recording(s). "
        "Switch to the *Model* tab when you have a few."
    )


def clear_table():
    return [], _to_df([]), None, "", "Table cleared."


# ----------------------------------------------------------------------------
# Tab 2: Model
# ----------------------------------------------------------------------------
def load_for_modeling(rows, uploaded, source):
    if source.startswith("Upload") and uploaded is not None:
        try:
            df = pd.read_csv(uploaded)
        except Exception as e:  # noqa: BLE001
            return None, gr.update(choices=[]), gr.update(choices=[]), f"Could not read CSV: {e}"
    else:
        df = _to_df(rows or [])
    num = [c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])]
    if len(num) < 2:
        return df, gr.update(choices=num), gr.update(choices=num), (
            "Need at least two numeric columns. Derive a few recordings first, "
            "or upload a CSV that includes your predictor columns."
        )
    return (df,
            gr.update(choices=num, value=num[0]),
            gr.update(choices=num, value=num[1]),
            f"Loaded {len(df)} rows and {len(num)} numeric variables. Pick two to compare.")


def _interpret(r, p, n):
    strength = "weak" if abs(r) < 0.3 else "moderate" if abs(r) < 0.6 else "strong"
    direction = "positive" if r > 0 else "negative"
    sig = ("**statistically significant** (p < 0.05)" if p < 0.05
           else "**not significant** (p ≥ 0.05) — could be chance, especially with few speakers")
    return (f"_Reading: a {strength} {direction} relationship, {sig}._\n\n"
            "With a small number of speakers this is a **hypothesis, not a finding**.")


def model_pair(df, x, y, method):
    if df is None or not len(df) or x is None or y is None:
        return None, "Load a table and pick two variables."
    if x == y:
        return None, "Pick two *different* variables."
    sub = df[[x, y]].apply(pd.to_numeric, errors="coerce").dropna()
    n = len(sub)
    if n < 3:
        return None, "Need at least 3 speakers with both values present."
    pear = stats.pearsonr(sub[x], sub[y])
    spear = stats.spearmanr(sub[x], sub[y])
    b, a = np.polyfit(sub[x], sub[y], 1)
    use_pear = method.startswith("Pearson")
    r, p = (pear if use_pear else spear)

    fig, ax = plt.subplots(figsize=(6.2, 4.6))
    ax.scatter(sub[x], sub[y], s=85, color=PRIMARY, edgecolor="white", zorder=3)
    xs = np.array([sub[x].min(), sub[x].max()])
    ax.plot(xs, a + b * xs, color=ACCENT, lw=2)
    ax.set_xlabel(x); ax.set_ylabel(y); ax.grid(alpha=0.3)
    ax.set_title(f"{method.split()[0]} = {r:+.2f}   (p = {p:.3f}, n = {n})", color=PRIMARY)
    fig.tight_layout()

    md = (f"### {x}  vs  {y}\n\n"
          f"- **{method.split()[0]} correlation** = {r:+.3f}  ·  p = {p:.3f}  ·  n = {n}\n"
          f"- **Regression line**: `{y} = {a:.2f} + {b:.3f} × {x}`\n"
          f"- Pearson r = {pear[0]:+.3f} (p={pear[1]:.3f}) · Spearman ρ = {spear[0]:+.3f} (p={spear[1]:.3f})\n\n"
          + _interpret(r, p, n))
    return fig, md


def corr_heatmap(df, method):
    if df is None or not len(df):
        return None, "Load a table first."
    num = df.select_dtypes("number")
    if num.shape[1] < 2:
        return None, "Need at least two numeric columns."
    C = num.corr(method="spearman" if method.startswith("Spearman") else "pearson")
    sz = min(1.2 + 0.5 * len(C), 13)
    fig, ax = plt.subplots(figsize=(sz, sz))
    im = ax.imshow(C, vmin=-1, vmax=1, cmap="RdBu_r")
    ax.set_xticks(range(len(C))); ax.set_xticklabels(C.columns, rotation=90, fontsize=7)
    ax.set_yticks(range(len(C))); ax.set_yticklabels(C.columns, fontsize=7)
    ax.set_title(f"{method.split()[0]} correlations across all variables", color=PRIMARY)
    fig.colorbar(im, shrink=0.7, label="r")
    fig.tight_layout()
    return fig, f"Heatmap of {num.shape[1]} variables. Blue = move together, red = move apart."


# ----------------------------------------------------------------------------
# UI
# ----------------------------------------------------------------------------
with gr.Blocks(title="Language Attrition: derive & model") as demo:
    gr.Markdown(
        "# Language Attrition: derive the numbers, then model them\n"
        "Upload a recording **and its transcript** and the app derives the linguistic "
        "factors (disfluency, fluency, complexity, pitch and voice quality). Then compare "
        "any two of them."
    )
    table_state = gr.State([])     # derived rows
    model_df_state = gr.State()    # dataframe loaded into the Model tab

    with gr.Tab("1 · Derive features"):
        with gr.Row():
            with gr.Column():
                audio = gr.Audio(type="filepath", label="Recording (≤ 10 min)")
                transcript_in = gr.File(label="Transcript (.json, word-timestamped)",
                                        file_types=[".json"])
                label = gr.Textbox(label="Speaker label", placeholder="e.g. A014 (optional)")
                with gr.Row():
                    add_btn = gr.Button("Derive features", variant="primary")
                    clear_btn = gr.Button("Clear table")
                status = gr.Markdown()
            with gr.Column():
                transcript_box = gr.Textbox(label="Transcript (from your upload)", lines=6)
        table = gr.Dataframe(label="Your feature table", interactive=False, wrap=True)
        csv_out = gr.File(label="Download feature table (.csv)")
        add_btn.click(add_recording, [audio, transcript_in, label, table_state],
                      [table_state, table, csv_out, transcript_box, status])
        clear_btn.click(clear_table, None,
                        [table_state, table, csv_out, transcript_box, status])

    with gr.Tab("2 · Model"):
        gr.Markdown(
            "Model the table you built in tab 1, **or** upload a CSV "
            "(e.g. the features joined with your questionnaire predictors)."
        )
        with gr.Row():
            source = gr.Radio(["Use table from tab 1", "Upload a CSV"],
                              value="Use table from tab 1", label="Data source")
            uploaded = gr.File(label="CSV (if uploading)", file_types=[".csv"])
        load_btn = gr.Button("Load data", variant="primary")
        load_status = gr.Markdown()
        with gr.Row():
            x_var = gr.Dropdown(label="X variable", choices=[])
            y_var = gr.Dropdown(label="Y variable", choices=[])
            method = gr.Radio(["Pearson (straight-line)", "Spearman (rank)"],
                              value="Spearman (rank)", label="Correlation type")
        with gr.Row():
            plot_btn = gr.Button("Plot & correlate", variant="primary")
            heat_btn = gr.Button("Correlation heatmap (all variables)")
        with gr.Row():
            plot_out = gr.Plot(label="Scatter + regression line")
            result_md = gr.Markdown()
        heat_out = gr.Plot(label="Correlation heatmap")

        load_btn.click(load_for_modeling, [table_state, uploaded, source],
                       [model_df_state, x_var, y_var, load_status])
        plot_btn.click(model_pair, [model_df_state, x_var, y_var, method],
                       [plot_out, result_md])
        heat_btn.click(corr_heatmap, [model_df_state, method], [heat_out, load_status])

if __name__ == "__main__":
    demo.queue(max_size=60, default_concurrency_limit=2).launch(share=True)