Update app.py

app.py

@@ -2,93 +2,183 @@ import gradio as gr
 import librosa
 import numpy as np
 import pandas as pd
-from sklearn.cluster import KMeans
-from sklearn.metrics.pairwise import cosine_similarity
+from sklearn.cluster import KMeans, AgglomerativeClustering, DBSCAN
+from sklearn.metrics.pairwise import cosine_similarity, euclidean_distances
+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
-from scipy.signal import get_window as scipy_get_window  # ✅ Fix here
 
 # ----------------------------
-# Core Functions
+# Enhanced Feature Extraction (with spectral bins)
 # ----------------------------
 
-def load_audio(file_path):
-    y, sr = librosa.load(file_path, sr=None)
-    return y, sr
-
-def segment_audio(y, sr, frame_length_ms, hop_length_ms, window_type):
-    frame_length = int(frame_length_ms * sr / 1000)
-    hop_length = int(hop_length_ms * sr / 1000)
-    if frame_length > len(y):
-        raise ValueError("Frame length is longer than audio duration.")
-    frames = librosa.util.frame(y, frame_length=frame_length, hop_length=hop_length)
-    
-    if window_type != "rectangular":
-        # ✅ Use scipy.signal.get_window instead of librosa.get_window
-        window = scipy_get_window(window_type, frame_length)
-        frames = frames * window[:, None]
-    return frames, sr
-
-def extract_features(frames, sr):
+def extract_features_with_spectrum(frames, sr):
     features = []
     n_mfcc = 13
+    n_fft = 2048
+
     for i in range(frames.shape[1]):
         frame = frames[:, i]
         feat = {}
 
-        # RMS Energy
+        # Basic features
         rms = np.mean(librosa.feature.rms(y=frame)[0])
         feat["rms"] = float(rms)
 
-        # Spectral Centroid
         sc = np.mean(librosa.feature.spectral_centroid(y=frame, sr=sr)[0])
         feat["spectral_centroid"] = float(sc)
 
-        # Zero Crossing Rate
         zcr = np.mean(librosa.feature.zero_crossing_rate(frame)[0])
         feat["zcr"] = float(zcr)
 
-        # MFCCs (1-13)
         mfccs = librosa.feature.mfcc(y=frame, sr=sr, n_mfcc=n_mfcc)
         for j in range(n_mfcc):
             feat[f"mfcc_{j+1}"] = float(np.mean(mfccs[j]))
 
+        # Spectral bins for lost frequencies
+        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)
+
+        # Split spectrum: low (<2kHz), mid (2-4kHz), high (>4kHz)
+        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]))
+        feat["mid_freq_energy"] = float(np.mean(S_db[mid_mask]))
+        feat["high_freq_energy"] = float(np.mean(S_db[high_mask]))
+
+        # Store full spectrum for later (optional)
+        feat["spectrum"] = S_db  # will be used for heatmap
+
         features.append(feat)
-    return pd.DataFrame(features)
+    return features
 
-def compare_frames(near_df, far_df, metrics):
-    min_len = min(len(near_df), len(far_df))
-    near = near_df.iloc[:min_len].reset_index(drop=True)
-    far = far_df.iloc[:min_len].reset_index(drop=True)
-    
+def compare_frames_enhanced(near_feats, far_feats, metrics):
+    min_len = min(len(near_feats), len(far_feats))
     results = {"frame_index": list(range(min_len))}
     
+    # Prepare vectors
+    near_df = pd.DataFrame([f for f in near_feats[:min_len]])
+    far_df = pd.DataFrame([f for f in far_feats[:min_len]])
+
+    # Remove non-numeric columns
+    near_vec = near_df.drop(columns=["spectrum"], errors="ignore").values
+    far_vec = far_df.drop(columns=["spectrum"], errors="ignore").values
+
+    # 1. Euclidean Distance
     if "Euclidean Distance" in metrics:
-        euclidean = np.linalg.norm(near.values - far.values, axis=1)
-        results["euclidean_dist"] = euclidean.tolist()
-    
+        results["euclidean_dist"] = np.linalg.norm(near_vec - far_vec, axis=1).tolist()
+
+    # 2. Cosine Similarity
     if "Cosine Similarity" in metrics:
