Update app.py

app.py

@@ -1,178 +1,316 @@
-# app.py
 import gradio as gr
-import numpy as np, soundfile as sf
-import librosa, scipy
-from pesq import pesq
-from pystoi import stoi
-from sklearn.ensemble import RandomForestRegressor
-import pyroomacoustics as pra
-
-# ------------- utility fns -------------
-def load_audio(path, sr=16000):
-    y, sr0 = sf.read(path)
-    if y.ndim>1:
-        y = np.mean(y,axis=1)
-    if sr0 != sr:
-        y = librosa.resample(y, orig_sr=sr0, target_sr=sr)
-    y = y - np.mean(y)
-    if np.max(np.abs(y))>0:
-        y = y / np.max(np.abs(y))
-    return y, sr
-
-def frame_audio(y, sr, win_ms=25, hop_ms=10):
-    win = int(win_ms*sr/1000)
-    hop = int(hop_ms*sr/1000)
-    frames = librosa.util.frame(y, frame_length=win, hop_length=hop).T
-    return frames, win, hop
-
-def hf_energy_db(frame, sr, low=4000):
-    S = np.abs(librosa.stft(frame, n_fft=1024, win_length=len(frame), center=False))
-    freqs = librosa.fft_frequencies(sr=sr, n_fft=1024)
-    mask = freqs >= low
-    if mask.sum()==0:
-        return -120.0
-    E = 20*np.log10(np.maximum(1e-12, np.mean(S[mask])))
-    return float(E)
-
-def frame_features(near_frame, far_frame, sr):
-    # spectral centroid, rms, zcr, hi-freq energy, coherence estimate via cross-spectrum
-    feats = {}
-    # rms
-    feats['rms_near'] = float(np.mean(near_frame**2)) if near_frame is not None else 0.0
-    feats['rms_far']  = float(np.mean(far_frame**2))
-    feats['centroid_near'] = float(np.mean(librosa.feature.spectral_centroid(y=near_frame, sr=sr))) if near_frame is not None else 0.0
-    feats['centroid_far']  = float(np.mean(librosa.feature.spectral_centroid(y=far_frame, sr=sr)))
-    feats['hi_near_db'] = hf_energy_db(near_frame, sr, low=4000) if near_frame is not None else -120.0
-    feats['hi_far_db']  = hf_energy_db(far_frame, sr, low=4000)
-    # basic coherence: compute magnitude-squared coherence using scipy.signal.coherence
-    try:
-        f, Cxy = scipy.signal.coherence(near_frame, far_frame, fs=sr, nperseg=min(len(near_frame),256))
-        feats['coherence_mean'] = float(np.mean(Cxy))
-    except:
-        feats['coherence_mean'] = 0.0
-    return feats
-
-# quick DRR proxy using energy early vs late (simple heuristic)
-def estimate_drr_from_pair(near, far, sr, early_ms=50):
-    # align roughly and compare early energy ratio (heuristic)
-    early = int(early_ms*sr/1000)
-    if len(near) < early or len(far) < early:
-        return 0.0
-    # direct energy proxy from near vs far first early segment
-    en_near = np.sum(near[:early]**2)
-    en_far  = np.sum(far[:early]**2)
-    # avoid div0
-    if en_far<=1e-12:
-        return 0.0
-    drr_db = 10*np.log10((en_near+1e-12)/(en_far+1e-12))
-    return float(drr_db)
-
-def normalize_metric(val, vmin, vmax):
-    return float(np.clip((val - vmin)/(vmax-vmin), 0, 1))
-
-# ------------- scoring pipeline -------------
-def score_pair(near_path, far_path):
-    sr = 16000
-    far, _ = load_audio(far_path, sr=sr)
-    near = None
-    if near_path:
-        near, _ = load_audio(near_path, sr=sr)
-
-    # global intrusive metrics (if near exists):
-    pesq_score = None
-    stoi_score = None
-    sisdr = None
-    if near is not None:
-        # align lengths
-        L = min(len(near), len(far))
+import librosa
+import numpy as np
+import pandas as pd
+from sklearn.cluster import KMeans, AgglomerativeClustering, DBSCAN
+from sklearn.metrics.pairwise import cosine_similarity
+from scipy.spatial.distance import jensenshannon
+from scipy.stats import pearsonr
+from scipy.signal import get_window as scipy_get_window
+import plotly.express as px
+import plotly.graph_objects as go
+import os
+import tempfile
+
+# ----------------------------
+# Segment Audio into Frames
+# ----------------------------
+def segment_audio(y, sr, frame_length_ms, hop_length_ms, window_type="hann"):
+    frame_length = int(frame_length_ms * sr / 1000)
+    hop_length = int(hop_length_ms * sr / 1000)
+    window = scipy_get_window(window_type if window_type != "rectangular" else "boxcar", frame_length)
+    frames = []
+    for i in range(0, len(y) - frame_length + 1, hop_length):
+        frame = y[i:i + frame_length] * window
+        frames.append(frame)
+    if frames:
+        frames = np.array(frames).T
+    else:
+        frames = np.zeros((frame_length, 1))
+    return frames, frame_length
+
+# ----------------------------
+# Feature Extraction
+# ----------------------------
+def extract_features_with_spectrum(frames, sr):
+    features = []
+    n_mfcc = 13
+    n_fft = min(2048, frames.shape[0])
+    for i in range(frames.shape[1]):
+        frame = frames[:, i]
+        if len(frame) < n_fft or np.max(np.abs(frame)) < 1e-10:
+            continue
+        feat = {}
         try:
