|
|
""" |
|
|
Gradio App for Entropy-Conserving Transformations |
|
|
|
|
|
This app demonstrates how divergence-free vector fields can transform |
|
|
arbitrary distributions towards Gaussian form while conserving entropy. |
|
|
""" |
|
|
|
|
|
import gradio as gr |
|
|
import matplotlib |
|
|
import matplotlib.pyplot as plt |
|
|
import numpy as np |
|
|
import pandas as pd |
|
|
|
|
|
from entra import DataFrameTransformer, VectorSampler |
|
|
|
|
|
matplotlib.use("Agg") |
|
|
|
|
|
|
|
|
def generate_uniform_data(n_per_dim: int = 20, dimensions: int = 2) -> pd.DataFrame: |
|
|
"""Generate uniform grid data using VectorSampler.""" |
|
|
if dimensions == 2: |
|
|
center = [0.0, 0.0] |
|
|
else: |
|
|
center = [0.0, 0.0, 0.0] |
|
|
|
|
|
sampler = VectorSampler( |
|
|
center=center, |
|
|
delta_x=1, |
|
|
num_points_per_dim=n_per_dim, |
|
|
distribution="uniform", |
|
|
) |
|
|
points = sampler.sample() |
|
|
|
|
|
if dimensions == 2: |
|
|
df = pd.DataFrame({"x": points[:, 0], "y": points[:, 1]}) |
|
|
else: |
|
|
df = pd.DataFrame({"x": points[:, 0], "y": points[:, 1], "z": points[:, 2]}) |
|
|
|
|
|
return df |
|
|
|
|
|
|
|
|
def generate_sample_csv(n_per_dim: int, dimensions: int): |
|
|
"""Generate sample CSV and return as downloadable file.""" |
|
|
df = generate_uniform_data(n_per_dim, dimensions) |
|
|
|
|
|
|
|
|
temp_path = "/tmp/generated_uniform_data.csv" |
|
|
df.to_csv(temp_path, index=False) |
|
|
|
|
|
n_points = len(df) |
|
|
cols = list(df.columns) |
|
|
preview = df.head(10).to_string() |
|
|
|
|
|
return ( |
|
|
temp_path, |
|
|
f"Generated {n_points} points with columns: {cols}\n\nPreview:\n{preview}", |
|
|
df, |
|
|
) |
|
|
|
|
|
|
|
|
def load_csv_file(file): |
|
|
"""Load uploaded CSV file.""" |
|
|
if file is None: |
|
|
return None, "No file uploaded", None |
|
|
|
|
|
df = pd.read_csv(file.name) |
|
|
n_points = len(df) |
|
|
cols = list(df.columns) |
|
|
preview = df.head(10).to_string() |
|
|
|
|
|
return ( |
|
|
file.name, |
|
|
f"Loaded {n_points} points with columns: {cols}\n\nPreview:\n{preview}", |
|
|
df, |
|
|
) |
|
|
|
|
|
|
|
|
def run_transformation( |
|
|
df_state, |
|
|
columns_str: str, |
|
|
sigma: float, |
|
|
max_iterations: int, |
|
|
progress=gr.Progress(), |
|
|
): |
|
|
"""Run the LM optimization and return results.""" |
|
|
if df_state is None: |
|
|
return ( |
|
|
None, |
|
|
None, |
|
|
None, |
|
|
"Error: No data loaded. Please upload or generate data first.", |
|
|
) |
|
|
|
|
|
df = df_state |
|
|
|
|
|
|
|
|
columns = [c.strip() for c in columns_str.split(",")] |
|
|
|
|
|
|
|
|
missing = [c for c in columns if c not in df.columns] |
|
|
if missing: |
|
|
return ( |
|
|
None, |
|
|
None, |
|
|
None, |
|
|
f"Error: Columns not found: {missing}. Available: {list(df.columns)}", |
|
|
) |
|
|
|
|
|
|
|
|
def progress_callback(iteration, max_iter, det_val, entropy_val): |
|
|
progress( |
|
|
iteration / max_iter, |
|
|
desc=f"Iter {iteration}/{max_iter} | Det: {det_val:.2e} | H: {entropy_val:.4f}", |
|
|
) |
|
|
|
|
|
|
|
|
transformer = DataFrameTransformer( |
|
|
sigma=sigma, |
|
|
max_iterations=max_iterations, |
|
|
verbose=False, |
|
|
progress_callback=progress_callback, |
|
|
) |
|
|
|
|
|
|
|
|
df_transformed = transformer.fit_transform(df, columns=columns) |
|
|
|
|
|
|
|
|
entropy = transformer.get_entropy_comparison(df, df_transformed) |
|
|
target_entropy = entropy["original"]["uniform_entropy"] |
|
|
|
|
|
|
|
|
fig_scatter = create_scatter_plot(df, df_transformed, columns) |
|
|
fig_hist = create_histogram_plot(df, df_transformed, columns) |
|
|
fig_history = create_history_plot(transformer.history_, target_entropy=target_entropy) |
|
|
|
|
|
|
|
|
results_text = format_results(entropy, transformer.history_) |
|
|
|
|
|
return fig_scatter, fig_hist, fig_history, results_text |
|
|
|
|
|
|
|
|
def create_scatter_plot(df_orig, df_trans, columns): |
|
|
"""Create before/after scatter plot.""" |
|
|
fig, axes = plt.subplots(1, 2, figsize=(12, 5)) |
|
|
|
|
|
if len(columns) >= 2: |
|
|
x_col, y_col = columns[0], columns[1] |
|
|
|
|
|
axes[0].scatter(df_orig[x_col], df_orig[y_col], c="blue", alpha=0.5, s=10) |
|
|
axes[0].set_xlabel(x_col) |
|
|
axes[0].set_ylabel(y_col) |
|
|
axes[0].set_title("Original Distribution") |
|
|
axes[0].set_aspect("equal") |
|
|
axes[0].grid(True, alpha=0.3) |
|
|
|
|
|
axes[1].scatter(df_trans[x_col], df_trans[y_col], c="red", alpha=0.5, s=10) |
|
|
axes[1].set_xlabel(x_col) |
|
|
axes[1].set_ylabel(y_col) |
|
|
axes[1].set_title("Transformed (Towards Gaussian)") |
|
|
axes[1].set_aspect("equal") |
|
|
axes[1].grid(True, alpha=0.3) |
|
|
|
|
|
plt.tight_layout() |
|
|
return fig |
|
|
|
|
|
|
|
|
def create_histogram_plot(df_orig, df_trans, columns): |
|
|
"""Create marginal histogram plots.""" |
|
|
n_cols = min(len(columns), 3) |
|
|
fig, axes = plt.subplots(n_cols, 2, figsize=(12, 4 * n_cols)) |
|
|
|
|
|
if n_cols == 1: |
|
|
axes = axes.reshape(1, -1) |
|
|
|
|
|
for i, col in enumerate(columns[:n_cols]): |
|
|
|
|
|
axes[i, 0].hist(df_orig[col], bins=30, density=True, alpha=0.7, color="blue") |
|
|
axes[i, 0].set_xlabel(col) |
|
|
axes[i, 0].set_ylabel("Density") |
|
|
axes[i, 0].set_title(f"Original {col} Marginal") |
|
|
|
|
|
|
|
|
axes[i, 1].hist(df_trans[col], bins=30, density=True, alpha=0.7, color="red") |
|
|
x_range = np.linspace(df_trans[col].min(), df_trans[col].max(), 100) |
|
|
mu = df_trans[col].mean() |
|
|
std = df_trans[col].std() |
|
|
gaussian = (1 / (std * np.sqrt(2 * np.pi))) * np.exp( |
|
|
-0.5 * ((x_range - mu) / std) ** 2 |
|
|
) |
|
|
axes[i, 1].plot(x_range, gaussian, "k--", linewidth=2, label="Gaussian fit") |
|
|
axes[i, 1].set_xlabel(col) |
|
|
axes[i, 1].set_ylabel("Density") |
|
|
axes[i, 1].set_title(f"Transformed {col} Marginal") |
|
|
axes[i, 1].legend() |
|
|
|
|
|
plt.tight_layout() |
|
|
return fig |
|
|
|
|
|
|
|
|
def create_history_plot(history, target_entropy=None): |
|
|
"""Create optimization history plot.""" |
|
|
fig, axes = plt.subplots(1, 2, figsize=(12, 4)) |
|
|
|
|
|
|
|
|
axes[0].semilogy(history["iteration"], history["determinant"], "b-o", markersize=4) |
|
|
axes[0].set_xlabel("Iteration") |
|
|
axes[0].set_ylabel("Covariance Determinant") |
|
|
axes[0].set_title("Determinant Minimization") |
|
|
axes[0].grid(True, alpha=0.3) |
|
|
|
|
|
|
|
|
axes[1].plot(history["iteration"], history["gaussian_entropy"], "r-o", markersize=4) |
|
|
if target_entropy is not None: |
|
|
axes[1].axhline( |
|
|
target_entropy, |
|
|
color="green", |
|
|
linestyle="--", |
|
|
linewidth=2, |
|
|
label=f"Target H(uniform) = {target_entropy:.4f}", |
|
|
) |
|
|
axes[1].legend() |
|
|
axes[1].set_xlabel("Iteration") |
|
|
axes[1].set_ylabel("H(Gaussian)") |
|
|
axes[1].set_title("Gaussian Entropy → Target Uniform Entropy") |
|
|
axes[1].grid(True, alpha=0.3) |
|
|
|
|
|
plt.tight_layout() |
|
|
return fig |
|
|
|
|
|
|
|
|
def format_results(entropy, history): |
|
|
"""Format results as text.""" |
|
|
det_reduction = ( |
|
|
entropy["original"]["determinant"] / entropy["transformed"]["determinant"] |
|
|
) |
|
|
target_entropy = entropy["original"]["uniform_entropy"] |
|
|
final_entropy = entropy["transformed"]["gaussian_entropy"] |
|
|
entropy_gap = final_entropy - target_entropy |
|
|
|
|
|
text = f""" |
|
|
TRANSFORMATION RESULTS |
|
|
{'=' * 50} |
|
|
|
|
|
Target Entropy (Uniform Distribution): |
|
|
H(uniform) = {target_entropy:.6f} nats |
|
|
|
|
|
This is the true entropy we want to reach. |
|
|
|
|
|
Gaussian Entropy of Transformed Data: |
|
|
H(Gaussian) = {final_entropy:.6f} nats |
|
|
|
|
|
This assumes the transformed data is Gaussian with the |
|
|
current covariance. When H(Gaussian) = H(uniform), the |
|
|
distribution is perfectly Gaussian. |
|
|
|
|
|
Gap to Target: |
|
|
H(Gaussian) - H(uniform) = {entropy_gap:.6f} nats |
|
|
(Should approach 0 for perfect Gaussianization) |
|
|
|
|
|
Covariance Determinant: |
|
|
Original: {entropy['original']['determinant']:.6e} |
|
|
Transformed: {entropy['transformed']['determinant']:.6e} |
|
|
Reduction: {det_reduction:.2f}x |
|
|
|
|
|
Optimization: |
|
|
Iterations with improvement: {len(history['iteration'])} |
|
|
Final determinant: {history['determinant'][-1]:.6e} |
|
|
Final H(Gaussian): {history['gaussian_entropy'][-1]:.6f} |
|
|
""" |
|
|
return text |
|
|
|
|
|
|
|
|
|
|
|
LM_EXPLANATION = """ |
|
|
## How the Levenberg-Marquardt Algorithm Works |
|
|
|
|
|
The **Levenberg-Marquardt (LM) algorithm** is used to minimize the covariance determinant. Unlike gradient descent, **LM has no learning rate** - here's why: |
|
|
|
|
|
### The Key Insight |
|
|
|
|
|
LM is designed for **least-squares problems** where you minimize a sum of squared residuals. Instead of taking steps proportional to the gradient (like gradient descent), LM solves a **local linear approximation** of the problem at each step. |
|
|
|
|
|
### How It Works |
|
|
|
|
|
1. **Compute the Jacobian** `J` - the matrix of partial derivatives of residuals with respect to parameters |
|
|
|
|
|
2. **Solve the normal equations**: |
|
|
``` |
|
|
(J^T J + λI) δ = -J^T r |
|
|
``` |
|
|
where `r` is the residual vector and `λ` is a damping parameter |
|
|
|
|
|
3. **The damping parameter λ replaces the learning rate**: |
|
|
- When `λ` is **large**: The step is small and in the gradient direction (like gradient descent with small learning rate) |
|
|
- When `λ` is **small**: The step approaches the Gauss-Newton step (a direct jump to the local minimum of the quadratic approximation) |
|
|
|
|
|
4. **Adaptive adjustment**: |
|
|
- If a step **decreases** the objective: Accept it and **decrease λ** (take bigger steps) |
|
|
- If a step **increases** the objective: Reject it and **increase λ** (take smaller, safer steps) |
|
|
|
|
|
### Why No Learning Rate? |
|
|
|
|
|
The LM algorithm **automatically adapts** its step size through the damping parameter λ: |
|
|
- It starts cautious (large λ, small steps) |
|
|
- As it finds a good direction, it becomes more aggressive (small λ, large steps) |
|
|
- If it overshoots, it backs off automatically |
|
|
|
|
|
This makes LM much more robust than gradient descent - you don't need to tune a learning rate! |
|
|
|
|
|
### In This Application |
|
|
|
|
|
We minimize `log(det(Cov))` where `Cov` is the covariance matrix of the transformed points. The transformation is parameterized by coefficients of divergence-free basis functions, ensuring the transformation is **volume-preserving** and thus **entropy-conserving**. |
|
|
""" |
|
|
|
|
|
THEORY_EXPLANATION = """ |
|
|
## Theoretical Background |
|
|
|
|
|
### Maximum Entropy Principle |
|
|
|
|
|
A fundamental theorem states: **Among all distributions with a given covariance matrix, the Gaussian has maximum entropy.** |
|
|
|
|
|
This means for any distribution with entropy `H₀` and covariance `Σ`: |
|
|
- The Gaussian with the same covariance has entropy `H_Gaussian(Σ) ≥ H₀` |
|
|
- Equality holds only when the distribution is Gaussian |
|
|
|
|
|
### The Key Insight |
|
|
|
|
|
If we apply a **volume-preserving transformation**: |
|
|
1. The entropy stays fixed at `H₀` (entropy is conserved) |
|
|
2. But the covariance changes |
|
|
|
|
|
By **minimizing the covariance determinant** while preserving entropy: |
|
|
- We reduce `H_Gaussian(Σ)` (the Gaussian entropy bound) |
|
|
- When `H_Gaussian(Σ) = H₀`, the distribution must be Gaussian! |
|
|
|
|
|
### Why Divergence-Free? |
|
|
|
|
|
Divergence-free vector fields define **volume-preserving** transformations: |
|
|
- The Jacobian determinant equals 1 everywhere |
|
|
- Total probability volume is conserved |
|
|
- **Entropy is conserved** under the transformation |
|
|
|
|
|
This is the incompressibility condition from fluid dynamics: `∇·v = 0` |
|
|
|
|
|
### The Operator |
|
|
|
|
|
We construct divergence-free basis functions using Lowitzsch's operator: |
|
|
|
|
|
**Ô = -I∇² + ∇∇ᵀ** |
|
|
|
|
|
Applied to Gaussian RBFs, this produces matrix-valued functions where each column is a divergence-free vector field. |
|
|
""" |
|
|
|
|
|
|
|
|
def create_app(): |
|
|
"""Create the Gradio interface.""" |
|
|
with gr.Blocks( |
|
|
title="Entropy-Conserving Transformations", theme=gr.themes.Soft() |
|
|
) as app: |
|
|
gr.Markdown( |
|
|
""" |
|
|
# Entropy-Conserving Transformations Using Divergence-Free Vector Fields |
|
|
|
|
|
Transform arbitrary distributions towards Gaussian form while **conserving entropy**. |
|
|
|
|
|
This demo uses divergence-free basis functions to create volume-preserving transformations, |
|
|
then minimizes the covariance determinant using the Levenberg-Marquardt algorithm. |
|
|
""" |
|
|
) |
|
|
|
|
|
gr.HTML(""" |
|
|
<div style="text-align: center; margin: 10px 0;"> |
|
|
<a href="https://www.paypal.com/donate?business=varun.kapoor@kapoorlabs.org¤cy_code=EUR" target="_blank" style="text-decoration: none;"> |
|
|
<button style="background-color: #0070ba; color: white; padding: 10px 20px; border: none; border-radius: 5px; cursor: pointer; font-size: 14px; font-weight: bold;"> |
|
|
☕ Buy me a coffee (PayPal) |
|
|
</button> |
|
|
</a> |
|
|
</div> |
|
|
""") |
|
|
|
|
|
|
|
|
df_state = gr.State(None) |
|
|
|
|
|
with gr.Tabs(): |
|
|
with gr.Tab("Transform Data"): |
|
|
with gr.Row(): |
|
|
with gr.Column(scale=1): |
|
|
gr.Markdown("### Step 1: Load or Generate Data") |
|
|
|
|
|
gr.Markdown( |
|
|
""" |
|
|
**No CSV file?** Use "Generate Sample Data" below to create a uniform grid. |
|
|
|
|
|
**Have your own CSV?** Format requirements: |
|
|
- Header row with column names |
|
|
- Numeric columns for coordinates (e.g., `x`, `y`, `z`) |
|
|
- Example: |
|
|
``` |
|
|
x,y |
|
|
-9.5,-9.5 |
|
|
-9.5,-8.5 |
|
|
... |
|
|
``` |
|
|
""" |
|
|
) |
|
|
|
|
|
with gr.Accordion( |
|
|
"Generate Sample Data (no CSV needed)", open=True |
|
|
): |
|
|
gr.Markdown( |
|
|
"*Creates a uniform grid using VectorSampler - perfect for testing*" |
|
|
) |
|
|
n_per_dim = gr.Slider( |
|
|
minimum=5, |
|
|
maximum=500, |
|
|
value=20, |
|
|
step=1, |
|
|
label="Points per dimension", |
|
|
) |
|
|
dimensions = gr.Radio( |
|
|
choices=[2, 3], value=2, label="Dimensions" |
|
|
) |
|
|
generate_btn = gr.Button( |
|
|
"Generate Uniform Distribution", |
|
|
variant="primary", |
|
|
) |
|
|
download_file = gr.File(label="Download generated CSV") |
|
|
|
|
|
with gr.Accordion("Upload Your Own CSV", open=False): |
|
|
file_upload = gr.File( |
|
|
label="Upload CSV file", file_types=[".csv"] |
|
|
) |
|
|
upload_btn = gr.Button("Load CSV", variant="secondary") |
|
|
|
|
|
data_info = gr.Textbox( |
|
|
label="Data Info", lines=8, interactive=False |
|
|
) |
|
|
|
|
|
gr.Markdown("### Step 2: Configure Transformation") |
|
|
|
|
|
columns_input = gr.Textbox( |
|
|
value="x, y", |
|
|
label="Columns to transform (comma-separated)", |
|
|
lines=3, |
|
|
) |
|
|
sigma = gr.Slider( |
|
|
minimum=0.1, |
|
|
maximum=200.0, |
|
|
value=5.0, |
|
|
step=0.1, |
|
|
label="Sigma (RBF width)", |
|
|
) |
|
|
max_iterations = gr.Slider( |
|
|
minimum=10, |
|
|
maximum=5000, |
|
|
value=100, |
|
|
step=10, |
|
|
label="Max iterations", |
|
|
) |
|
|
|
|
|
transform_btn = gr.Button( |
|
|
"Run Transformation", variant="primary", size="lg" |
|
|
) |
|
|
|
|
|
with gr.Column(scale=2): |
|
|
gr.Markdown("### Results") |
|
|
|
|
|
results_text = gr.Textbox( |
|
|
label="Transformation Results", |
|
|
lines=20, |
|
|
interactive=False, |
|
|
) |
|
|
|
|
|
with gr.Row(): |
|
|
scatter_plot = gr.Plot(label="Before/After Scatter") |
|
|
|
|
|
with gr.Row(): |
|
|
hist_plot = gr.Plot(label="Marginal Distributions") |
|
|
|
|
|
with gr.Row(): |
|
|
history_plot = gr.Plot(label="Optimization History") |
|
|
|
|
|
with gr.Tab("How LM Works"): |
|
|
gr.Markdown(LM_EXPLANATION) |
|
|
|
|
|
with gr.Tab("Theory"): |
|
|
gr.Markdown(THEORY_EXPLANATION) |
|
|
|
|
|
|
|
|
def on_generate(n, dims): |
|
|
path, info, df = generate_sample_csv(n, dims) |
|
|
return path, info, df |
|
|
|
|
|
def on_upload(file): |
|
|
path, info, df = load_csv_file(file) |
|
|
return info, df |
|
|
|
|
|
generate_btn.click( |
|
|
fn=on_generate, |
|
|
inputs=[n_per_dim, dimensions], |
|
|
outputs=[download_file, data_info, df_state], |
|
|
) |
|
|
|
|
|
upload_btn.click( |
|
|
fn=on_upload, inputs=[file_upload], outputs=[data_info, df_state] |
|
|
) |
|
|
|
|
|
transform_btn.click( |
|
|
fn=run_transformation, |
|
|
inputs=[ |
|
|
df_state, |
|
|
columns_input, |
|
|
sigma, |
|
|
max_iterations, |
|
|
], |
|
|
outputs=[scatter_plot, hist_plot, history_plot, results_text], |
|
|
) |
|
|
|
|
|
return app |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
app = create_app() |
|
|
app.launch() |
|
|
|