-        cos_sim = []
+        cos_vals = []
         for i in range(min_len):
-            a = near.iloc[i].values.reshape(1, -1)
-            b = far.iloc[i].values.reshape(1, -1)
-            sim = cosine_similarity(a, b)[0][0]
-            cos_sim.append(float(sim))
-        results["cosine_similarity"] = cos_sim
-    
+            a, b = near_vec[i].reshape(1, -1), far_vec[i].reshape(1, -1)
+            cos_vals.append(float(cosine_similarity(a, b)[0][0]))
+        results["cosine_similarity"] = cos_vals
+
+    # 3. Pearson Correlation
+    if "Pearson Correlation" in metrics:
+        corr_vals = []
+        for i in range(min_len):
+            corr, _ = pearsonr(near_vec[i], far_vec[i])
+            corr_vals.append(float(corr) if not np.isnan(corr) else 0.0)
+        results["pearson_corr"] = corr_vals
+
+    # 4. KL Divergence (on normalized features)
+    if "KL Divergence" in metrics:
+        kl_vals = []
+        for i in range(min_len):
+            p = near_vec[i] - near_vec[i].min() + 1e-8
+            q = far_vec[i] - far_vec[i].min() + 1e-8
+            p /= p.sum()
+            q /= q.sum()
+            kl = np.sum(p * np.log(p / q))
+            kl_vals.append(float(kl))
+        results["kl_divergence"] = kl_vals
+
+    # 5. Jensen-Shannon Divergence (symmetric, safer)
+    if "Jensen-Shannon Divergence" in metrics:
+        js_vals = []
+        for i in range(min_len):
+            p = near_vec[i] - near_vec[i].min() + 1e-8
+            q = far_vec[i] - far_vec[i].min() + 1e-8
+            p /= p.sum()
+            q /= q.sum()
+            js = jensenshannon(p, q)
+            js_vals.append(float(js))
+        results["js_divergence"] = js_vals
+
+    # 6. Lost High Frequencies Ratio
+    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: how much high-freq energy is lost (positive = loss)
+            ratio = near_high - far_high  # in dB
+            loss_ratios.append(float(ratio))
+        results["high_freq_loss_db"] = loss_ratios
+
+    # 7. Spectral Centroid Shift
+    if "Spectral Centroid Shift" in metrics:
+        shifts = []
+        for i in range(min_len):
+            shift = near_feats[i]["spectral_centroid"] - far_feats[i]["spectral_centroid"]
+            shifts.append(float(shift))
+        results["centroid_shift"] = shifts
+
     return pd.DataFrame(results)
 
-def cluster_frames(features_df, n_clusters):
-    X = features_df.values
-    kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
-    labels = kmeans.fit_predict(X)
+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.")
+    
+    X = features_df[cluster_features].values
+
+    if algo == "KMeans":
+        model = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
+    elif algo == "Agglomerative":
+        model = AgglomerativeClustering(n_clusters=n_clusters)
+    elif algo == "DBSCAN":
+        model = DBSCAN(eps=eps, min_samples=3)
+    else:
+        raise ValueError("Unknown clustering algorithm")
+
+    labels = model.fit_predict(X)
     features_df = features_df.copy()
     features_df["cluster"] = labels
     return features_df
 
+def plot_spectral_difference(near_feats, far_feats, frame_idx=0):
+    if frame_idx >= len(near_feats):
+        frame_idx = 0
+    near_spec = near_feats[frame_idx]["spectrum"]
+    far_spec = far_feats[frame_idx]["spectrum"]
+    diff = near_spec - far_spec  # positive = energy lost in far-field
+
+    fig = go.Figure(data=go.Heatmap(
+        z=[diff],
+        colorscale='RdBu',
+        zmid=0,
+        colorbar=dict(title="dB Difference")
+    ))
+    fig.update_layout(
+        title=f"Spectral Difference (Frame {frame_idx}): Near - Far",
+        xaxis_title="Frequency Bins",
+        yaxis_title="",
+        height=300
+    )
+    return fig
+
 # ----------------------------