-            pesq_score = pesq(sr, near[:L], far[:L], 'wb')
-        except:
-            pesq_score = None
+            feat["rms"] = float(np.mean(librosa.feature.rms(y=frame)[0]))
+        except: feat["rms"] = 0.0
         try:
-            stoi_score = stoi(near[:L], far[:L], sr, extended=False)
+            feat["spectral_centroid"] = float(np.mean(librosa.feature.spectral_centroid(y=frame, sr=sr)[0]))
+        except: feat["spectral_centroid"] = 0.0
+        try:
+            feat["zcr"] = float(np.mean(librosa.feature.zero_crossing_rate(frame)[0]))
+        except: feat["zcr"] = 0.0
+        try:
+            mfccs = librosa.feature.mfcc(y=frame, sr=sr, n_mfcc=n_mfcc, n_fft=n_fft)
+            for j in range(n_mfcc):
+                feat[f"mfcc_{j+1}"] = float(np.mean(mfccs[j]))
         except:
-            stoi_score = None
-        # sisdr quick: use pra.metrics
+            for j in range(n_mfcc):
+                feat[f"mfcc_{j+1}"] = 0.0
         try:
-            sisdr = float(pra.metrics.sdr(near[:L], far[:L])[0])
+            S = np.abs(librosa.stft(frame, n_fft=n_fft))
+            S_db = librosa.amplitude_to_db(S, ref=np.max)
+            freqs = librosa.fft_frequencies(sr=sr, n_fft=n_fft)
+            low_mask = freqs <= 2000
+            mid_mask = (freqs > 2000) & (freqs <= 4000)
+            high_mask = freqs > 4000
+            feat["low_freq_energy"] = float(np.mean(S_db[low_mask])) if np.any(low_mask) else 0.0
+            feat["mid_freq_energy"] = float(np.mean(S_db[mid_mask])) if np.any(mid_mask) else 0.0
+            feat["high_freq_energy"] = float(np.mean(S_db[high_mask])) if np.any(high_mask) else 0.0
+            feat["spectrum"] = S_db
         except:
