import os
import warnings
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from umap import UMAP
from sklearn.cluster import KMeans
from scipy.stats import entropy as shannon_entropy
from scipy import special as sp_special
from scipy.interpolate import griddata
from sklearn.metrics.pairwise import cosine_similarity
from scipy.spatial.distance import cdist
import soundfile as sf
import gradio as gr
# ================================================================
# Unified Communication Manifold Explorer & CMT Visualizer v4.0
# - Adds side-by-side comparison capabilities from HTML draft
# - Implements cross-species neighbor finding for grammar mapping
# - Separates human and dog audio with automatic pairing
# - Enhanced dual visualization for comparative analysis
# ================================================================
# - Adds Interactive Holography tab for full field reconstruction.
# - Interpolates the continuous CMT state-space (Φ field).
# - Visualizes topology, vector flow, and phase interference.
# - Adds informational-entropy-geometry visualization.
# - Prioritizes specific Colab paths for data loading.
# ================================================================
warnings.filterwarnings("ignore", category=FutureWarning)
warnings.filterwarnings("ignore", category=UserWarning)
print("Initializing the Interactive CMT Holography Explorer...")
# ---------------------------------------------------------------
# Data setup
# ---------------------------------------------------------------
# Paths for local execution (used for dummy data generation fallback)
BASE_DIR = os.path.abspath(os.getcwd())
DATA_DIR = os.path.join(BASE_DIR, "data")
DOG_DIR = os.path.join(DATA_DIR, "dog")
HUMAN_DIR = os.path.join(DATA_DIR, "human")
# Explicit paths for Colab environment
CSV_DOG = "/content/cmt_dog_sound_analysis.csv"
CSV_HUMAN = "/content/cmt_human_speech_analysis.csv"
# These are for creating dummy audio files if needed
os.makedirs(DOG_DIR, exist_ok=True)
os.makedirs(os.path.join(HUMAN_DIR, "Actor_01"), exist_ok=True)
# --- Audio Data Configuration (Must match your data source locations) ---
DOG_AUDIO_BASE_PATH = '/content/drive/MyDrive/combined'
HUMAN_AUDIO_BASE_PATH = '/content/drive/MyDrive/human'
# ---------------------------------------------------------------
# Cross-Species Analysis Functions
# ---------------------------------------------------------------
def find_nearest_cross_species_neighbor(selected_row, df_combined, n_neighbors=5):
"""
Finds the closest neighbor from the opposite species using feature similarity.
This enables cross-species pattern mapping for grammar development.
"""
selected_source = selected_row['source']
opposite_source = 'Human' if selected_source == 'Dog' else 'Dog'
# Get feature columns for similarity calculation
feature_cols = [c for c in df_combined.columns if c.startswith("feature_")]
if not feature_cols:
# Fallback to any numeric columns if no feature columns exist
numeric_cols = df_combined.select_dtypes(include=[np.number]).columns
feature_cols = [c for c in numeric_cols if c not in ['x', 'y', 'z', 'cluster']]
if not feature_cols:
# Random selection if no suitable features found
opposite_species_data = df_combined[df_combined['source'] == opposite_source]
if len(opposite_species_data) > 0:
return opposite_species_data.iloc[0]
return None
# Extract features for the selected row
selected_features = selected_row[feature_cols].values.reshape(1, -1)
selected_features = np.nan_to_num(selected_features)
# Get all rows from the opposite species
opposite_species_data = df_combined[df_combined['source'] == opposite_source]
if len(opposite_species_data) == 0:
return None
# Extract features for opposite species
opposite_features = opposite_species_data[feature_cols].values
opposite_features = np.nan_to_num(opposite_features)
# Calculate cosine similarity (better for high-dimensional feature spaces)
similarities = cosine_similarity(selected_features, opposite_features)[0]
# Find the index of the most similar neighbor
most_similar_idx = np.argmax(similarities)
nearest_neighbor = opposite_species_data.iloc[most_similar_idx]
return nearest_neighbor
# ---------------------------------------------------------------
# Load datasets (Colab-first paths)
# ---------------------------------------------------------------
if os.path.exists(CSV_DOG) and os.path.exists(CSV_HUMAN):
print(f"Found existing data files. Loading from:\n- {CSV_DOG}\n- {CSV_HUMAN}")
df_dog = pd.read_csv(CSV_DOG)
df_human = pd.read_csv(CSV_HUMAN)
print("Successfully loaded data from specified paths.")
else:
print("Could not find one or both CSV files. Generating and using in-memory dummy data.")
# This section is for DUMMY DATA GENERATION ONLY.
# It runs if the primary CSVs are not found and does NOT write files.
n_dummy_items_per_category = 50
rng = np.random.default_rng(42)
dog_labels = ["bark", "growl", "whine", "pant"] * (n_dummy_items_per_category // 4)
human_labels = ["speech", "laugh", "cry", "shout"] * (n_dummy_items_per_category // 4)
dog_rows = {
"feature_1": rng.random(n_dummy_items_per_category), "feature_2": rng.random(n_dummy_items_per_category), "feature_3": rng.random(n_dummy_items_per_category),
"label": dog_labels, "filepath": [f"dog_{i}.wav" for i in range(n_dummy_items_per_category)],
"diag_srl_gamma": rng.uniform(0.5, 5.0, n_dummy_items_per_category), "diag_alpha_gamma": rng.uniform(0.1, 2.0, n_dummy_items_per_category),
"zeta_curvature": rng.uniform(-1, 1, n_dummy_items_per_category), "torsion_index": rng.uniform(0, 1, n_dummy_items_per_category),
}
human_rows = {
"feature_1": rng.random(n_dummy_items_per_category), "feature_2": rng.random(n_dummy_items_per_category), "feature_3": rng.random(n_dummy_items_per_category),
"label": human_labels, "filepath": [f"human_{i}.wav" for i in range(n_dummy_items_per_category)],
"diag_srl_gamma": rng.uniform(0.5, 5.0, n_dummy_items_per_category), "diag_alpha_gamma": rng.uniform(0.1, 2.0, n_dummy_items_per_category),
"zeta_curvature": rng.uniform(-1, 1, n_dummy_items_per_category), "torsion_index": rng.uniform(0, 1, n_dummy_items_per_category),
}
df_dog = pd.DataFrame(dog_rows)
df_human = pd.DataFrame(human_rows)
# We still create dummy audio files for the UI to use if needed
sr = 22050
dur = 2.0
t = np.linspace(0, dur, int(sr * dur), endpoint=False)
for i in range(n_dummy_items_per_category):
tone_freq = 220 + 20 * (i % 5)
audio = 0.1 * np.sin(2 * np.pi * tone_freq * t) + 0.02 * rng.standard_normal(t.shape)
audio = audio / (np.max(np.abs(audio)) + 1e-9)
dog_label = dog_labels[i]
dog_label_dir = os.path.join(DOG_DIR, dog_label)
os.makedirs(dog_label_dir, exist_ok=True)
sf.write(os.path.join(dog_label_dir, f"dog_{i}.wav"), audio, sr)
sf.write(os.path.join(HUMAN_DIR, "Actor_01", f"human_{i}.wav"), audio, sr)
print(f"Loaded {len(df_dog)} dog rows and {len(df_human)} human rows.")
df_dog["source"], df_human["source"] = "Dog", "Human"
df_combined = pd.concat([df_dog, df_human], ignore_index=True)
# ---------------------------------------------------------------
# Expanded CMT implementation
# ---------------------------------------------------------------
class ExpandedCMT:
def __init__(self):
self.c1, self.c2 = 0.587 + 1.223j, -0.994 + 0.0j
# A large but finite number to represent the pole at z=1 for Zeta
self.ZETA_POLE_REGULARIZATION = 1e6 - 1e6j
self.lens_library = {
"gamma": sp_special.gamma,
"zeta": self._regularized_zeta, # Use the robust zeta function
"airy": lambda z: sp_special.airy(z)[0],
"bessel": lambda z: sp_special.jv(0, z),
}
def _regularized_zeta(self, z: np.ndarray) -> np.ndarray:
"""
A wrapper around scipy's zeta function to handle the pole at z=1.
"""
# Create a copy to avoid modifying the original array
z_out = np.copy(z).astype(np.complex128)
# Find where the real part is close to 1 and the imaginary part is close to 0
pole_condition = np.isclose(np.real(z), 1.0) & np.isclose(np.imag(z), 0.0)
# Apply the standard zeta function to non-pole points
non_pole_points = ~pole_condition
z_out[non_pole_points] = sp_special.zeta(z[non_pole_points], 1)
# Apply the regularization constant to the pole points
z_out[pole_condition] = self.ZETA_POLE_REGULARIZATION
return z_out
def _robust_normalize(self, signal: np.ndarray) -> np.ndarray:
if signal.size == 0: return signal
Q1, Q3 = np.percentile(signal, [25, 75])
IQR = Q3 - Q1
if IQR < 1e-9:
median, mad = np.median(signal), np.median(np.abs(signal - np.median(signal)))
return np.zeros_like(signal) if mad < 1e-9 else (signal - median) / (mad + 1e-9)
lower, upper = Q1 - 1.5 * IQR, Q3 + 1.5 * IQR
clipped = np.clip(signal, lower, upper)
s_min, s_max = np.min(clipped), np.max(clipped)
return np.zeros_like(signal) if s_max == s_min else 2.0 * (clipped - s_min) / (s_max - s_min) - 1.0
def _encode(self, signal: np.ndarray) -> np.ndarray:
N = len(signal)
if N == 0: return signal.astype(np.complex128)
i = np.arange(N)
theta = 2.0 * np.pi * i / N
f_k, A_k = np.array([271, 341, 491]), np.array([0.033, 0.050, 0.100])
phi = np.sum(A_k[:, None] * np.sin(2.0 * np.pi * f_k[:, None] * i / N), axis=0)
Theta = theta + phi
exp_iTheta = np.exp(1j * Theta)
g, m = signal * exp_iTheta, np.abs(signal) * exp_iTheta
return 0.5 * g + 0.5 * m
def _apply_lens(self, encoded_signal: np.ndarray, lens_type: str):
lens_fn = self.lens_library.get(lens_type)
if not lens_fn: raise ValueError(f"Lens '{lens_type}' not found.")
with np.errstate(all="ignore"):
w = lens_fn(encoded_signal)
phi_trajectory = self.c1 * np.angle(w) + self.c2 * np.abs(encoded_signal)
finite_mask = np.isfinite(phi_trajectory)
return phi_trajectory[finite_mask], w[finite_mask], encoded_signal[finite_mask], len(encoded_signal), len(phi_trajectory[finite_mask])
# ---------------------------------------------------------------
# Feature preparation and UMAP embedding
# ---------------------------------------------------------------
feature_cols = [c for c in df_combined.columns if c.startswith("feature_")]
features = np.nan_to_num(df_combined[feature_cols].to_numpy())
reducer = UMAP(n_components=3, n_neighbors=15, min_dist=0.1, random_state=42)
df_combined[["x", "y", "z"]] = reducer.fit_transform(features)
kmeans = KMeans(n_clusters=max(4, min(12, int(np.sqrt(len(df_combined))))), random_state=42, n_init=10)
df_combined["cluster"] = kmeans.fit_predict(features)
df_combined["chaos_score"] = np.log1p(df_combined.get("diag_srl_gamma", 0)) / (df_combined.get("diag_alpha_gamma", 1) + 1e-2)
# ---------------------------------------------------------------
# Core Visualization and Analysis Functions
# ---------------------------------------------------------------
def resolve_audio_path(row: pd.Series) -> str:
"""
Intelligently reconstructs the full path to an audio file
based on the logic from the data generation scripts.
"""
basename = str(row.get("filepath", ""))
source = row.get("source", "")
label = row.get("label", "")
# For "Dog" data, the structure is: {base_path}/{label}/{filename}
if source == "Dog":
expected_path = os.path.join(DOG_AUDIO_BASE_PATH, label, basename)
if os.path.exists(expected_path):
return expected_path
# For "Human" data, we must search within all "Actor_XX" subfolders
elif source == "Human":
if os.path.isdir(HUMAN_AUDIO_BASE_PATH):
for actor_folder in os.listdir(HUMAN_AUDIO_BASE_PATH):
if actor_folder.startswith("Actor_"):
expected_path = os.path.join(HUMAN_AUDIO_BASE_PATH, actor_folder, basename)
if os.path.exists(expected_path):
return expected_path
# Fallback for dummy data or other cases
if os.path.exists(basename):
return basename
# If all else fails, return the original basename and let it error out with a clear message
return basename
def get_cmt_data(filepath: str, lens: str):
try:
y, _ = sf.read(filepath)
if y.ndim > 1: y = np.mean(y, axis=1)
except Exception as e:
print(f"Error reading audio file {filepath}: {e}")
return None
cmt = ExpandedCMT()
normalized = cmt._robust_normalize(y)
encoded = cmt._encode(normalized)
# The _apply_lens function now returns additional diagnostic info
phi, w, z, original_count, final_count = cmt._apply_lens(encoded, lens)
return {
"phi": phi, "w": w, "z": z,
"original_count": original_count,
"final_count": final_count
}
def generate_holographic_field(z: np.ndarray, phi: np.ndarray, resolution: int):
if z is None or phi is None or len(z) < 4: return None
points = np.vstack([np.real(z), np.imag(z)]).T
grid_x, grid_y = np.mgrid[
np.min(points[:,0]):np.max(points[:,0]):complex(0, resolution),
np.min(points[:,1]):np.max(points[:,1]):complex(0, resolution)
]
grid_phi_real = griddata(points, np.real(phi), (grid_x, grid_y), method='cubic')
grid_phi_imag = griddata(points, np.imag(phi), (grid_x, grid_y), method='cubic')
grid_phi = np.nan_to_num(grid_phi_real + 1j * grid_phi_imag)
return grid_x, grid_y, grid_phi
def create_holography_plot(z, phi, resolution, wavelength):
field_data = generate_holographic_field(z, phi, resolution)
if field_data is None: return go.Figure(layout={"title": "Not enough data for holography"})
grid_x, grid_y, grid_phi = field_data
mag_phi = np.abs(grid_phi)
phase_phi = np.angle(grid_phi)
# --- Wavelength to Colorscale Mapping ---
def wavelength_to_rgb(wl):
# Simple approximation to map visible spectrum to RGB
if 380 <= wl < 440: return f'rgb({-(wl - 440) / (440 - 380) * 255}, 0, 255)' # Violet
elif 440 <= wl < 495: return f'rgb(0, {(wl - 440) / (495 - 440) * 255}, 255)' # Blue
elif 495 <= wl < 570: return f'rgb(0, 255, {-(wl - 570) / (570 - 495) * 255})' # Green
elif 570 <= wl < 590: return f'rgb({(wl - 570) / (590 - 570) * 255}, 255, 0)' # Yellow
elif 590 <= wl < 620: return f'rgb(255, {-(wl - 620) / (620 - 590) * 255}, 0)' # Orange
elif 620 <= wl <= 750: return f'rgb(255, 0, 0)' # Red
return 'rgb(255,255,255)'
mid_color = wavelength_to_rgb(wavelength)
custom_colorscale = [[0, 'rgb(20,0,40)'], [0.5, mid_color], [1, 'rgb(255,255,255)']]
fig = go.Figure()
# 1. The Holographic Surface (Topology + Phase Interference)
fig.add_trace(go.Surface(
x=grid_x, y=grid_y, z=mag_phi,
surfacecolor=phase_phi,
colorscale=custom_colorscale,
cmin=-np.pi, cmax=np.pi,
colorbar=dict(title='Φ Phase'),
name='Holographic Field',
contours_z=dict(show=True, usecolormap=True, highlightcolor="limegreen", project_z=True, highlightwidth=10)
))
# 2. The original data points projected onto the surface
fig.add_trace(go.Scatter3d(
x=np.real(z), y=np.imag(z), z=np.abs(phi) + 0.05, # slight offset
mode='markers',
marker=dict(size=3, color='black', symbol='x'),
name='Data Points'
))
# 3. The Vector Flow Field (using cones for direction)
grad_y, grad_x = np.gradient(mag_phi)
fig.add_trace(go.Cone(
x=grid_x.flatten(), y=grid_y.flatten(), z=mag_phi.flatten(),
u=-grad_x.flatten(), v=-grad_y.flatten(), w=np.full_like(mag_phi.flatten(), -0.1),
sizemode="absolute", sizeref=0.1,
anchor="tip",
colorscale='Greys',
showscale=False,
name='Vector Flow'
))
fig.update_layout(
title="Interactive Holographic Field Reconstruction",
scene=dict(
xaxis_title="Re(z) - Encoded Signal",
yaxis_title="Im(z) - Encoded Signal",
zaxis_title="|Φ| - Field Magnitude"
),
margin=dict(l=0, r=0, b=0, t=40)
)
return fig
def create_diagnostic_plots(z, w):
"""Creates a 2D plot showing the Aperture (z) and Lens Response (w)."""
if z is None or w is None:
return go.Figure(layout={"title": "Not enough data for diagnostic plots"})
fig = go.Figure()
# Aperture (Encoded Signal)
fig.add_trace(go.Scatter(
x=np.real(z), y=np.imag(z), mode='markers',
marker=dict(size=5, color='blue', opacity=0.6),
name='Aperture (z)'
))
# Lens Response
fig.add_trace(go.Scatter(
x=np.real(w), y=np.imag(w), mode='markers',
marker=dict(size=5, color='red', opacity=0.6, symbol='x'),
name='Lens Response (w)'
))
fig.update_layout(
title="Diagnostic View: Aperture and Lens Response",
xaxis_title="Real Part",
yaxis_title="Imaginary Part",
legend_title="Signal Stage",
margin=dict(l=20, r=20, t=60, b=20)
)
return fig
def create_dual_holography_plot(z1, phi1, z2, phi2, resolution, wavelength, title1="Primary", title2="Comparison"):
"""Creates side-by-side holographic visualizations for comparison."""
field_data1 = generate_holographic_field(z1, phi1, resolution)
field_data2 = generate_holographic_field(z2, phi2, resolution)
if field_data1 is None or field_data2 is None:
return go.Figure(layout={"title": "Insufficient data for dual holography"})
grid_x1, grid_y1, grid_phi1 = field_data1
grid_x2, grid_y2, grid_phi2 = field_data2
mag_phi1, phase_phi1 = np.abs(grid_phi1), np.angle(grid_phi1)
mag_phi2, phase_phi2 = np.abs(grid_phi2), np.angle(grid_phi2)
# Wavelength to colorscale mapping
def wavelength_to_rgb(wl):
if 380 <= wl < 440: return f'rgb({int(-(wl - 440) / (440 - 380) * 255)}, 0, 255)'
elif 440 <= wl < 495: return f'rgb(0, {int((wl - 440) / (495 - 440) * 255)}, 255)'
elif 495 <= wl < 570: return f'rgb(0, 255, {int(-(wl - 570) / (570 - 495) * 255)})'
elif 570 <= wl < 590: return f'rgb({int((wl - 570) / (590 - 570) * 255)}, 255, 0)'
elif 590 <= wl < 620: return f'rgb(255, {int(-(wl - 620) / (620 - 590) * 255)}, 0)'
elif 620 <= wl <= 750: return 'rgb(255, 0, 0)'
return 'rgb(255,255,255)'
mid_color = wavelength_to_rgb(wavelength)
custom_colorscale = [[0, 'rgb(20,0,40)'], [0.5, mid_color], [1, 'rgb(255,255,255)']]
fig = make_subplots(
rows=1, cols=2,
specs=[[{'type': 'scene'}, {'type': 'scene'}]],
subplot_titles=[title1, title2]
)
# Left plot (Primary)
fig.add_trace(go.Surface(
x=grid_x1, y=grid_y1, z=mag_phi1,
surfacecolor=phase_phi1,
colorscale=custom_colorscale,
cmin=-np.pi, cmax=np.pi,
showscale=False,
name=title1,
contours_z=dict(show=True, usecolormap=True, highlightcolor="limegreen", project_z=True)
), row=1, col=1)
# Right plot (Comparison)
fig.add_trace(go.Surface(
x=grid_x2, y=grid_y2, z=mag_phi2,
surfacecolor=phase_phi2,
colorscale=custom_colorscale,
cmin=-np.pi, cmax=np.pi,
showscale=False,
name=title2,
contours_z=dict(show=True, usecolormap=True, highlightcolor="limegreen", project_z=True)
), row=1, col=2)
# Add data points
if z1 is not None and phi1 is not None:
fig.add_trace(go.Scatter3d(
x=np.real(z1), y=np.imag(z1), z=np.abs(phi1) + 0.05,
mode='markers', marker=dict(size=3, color='black', symbol='x'),
name=f'{title1} Points', showlegend=False
), row=1, col=1)
if z2 is not None and phi2 is not None:
fig.add_trace(go.Scatter3d(
x=np.real(z2), y=np.imag(z2), z=np.abs(phi2) + 0.05,
mode='markers', marker=dict(size=3, color='black', symbol='x'),
name=f'{title2} Points', showlegend=False
), row=1, col=2)
fig.update_layout(
title="Side-by-Side Cross-Species Holographic Comparison",
scene=dict(
xaxis_title="Re(z)", yaxis_title="Im(z)", zaxis_title="|Φ|",
camera=dict(eye=dict(x=1.5, y=1.5, z=1.5))
),
scene2=dict(
xaxis_title="Re(z)", yaxis_title="Im(z)", zaxis_title="|Φ|",
camera=dict(eye=dict(x=1.5, y=1.5, z=1.5))
),
margin=dict(l=0, r=0, b=0, t=60),
height=600
)
return fig
def create_dual_diagnostic_plots(z1, w1, z2, w2, title1="Primary", title2="Comparison"):
"""Creates side-by-side diagnostic plots for cross-species comparison."""
fig = make_subplots(
rows=1, cols=2,
subplot_titles=[f"{title1}: Aperture & Lens Response", f"{title2}: Aperture & Lens Response"]
)
if z1 is not None and w1 is not None:
# Primary aperture and response
fig.add_trace(go.Scatter(
x=np.real(z1), y=np.imag(z1), mode='markers',
marker=dict(size=5, color='blue', opacity=0.6),
name=f'{title1} Aperture', showlegend=True
), row=1, col=1)
fig.add_trace(go.Scatter(
x=np.real(w1), y=np.imag(w1), mode='markers',
marker=dict(size=5, color='red', opacity=0.6, symbol='x'),
name=f'{title1} Response', showlegend=True
), row=1, col=1)
if z2 is not None and w2 is not None:
# Comparison aperture and response
fig.add_trace(go.Scatter(
x=np.real(z2), y=np.imag(z2), mode='markers',
marker=dict(size=5, color='darkblue', opacity=0.6),
name=f'{title2} Aperture', showlegend=True
), row=1, col=2)
fig.add_trace(go.Scatter(
x=np.real(w2), y=np.imag(w2), mode='markers',
marker=dict(size=5, color='darkred', opacity=0.6, symbol='x'),
name=f'{title2} Response', showlegend=True
), row=1, col=2)
fig.update_layout(
title="Cross-Species Diagnostic Comparison",
height=400,
margin=dict(l=20, r=20, t=60, b=20)
)
fig.update_xaxes(title_text="Real Part", row=1, col=1)
fig.update_yaxes(title_text="Imaginary Part", row=1, col=1)
fig.update_xaxes(title_text="Real Part", row=1, col=2)
fig.update_yaxes(title_text="Imaginary Part", row=1, col=2)
return fig
def create_entropy_geometry_plot(phi: np.ndarray):
"""Creates a plot showing magnitude/phase distributions and their entropy."""
if phi is None or len(phi) < 2:
return go.Figure(layout={"title": "Not enough data for entropy analysis"})
magnitudes = np.abs(phi)
phases = np.angle(phi)
# Calculate entropy
mag_hist, _ = np.histogram(magnitudes, bins='auto', density=True)
phase_hist, _ = np.histogram(phases, bins='auto', density=True)
mag_entropy = shannon_entropy(mag_hist)
phase_entropy = shannon_entropy(phase_hist)
fig = make_subplots(rows=1, cols=2, subplot_titles=(
f"Magnitude Distribution (Entropy: {mag_entropy:.3f})",
f"Phase Distribution (Entropy: {phase_entropy:.3f})"
))
fig.add_trace(go.Histogram(x=magnitudes, name='Magnitude', nbinsx=50), row=1, col=1)
fig.add_trace(go.Histogram(x=phases, name='Phase', nbinsx=50), row=1, col=2)
fig.update_layout(
title_text="Informational-Entropy Geometry",
showlegend=False,
bargap=0.1,
margin=dict(l=20, r=20, t=60, b=20)
)
fig.update_xaxes(title_text="|Φ|", row=1, col=1)
fig.update_yaxes(title_text="Count", row=1, col=1)
fig.update_xaxes(title_text="angle(Φ)", row=1, col=2)
fig.update_yaxes(title_text="Count", row=1, col=2)
return fig
# ---------------------------------------------------------------
# Gradio UI
# ---------------------------------------------------------------
with gr.Blocks(theme=gr.themes.Soft(primary_hue="teal", secondary_hue="cyan")) as demo:
gr.Markdown("# Exhaustive CMT Explorer for Interspecies Communication v3.2")
file_choices = df_combined["filepath"].astype(str).tolist()
default_primary = file_choices[0] if file_choices else ""
with gr.Tabs():
with gr.TabItem("Unified Manifold"):
gr.Plot(value=lambda: go.Figure(data=[go.Scatter3d(
x=df_combined["x"], y=df_combined["y"], z=df_combined["z"],
mode="markers", marker=dict(color=df_combined["cluster"], size=5, colorscale="Viridis", showscale=True, colorbar={"title": "Cluster ID"}),
text=df_combined.apply(lambda r: f"{r['source']}: {r.get('label', '')}
File: {r['filepath']}", axis=1),
hoverinfo="text"
)], layout=dict(title="Communication Manifold (UMAP Projection)")), label="UMAP Manifold")
with gr.TabItem("Interactive Holography"):
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### Cross-Species Holography Controls")
# Species selection and automatic pairing
species_dropdown = gr.Dropdown(
label="Select Species",
choices=["Dog", "Human"],
value="Dog"
)
# Primary file selection (filtered by species)
primary_dropdown = gr.Dropdown(
label="Primary Audio File",
choices=[],
value=""
)
# Automatically found neighbor (from opposite species)
neighbor_dropdown = gr.Dropdown(
label="Auto-Found Cross-Species Neighbor",
choices=[],
value="",
interactive=True # Allow manual override
)
holo_lens_dropdown = gr.Dropdown(label="CMT Lens", choices=["gamma", "zeta", "airy", "bessel"], value="gamma")
holo_resolution_slider = gr.Slider(label="Field Resolution", minimum=20, maximum=100, step=5, value=40)
holo_wavelength_slider = gr.Slider(label="Illumination Wavelength (nm)", minimum=380, maximum=750, step=5, value=550)
# Information panels
primary_info_html = gr.HTML(label="Primary Audio Info")
neighbor_info_html = gr.HTML(label="Neighbor Audio Info")
# Audio players
primary_audio_out = gr.Audio(label="Primary Audio")
neighbor_audio_out = gr.Audio(label="Neighbor Audio")
with gr.Column(scale=2):
dual_holography_plot = gr.Plot(label="Side-by-Side Holographic Comparison")
dual_diagnostic_plot = gr.Plot(label="Cross-Species Diagnostic Comparison")
def update_file_choices(species):
"""Update the primary file dropdown based on selected species."""
species_files = df_combined[df_combined["source"] == species]["filepath"].astype(str).tolist()
return gr.Dropdown.update(choices=species_files, value=species_files[0] if species_files else "")
def update_cross_species_view(species, primary_file, neighbor_file, lens, resolution, wavelength):
if not primary_file:
empty_fig = go.Figure(layout={"title": "Please select a primary file."})
return empty_fig, empty_fig, "", "", None, None, []
# Get primary row
primary_row = df_combined[
(df_combined["filepath"] == primary_file) &
(df_combined["source"] == species)
].iloc[0] if len(df_combined[
(df_combined["filepath"] == primary_file) &
(df_combined["source"] == species)
]) > 0 else None
if primary_row is None:
empty_fig = go.Figure(layout={"title": "Primary file not found."})
return empty_fig, empty_fig, "", "", None, None, []
# Find cross-species neighbor if not manually selected
if not neighbor_file:
neighbor_row = find_nearest_cross_species_neighbor(primary_row, df_combined)
if neighbor_row is not None:
neighbor_file = neighbor_row['filepath']
else:
# Get manually selected neighbor
opposite_species = 'Human' if species == 'Dog' else 'Dog'
neighbor_row = df_combined[
(df_combined["filepath"] == neighbor_file) &
(df_combined["source"] == opposite_species)
].iloc[0] if len(df_combined[
(df_combined["filepath"] == neighbor_file) &
(df_combined["source"] == opposite_species)
]) > 0 else None
# Get CMT data for both files
primary_fp = resolve_audio_path(primary_row)
primary_cmt = get_cmt_data(primary_fp, lens)
neighbor_cmt = None
if neighbor_row is not None:
neighbor_fp = resolve_audio_path(neighbor_row)
neighbor_cmt = get_cmt_data(neighbor_fp, lens)
# Create visualizations
if primary_cmt and neighbor_cmt:
primary_title = f"{species}: {primary_row.get('label', 'Unknown')}"
neighbor_title = f"{neighbor_row['source']}: {neighbor_row.get('label', 'Unknown')}"
dual_holo_fig = create_dual_holography_plot(
primary_cmt["z"], primary_cmt["phi"],
neighbor_cmt["z"], neighbor_cmt["phi"],
resolution, wavelength, primary_title, neighbor_title
)
dual_diag_fig = create_dual_diagnostic_plots(
primary_cmt["z"], primary_cmt["w"],
neighbor_cmt["z"], neighbor_cmt["w"],
primary_title, neighbor_title
)
else:
dual_holo_fig = go.Figure(layout={"title": "Error processing audio files"})
dual_diag_fig = go.Figure(layout={"title": "Error processing audio files"})
# Build info strings
primary_info = f"""
Primary: {primary_row['filepath']}
Species: {primary_row['source']}
Label: {primary_row.get('label', 'N/A')}
Data Points: {primary_cmt['final_count'] if primary_cmt else 0} / {primary_cmt['original_count'] if primary_cmt else 0}
"""
neighbor_info = ""
if neighbor_row is not None:
neighbor_info = f"""
Neighbor: {neighbor_row['filepath']}
Species: {neighbor_row['source']}
Label: {neighbor_row.get('label', 'N/A')}
Data Points: {neighbor_cmt['final_count'] if neighbor_cmt else 0} / {neighbor_cmt['original_count'] if neighbor_cmt else 0}
"""
# Update neighbor dropdown choices
opposite_species = 'Human' if species == 'Dog' else 'Dog'
neighbor_choices = df_combined[df_combined["source"] == opposite_species]["filepath"].astype(str).tolist()
# Audio files
primary_audio = primary_fp if primary_fp and os.path.exists(primary_fp) else None
neighbor_audio = neighbor_fp if neighbor_row and neighbor_fp and os.path.exists(neighbor_fp) else None
return (dual_holo_fig, dual_diag_fig, primary_info, neighbor_info,
primary_audio, neighbor_audio,
gr.Dropdown.update(choices=neighbor_choices, value=neighbor_file if neighbor_row else ""))
# Event handlers
species_dropdown.change(
update_file_choices,
inputs=[species_dropdown],
outputs=[primary_dropdown]
)
cross_species_inputs = [species_dropdown, primary_dropdown, neighbor_dropdown,
holo_lens_dropdown, holo_resolution_slider, holo_wavelength_slider]
cross_species_outputs = [dual_holography_plot, dual_diagnostic_plot,
primary_info_html, neighbor_info_html,
primary_audio_out, neighbor_audio_out, neighbor_dropdown]
for component in cross_species_inputs:
component.change(update_cross_species_view,
inputs=cross_species_inputs,
outputs=cross_species_outputs)
# Initialize on load
demo.load(lambda: update_file_choices("Dog"), outputs=[primary_dropdown])
demo.load(update_cross_species_view,
inputs=cross_species_inputs,
outputs=cross_species_outputs)
if __name__ == "__main__":
demo.launch(share=True, debug=True)