-# Gradio App Logic
+# Main Analysis Function
 # ----------------------------
 
 def analyze_audio_pair(
@@ -97,75 +187,77 @@ def analyze_audio_pair(
     frame_length_ms,
     hop_length_ms,
     window_type,
+    comparison_metrics,
+    cluster_features,
+    clustering_algo,
     n_clusters,
-    comparison_metrics
+    dbscan_eps
 ):
     if not near_file or not far_file:
-        raise gr.Error("Please upload both near-field and far-field audio files.")
-    
-    # Load audios
+        raise gr.Error("Upload both audio files.")
+
     y_near, sr_near = load_audio(near_file)
     y_far, sr_far = load_audio(far_file)
 
-    # Resample if needed
     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
 
-    # Segment
     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)
 
-    # Extract features
-    near_features = extract_features(frames_near, sr)
-    far_features = extract_features(frames_far, sr)
+    near_feats = extract_features_with_spectrum(frames_near, sr)
+    far_feats = extract_features_with_spectrum(frames_far, sr)
 
-    # Compare
-    comparison_df = compare_frames(near_features, far_features, comparison_metrics)
+    # Comparison
+    comparison_df = compare_frames_enhanced(near_feats, far_feats, comparison_metrics)
 
-    # Cluster near-field
-    clustered_df = cluster_frames(near_features, n_clusters)
+    # Clustering (on near-field)
+    near_df = pd.DataFrame(near_feats)
+    near_df = near_df.drop(columns=["spectrum"], errors="ignore")
+    clustered_df = cluster_frames_custom(near_df, cluster_features, clustering_algo, n_clusters, dbscan_eps)
 
     # Plots
     plot_comparison = None
-    if "Euclidean Distance" in comparison_metrics:
+    if comparison_df.shape[1] > 1:
+        metric_to_plot = [col for col in comparison_df.columns if col != "frame_index"][0]
         plot_comparison = px.line(
             comparison_df,
             x="frame_index",
-            y="euclidean_dist",
-            title="Euclidean Distance Between Near & Far Frames"
+            y=metric_to_plot,
+            title=f"{metric_to_plot.replace('_', ' ').title()} Over Time"
         )
-    elif "Cosine Similarity" in comparison_metrics:
-        plot_comparison = px.line(
-            comparison_df,
-            x="frame_index",
-            y="cosine_similarity",
-            title="Cosine Similarity Between Near & Far Frames"
-        )
-    else:
-        # Fallback
-        col = comparison_df.columns[1] if len(comparison_df.columns) > 1 else "frame_index"
-        plot_comparison = px.line(comparison_df, x="frame_index", y=col, title="Frame Comparison")
 
-    # Scatter plot
+    # Scatter: user-selected features
     plot_scatter = None
-    if "rms" in clustered_df.columns and "spectral_centroid" in clustered_df.columns:
-        plot_scatter = px.scatter(
-            clustered_df,
-            x="rms",
-            y="spectral_centroid",
-            color="cluster",
-            title="Clustered Frames (RMS vs Spectral Centroid)",
-            labels={"rms": "RMS Energy", "spectral_centroid": "Spectral Centroid"}
-        )
+    if len(cluster_features) >= 2:
+        x_feat, y_feat = cluster_features[0], cluster_features[1]
+        if x_feat in clustered_df.columns and y_feat in clustered_df.columns:
+            plot_scatter = px.scatter(
+                clustered_df,
+                x=x_feat,
+                y=y_feat,
+                color="cluster",
+                title=f"Clustering: {x_feat} vs {y_feat}",
+                hover_data=["cluster"]
+            )
     else:
-        plot_scatter = px.scatter(title="Not enough features for scatter plot")
+        plot_scatter = px.scatter(title="Select ≥2 features for scatter plot")
+
+    # Spectral difference heatmap (first frame)
+    spec_heatmap = plot_spectral_difference(near_feats, far_feats, frame_idx=0)
 
-    return plot_comparison, comparison_df, plot_scatter, clustered_df
+    return (
+        plot_comparison,
+        comparison_df,
+        plot_scatter,
+        clustered_df,
+        spec_heatmap
+    )
 