-            sisdr = None
-
-    # frame-level features
-    frames_far, win, hop = frame_audio(far, sr)
-    frames_near = None
-    if near is not None:
-        near_cut = near[:len(far)]
-        frames_near, _, _ = frame_audio(near_cut, sr,)
-
-    feats = []
-    for i in range(len(frames_far)):
-        nf = frames_near[i] if frames_near is not None and i < len(frames_near) else None
-        ff = frames_far[i]
-        feats.append(frame_features(nf, ff, sr))
-
-    # aggregate metrics: example normalization ranges (you should tune)
-    # PESQ ~ [1..4.5], STOI [0..1], DRR [-20..20 dB], coherence [0..1], hi-loss in dB [-40..10]
-    q_pesq = normalize_metric(pesq_score if pesq_score is not None else 2.5, 1.0, 4.5)
-    q_stoi  = normalize_metric(stoi_score if stoi_score is not None else 0.5, 0.0, 1.0)
-    # DRR: estimate using early energy proxy between near&far across whole file
-    q_drr = 0.5
-    if near is not None:
-        drr = estimate_drr_from_pair(near, far, sr)
-        q_drr = normalize_metric(drr, -20, 20)
-    # hi-freq loss average
-    hi_loss = np.mean([f['hi_near_db'] - f['hi_far_db'] if 'hi_near_db' in f else 0.0 for f in feats])
-    q_hf = normalize_metric(-hi_loss, -40, 0)  # smaller loss -> higher score
-
-    # coherence average
-    q_coh = np.mean([f['coherence_mean'] for f in feats])
-
-    # example weighted aggregate (intrusive case)
-    if near is not None:
-        weights = {
-            'pesq':0.30, 'stoi':0.20, 'drr':0.20, 'hf':0.10, 'coh':0.20
-        }
-        score = (weights['pesq']*q_pesq + weights['stoi']*q_stoi + weights['drr']*q_drr + weights['hf']*q_hf + weights['coh']*q_coh) / sum(weights.values())
-    else:
-        # non-intrusive fallback: combine hf, coherence, centroid shift heuristics
-        avg_centroid_far = np.mean([f['centroid_far'] for f in feats])
-        q_centroid = normalize_metric(avg_centroid_far, 500, 3500)
-        score = (0.4*q_coh + 0.4*q_hf + 0.2*q_centroid)
-
-    percent = float(score*100)
-    # frames needing fix
-    frame_scores = []
-    for f in feats:
-        # example per-frame heuristic: combine coherence & hf loss
-        s = 0.6*f['coherence_mean'] + 0.4*normalize_metric( -(f['hi_near_db'] - f['hi_far_db']), -40,0 )
-        frame_scores.append(float(s))
-    problem_frames = [i for i,v in enumerate(frame_scores) if v < 0.5]
-
-    return {
-        "score_percent": percent,
-        "pesq": pesq_score,
-        "stoi": stoi_score,
-        "drr_db": drr if 'drr' in locals() else None,
-        "avg_coherence": q_coh,
-        "hi_loss_db": hi_loss,
-        "problem_frames": problem_frames
+            feat["low_freq_energy"] = feat["mid_freq_energy"] = feat["high_freq_energy"] = 0.0
+            feat["spectrum"] = np.zeros((n_fft // 2 + 1, 1))
+        features.append(feat)
+    if not features:
+        feat = { "rms": 0.0, "spectral_centroid": 0.0, "zcr": 0.0,
+                 "low_freq_energy": 0.0, "mid_freq_energy": 0.0, "high_freq_energy": 0.0,
+                 "spectrum": np.zeros((n_fft // 2 + 1, 1)) }
+        for j in range(n_mfcc): feat[f"mfcc_{j+1}"] = 0.0
+        features.append(feat)
+    return features
+
+# ----------------------------
+# Frame Comparison (core metrics)
+# ----------------------------
+def compare_frames_enhanced(near_feats, far_feats, metrics):
+    min_len = min(len(near_feats), len(far_feats))
+    if min_len == 0:
+        return pd.DataFrame({"frame_index": []})
+
+    results = {"frame_index": list(range(min_len))}
+    near_df = pd.DataFrame([f for f in near_feats[:min_len]])
+    far_df = pd.DataFrame([f for f in far_feats[:min_len]])
+    near_vec = near_df.drop(columns=["spectrum"], errors="ignore").values
+    far_vec = far_df.drop(columns=["spectrum"], errors="ignore").values
+
+    # Euclidean Distance
+    if "Euclidean Distance" in metrics:
+        results["euclidean_dist"] = np.linalg.norm(near_vec - far_vec, axis=1).tolist()
+    # Cosine Similarity
+    if "Cosine Similarity" in metrics:
+        cos_vals = []
+        for i in range(min_len):
+            a, b = near_vec[i].reshape(1, -1), far_vec[i].reshape(1, -1)
+            if np.all(a == 0) or np.all(b == 0):
+                cos_vals.append(0.0)
+            else:
+                cos_vals.append(float(cosine_similarity(a, b)[0][0]))
+        results["cosine_similarity"] = cos_vals
+    # High-Freq Loss Ratio (Quality)
+    if "High-Freq Loss Ratio" in metrics:
+        loss_ratios = []
+        for i in range(min_len):
+            near_high = near_feats[i]["high_freq_energy"]
+            far_high = far_feats[i]["high_freq_energy"]
+            ratio = max(0.0, 1.0 - abs(near_high - far_high) / (abs(near_high) + 1e-6))
+            loss_ratios.append(float(ratio))
+        results["high_freq_quality"] = loss_ratios
+
+    # 🔹 Energy Ratio
+    energy_ratio = []
+    for i in range(min_len):
+        near_rms = near_feats[i]["rms"]; far_rms = far_feats[i]["rms"]
+        ratio = (far_rms + 1e-6) / (near_rms + 1e-6)
+        energy_ratio.append(float(np.clip(ratio, 0, 1)))
+    results["energy_ratio"] = energy_ratio
+
+    # 🔹 Clarity Ratio
+    clarity_ratio = []
+    for i in range(min_len):
+        near_low, near_high = near_feats[i]["low_freq_energy"], near_feats[i]["high_freq_energy"]
+        far_low, far_high = far_feats[i]["low_freq_energy"], far_feats[i]["high_freq_energy"]
+        near_ratio, far_ratio = (near_low - near_high), (far_low - far_high)
+        diff = 1 - abs(far_ratio - near_ratio) / (abs(near_ratio) + 1e-6)
+        clarity_ratio.append(np.clip(diff, 0, 1))
+    results["clarity_ratio"] = clarity_ratio
+
+    # 🔹 Spectral Overlap
+    overlap_scores = []
+    for i in range(min_len):
+        near_spec = near_feats[i]["spectrum"].flatten()
+        far_spec = far_feats[i]["spectrum"].flatten()
+        if np.all(near_spec == 0) or np.all(far_spec == 0):
+            overlap_scores.append(0.0)
+        else:
+            overlap = float(cosine_similarity(near_spec.reshape(1, -1), far_spec.reshape(1, -1))[0][0])
+            overlap_scores.append(overlap)
+    results["spectral_overlap"] = overlap_scores
+
+    # 🔹 Combined Weighted Quality
+    weights = {
+        "cosine_similarity": 0.3,
+        "high_freq_quality": 0.25,
+        "energy_ratio": 0.2,
+        "clarity_ratio": 0.15,
+        "spectral_overlap": 0.1
     }
+    combined_quality = []
+    for i in range(min_len):
+        val = sum(results[k][i] * w for k, w in weights.items() if k in results)
+        combined_quality.append(float(val / sum(weights.values())))
+    results["combined_quality"] = combined_quality
+    return pd.DataFrame(results)
+
+# ----------------------------
+# Clustering + Overlay
+# ----------------------------
+def cluster_frames_custom(features_df, cluster_features, algo, n_clusters=5, eps=0.5):
+    if not cluster_features:
+        raise gr.Error("Please select at least one feature for clustering.")
+    if len(features_df) == 0:
+        features_df["cluster"] = []
+        return features_df
+    X = features_df[cluster_features].values
+    if algo == "KMeans":
+        n_clusters = min(n_clusters, len(X))
+        model = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
+        labels = model.fit_predict(X)
+    elif algo == "Agglomerative":
+        n_clusters = min(n_clusters, len(X))
+        model = AgglomerativeClustering(n_clusters=n_clusters)
+        labels = model.fit_predict(X)
+    elif algo == "DBSCAN":
+        model = DBSCAN(eps=eps, min_samples=min(3, len(X)))
+        labels = model.fit_predict(X)
+    else:
+        raise ValueError("Unknown clustering algorithm")
+    features_df = features_df.copy()
+    features_df["cluster"] = labels
+    return features_df
+
+def plot_spectral_difference(near_feats, far_feats, frame_idx=0):
+    if not near_feats or not far_feats or frame_idx >= len(near_feats) or frame_idx >= len(far_feats):
+        fig = go.Figure(); fig.update_layout(title="No data available"); return fig
+    near_spec = near_feats[frame_idx]["spectrum"]; far_spec = far_feats[frame_idx]["spectrum"]
+    min_freq_bins = min(near_spec.shape[0], far_spec.shape[0])
+    diff = near_spec[:min_freq_bins] - far_spec[:min_freq_bins]
+    fig = go.Figure(data=go.Heatmap(z=diff, colorscale='RdBu', zmid=0))
+    fig.update_layout(title=f"Spectral Difference (Frame {frame_idx})", height=300)
+    return fig
+
+def plot_cluster_overlay(df, cluster_metric, overlay_metric):
+    if cluster_metric not in df.columns or overlay_metric not in df.columns:
+        fig = go.Figure(); fig.update_layout(title="Metrics not found"); return fig
+    fig = px.scatter(df, x=cluster_metric, y=overlay_metric, color=overlay_metric,
+                     color_continuous_scale='Viridis',
+                     title=f"Cluster Overlay: {cluster_metric} vs {overlay_metric}")
+    fig.update_layout(height=400)
+    return fig
+
+# ----------------------------
+# Main Analysis Function
+# ----------------------------
+def analyze_audio_pair(
+    near_file, far_file,
+    frame_length_ms, hop_length_ms, window_type,
+    comparison_metrics, cluster_features, clustering_algo, n_clusters, dbscan_eps
+):
+    if not near_file or not far_file:
+        raise gr.Error("Upload both audio files.")
+    try:
+        y_near, sr_near = librosa.load(near_file.name, sr=None)
+        y_far, sr_far = librosa.load(far_file.name, sr=None)
+    except Exception as e:
+        raise gr.Error(f"Error loading audio: {str(e)}")
+    if sr_near != sr_far:
+        y_far = librosa.resample(y_far, orig_sr=sr_far, target_sr=sr_near)
+        sr = sr_near
+    else:
+        sr = sr_near
+    frames_near, _ = segment_audio(y_near, sr, frame_length_ms, hop_length_ms, window_type)
+    frames_far, _ = segment_audio(y_far, sr, frame_length_ms, hop_length_ms, window_type)
+    near_feats = extract_features_with_spectrum(frames_near, sr)
+    far_feats = extract_features_with_spectrum(frames_far, sr)
+    comparison_df = compare_frames_enhanced(near_feats, far_feats, comparison_metrics)
+    near_df = pd.DataFrame(near_feats).drop(columns=["spectrum"], errors="ignore")
+    clustered_df = cluster_frames_custom(near_df, cluster_features, clustering_algo, n_clusters, dbscan_eps)
+    # Plots
+    metric_cols = [col for col in comparison_df.columns if col != "frame_index"]
+    plot_comparison = px.line(comparison_df, x="frame_index", y=metric_cols[0],
+                              title=f"{metric_cols[0].replace('_',' ').title()} Over Time") if metric_cols else px.line()
+    if len(cluster_features) >= 2 and len(clustered_df) > 0:
+        x_feat, y_feat = cluster_features[0], cluster_features[1]
+        plot_scatter = px.scatter(clustered_df, x=x_feat, y=y_feat, color="cluster",
+                                  title=f"Clustering: {x_feat} vs {y_feat}")
+    else:
+        plot_scatter = px.scatter(title="Select ≥2 features for clustering")
+    spec_heatmap = plot_spectral_difference(near_feats, far_feats, frame_idx=0)
+    overlay_fig = plot_cluster_overlay(clustered_df, cluster_features[0], "combined_quality")
+    return plot_comparison, comparison_df, plot_scatter, clustered_df, spec_heatmap, overlay_fig
+
+def export_results(comparison_df, clustered_df):
+    temp_dir = tempfile.mkdtemp()
+    comp_path = os.path.join(temp_dir, "frame_comparisons.csv")
+    cluster_path = os.path.join(temp_dir, "clustered_frames.csv")
+    comparison_df.to_csv(comp_path, index=False)
+    clustered_df.to_csv(cluster_path, index=False)
+    return [comp_path, cluster_path]
+
+# ----------------------------
+# Gradio UI
+# ----------------------------
+dummy_features = ["rms", "spectral_centroid", "zcr"] + [f"mfcc_{i}" for i in range(1,14)] + \
+                 ["low_freq_energy", "mid_freq_energy", "high_freq_energy"]
+
+with gr.Blocks(title="Advanced Near vs Far Field Analyzer") as demo:
+    gr.Markdown("# 🎙️ Advanced Near vs Far Field Speech Analyzer")
+    with gr.Row():
+        near_file = gr.File(label="Near-Field Audio (.wav)", file_types=[".wav"])
+        far_file = gr.File(label="Far-Field Audio (.wav)")
+
+    with gr.Accordion("⚙️ Frame Settings", open=True):
+        frame_length_ms = gr.Slider(10, 500, value=50, step=1, label="Frame Length (ms)")
+        hop_length_ms = gr.Slider(1, 250, value=25, step=1, label="Hop Length (ms)")
+        window_type = gr.Dropdown(["hann", "hamming", "rectangular"], value="hann", label="Window Type")
+
+    with gr.Accordion("📊 Comparison Metrics", open=True):
+        comparison_metrics = gr.CheckboxGroup(
+            choices=[
+                "Euclidean Distance", "Cosine Similarity", "High-Freq Loss Ratio"
+            ],
+            value=["Cosine Similarity", "High-Freq Loss Ratio"],
+            label="Select Metrics"
+        )
+
+    with gr.Accordion("🧩 Clustering Configuration", open=True):
+        cluster_features = gr.CheckboxGroup(
+            choices=dummy_features, value=["rms", "spectral_centroid", "high_freq_energy"],
+            label="Features for Clustering")
+        clustering_algo = gr.Radio(["KMeans", "Agglomerative", "DBSCAN"], value="KMeans", label="Clustering Algorithm")
+        n_clusters = gr.Slider(2, 20, value=5, step=1, label="Clusters (for KMeans/Agglomerative)")
+        dbscan_eps = gr.Slider(0.1, 2.0, value=0.5, step=0.1, label="DBSCAN eps")
+
+    btn = gr.Button("🚀 Analyze")
+
+    with gr.Tabs():
+        with gr.Tab("📈 Frame Comparison"):
+            comp_plot = gr.Plot(); comp_table = gr.Dataframe()
+        with gr.Tab("🧩 Clustering"):
+            cluster_plot = gr.Plot(); cluster_table = gr.Dataframe()
+        with gr.Tab("🔍 Spectral Analysis"):
+            spec_heatmap = gr.Plot(label="Spectral Difference (Near - Far)")
+        with gr.Tab("🧭 Metric Overlay"):
+            overlay_plot = gr.Plot(label="Metric Overlay")
+
+    with gr.Tab("📤 Export"):
+        export_btn = gr.Button("💾 Download CSVs"); export_files = gr.Files()
+
+    btn.click(fn=analyze_audio_pair,
+              inputs=[near_file, far_file, frame_length_ms, hop_length_ms, window_type,
+                      comparison_metrics, cluster_features, clustering_algo, n_clusters, dbscan_eps],
+              outputs=[comp_plot, comp_table, cluster_plot, cluster_table, spec_heatmap, overlay_plot])
+    export_btn.click(fn=export_results, inputs=[comp_table, cluster_table], outputs=export_files)
 
-# ------------- Gradio UI -------------
-def analyze(near, far):
-    res = score_pair(near.name if near else None, far.name)
-    html = f"<h3>Far-field quality: {res['score_percent']:.1f}%</h3>"
-    html += "<ul>"
-    html += f"<li>PESQ: {res['pesq']}</li>"
-    html += f"<li>STOI: {res['stoi']}</li>"
-    html += f"<li>DRR (proxy, dB): {res['drr_db']}</li>"
-    html += f"<li>Avg coherence: {res['avg_coherence']:.3f}</li>"
-    html += f"<li>Avg high-freq loss (dB): {res['hi_loss_db']:.2f}</li>"
-    html += f"<li>Problem frames (indices): {res['problem_frames']}</li>"
-    html += "</ul>"
-    return html
-
-iface = gr.Interface(fn=analyze, inputs=[gr.File(label="Near (optional)"), gr.File(label="Far")], outputs=gr.HTML, title="Far-field degradation score")
 if __name__ == "__main__":
-    iface.launch()
+    demo.launch()