-def export_results(plot_comparison, comparison_df, plot_scatter, clustered_df):
+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")
@@ -174,31 +266,56 @@ def export_results(plot_comparison, comparison_df, plot_scatter, clustered_df):
     return [comp_path, cluster_path]
 
 # ----------------------------
-# Gradio Interface
+# Gradio UI
 # ----------------------------
 
-with gr.Blocks(title="Near vs Far Field Audio Analyzer") as demo:
-    gr.Markdown("# 🎙️ Near vs Far Field Speech Analysis Platform")
-    gr.Markdown("Upload simultaneous near-field and far-field `.wav` recordings. Analyze, compare, and cluster frames.")
+# Get feature names dynamically
+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")
+    gr.Markdown("Upload simultaneous recordings. Analyze **lost frequencies**, **frame degradation**, and **cluster by custom attributes**.")
 
     with gr.Row():
-        near_file = gr.File(label="Upload Near-Field Audio (.wav)", file_types=[".wav"])
-        far_file = gr.File(label="Upload Far-Field Audio (.wav)", file_types=[".wav"])
+        near_file = gr.File(label="Near-Field Audio (.wav)", file_types=[".wav"])
+        far_file = gr.File(label="Far-Field Audio (.wav)", file_types=[".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 & Clustering", open=True):
-        n_clusters = gr.Slider(2, 20, value=5, step=1, label="Number of Clusters (KMeans)")
+    with gr.Accordion("📊 Comparison Metrics", open=True):
         comparison_metrics = gr.CheckboxGroup(
-            ["Euclidean Distance", "Cosine Similarity"],
-            value=["Euclidean Distance", "Cosine Similarity"],
-            label="Comparison Metrics"
+            choices=[
+                "Euclidean Distance",
+                "Cosine Similarity",
+                "Pearson Correlation",
+                "KL Divergence",
+                "Jensen-Shannon Divergence",
+                "High-Freq Loss Ratio",
+                "Spectral Centroid Shift"
+            ],
+            value=["High-Freq Loss Ratio", "Cosine Similarity"],
+            label="Select Comparison Metrics"
         )
 
-    btn = gr.Button("🚀 Analyze Audio Pair")
+    with gr.Accordion("🧩 Clustering Configuration", open=True):
+        cluster_features = gr.CheckboxGroup(
+            choices=dummy_features,
+            value=["rms", "spectral_centroid", "high_freq_energy"],
+            label="Features to Use 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="Number of Clusters (for KMeans/Agglomerative)")
+        dbscan_eps = gr.Slider(0.1, 2.0, value=0.5, step=0.1, label="DBSCAN eps (neighborhood radius)")
+
+    btn = gr.Button("🚀 Analyze")
 
     with gr.Tabs():
         with gr.Tab("📈 Frame Comparison"):
@@ -209,7 +326,10 @@ with gr.Blocks(title="Near vs Far Field Audio Analyzer") as demo:
             cluster_plot = gr.Plot()
             cluster_table = gr.Dataframe()
 
-    with gr.Tab("📤 Export Results"):
+        with gr.Tab("🔍 Spectral Analysis"):
+            spec_heatmap = gr.Plot(label="Spectral Difference (Near - Far)")
+
+    with gr.Tab("📤 Export"):
         export_btn = gr.Button("💾 Download CSVs")
         export_files = gr.Files()
 
@@ -218,21 +338,20 @@ with gr.Blocks(title="Near vs Far Field Audio Analyzer") as demo:
         inputs=[
             near_file, far_file,
             frame_length_ms, hop_length_ms, window_type,
+            comparison_metrics,
+            cluster_features,
+            clustering_algo,
             n_clusters,
-            comparison_metrics
+            dbscan_eps
         ],
-        outputs=[comp_plot, comp_table, cluster_plot, cluster_table]
+        outputs=[comp_plot, comp_table, cluster_plot, cluster_table, spec_heatmap]
     )
 
     export_btn.click(
         fn=export_results,
-        inputs=[comp_plot, comp_table, cluster_plot, cluster_table],
+        inputs=[comp_table, cluster_table],
         outputs=export_files
     )
 
-# ----------------------------
-# Launch
-# ----------------------------
-
 if __name__ == "__main__":
     demo.launch()