| """
|
| Gradio application for the Physics-Informed Bayesian Optimization Platform.
|
|
|
| Provides an interactive UI for:
|
| 1. Defining parameter spaces
|
| 2. Specifying physics models (Python code)
|
| 3. Uploading initial experimental data
|
| 4. Running BO campaigns
|
| 5. Visualizing results
|
| """
|
|
|
| import io
|
| import json
|
| import traceback
|
| from typing import Optional
|
|
|
| import gradio as gr
|
| import matplotlib
|
| matplotlib.use("Agg")
|
| import matplotlib.pyplot as plt
|
| import numpy as np
|
| import pandas as pd
|
| import torch
|
| from torch import Tensor
|
|
|
|
|
|
|
|
|
|
|
| BUILTIN_PHYSICS = {
|
| "Arrhenius Kinetics": {
|
| "code": """\
|
| def physics_model(X):
|
| \"\"\"Arrhenius kinetics: rate = A * exp(-Ea / (R*T)) * C^n\"\"\"
|
| T = X[:, 0] # temperature (K)
|
| C = X[:, 1] # concentration
|
| A = 1e8 # pre-exponential factor
|
| Ea = 50.0 # activation energy (kJ/mol)
|
| R = 8.314e-3 # gas constant (kJ/mol·K)
|
| n = 0.5 # reaction order
|
| return A * torch.exp(-Ea / (R * T)) * C ** n
|
| """,
|
| "params": "temperature (K): 300-800\nconcentration: 0.1-10",
|
| },
|
| "Flory-Huggins Mixing": {
|
| "code": """\
|
| def physics_model(X):
|
| \"\"\"Flory-Huggins free energy of mixing for binary polymer blend.\"\"\"
|
| phi = X[:, 0] # volume fraction (0-1)
|
| chi = X[:, 1] # Flory-Huggins parameter
|
| N = 100.0 # degree of polymerisation
|
| entropy = phi * torch.log(phi + 1e-8) / N + (1 - phi) * torch.log(1 - phi + 1e-8) / N
|
| enthalpy = chi * phi * (1 - phi)
|
| return -(entropy + enthalpy) # negative ΔG_mix (higher = better mixing)
|
| """,
|
| "params": "volume_fraction: 0.05-0.95\nchi_parameter: 0.0-2.0",
|
| },
|
| "Polymer Recyclability": {
|
| "code": """\
|
| def physics_model(X):
|
| \"\"\"Simplified recyclability metric for polymer formulation.\"\"\"
|
| ratio = X[:, 0] # monomer ratio
|
| temp = X[:, 1] # temperature (K)
|
| catalyst = X[:, 2] # catalyst loading (wt%)
|
| mixing = -ratio * torch.log(ratio + 1e-8) - (1 - ratio) * torch.log(1 - ratio + 1e-8)
|
| chi = 0.5 - 0.3 * (ratio - 0.5) ** 2
|
| mixing_fe = mixing - chi * ratio * (1 - ratio)
|
| rate = torch.exp(-50.0 / (8.314e-3 * temp))
|
| cat_eff = 1 - torch.exp(-0.8 * catalyst)
|
| return 5.0 * mixing_fe * rate * cat_eff + 2.0
|
| """,
|
| "params": "monomer_ratio: 0.1-0.9\ntemperature (K): 350-500\ncatalyst_loading (wt%): 0.5-5.0",
|
| },
|
| "Custom (enter code below)": {"code": "", "params": ""},
|
| }
|
|
|
| DEMO_CSV = """\
|
| temperature,concentration,yield
|
| 350,1.0,0.12
|
| 400,3.0,0.45
|
| 450,5.0,0.78
|
| 500,2.0,0.55
|
| 480,7.0,0.91
|
| """
|
|
|
|
|
|
|
|
|
|
|
| def _compile_physics_fn(code: str):
|
| """Safely compile user-provided physics model code.
|
|
|
| The code must define a function called `physics_model(X)`.
|
| """
|
| allowed_globals = {"torch": torch, "np": np, "Tensor": Tensor, "__builtins__": {}}
|
|
|
| import builtins
|
| safe_builtins = {
|
| k: getattr(builtins, k)
|
| for k in ("range", "len", "float", "int", "abs", "max", "min", "print", "list", "tuple", "dict", "True", "False", "None")
|
| }
|
| allowed_globals["__builtins__"] = safe_builtins
|
|
|
| local_ns = {}
|
| exec(code, allowed_globals, local_ns)
|
| if "physics_model" not in local_ns:
|
| raise ValueError("Code must define a function called `physics_model(X)`.")
|
| return local_ns["physics_model"]
|
|
|
|
|
|
|
|
|
|
|
|
|
| def _parse_params(text: str):
|
| """Parse parameter definitions from multiline text.
|
|
|
| Format per line: name: lower-upper
|
| Example: temperature (K): 300-800
|
| """
|
| from physics_informed_bo.experiment.parameter_space import ParameterSpace
|
|
|
| space = ParameterSpace()
|
| names = []
|
| for line in text.strip().splitlines():
|
| line = line.strip()
|
| if not line:
|
| continue
|
| name_part, bounds_part = line.rsplit(":", 1)
|
| name = name_part.strip()
|
| lo, hi = bounds_part.strip().split("-")
|
| space.add_continuous(name, float(lo), float(hi))
|
| names.append(name)
|
| return space, names
|
|
|
|
|
|
|
|
|
|
|
|
|
| def run_optimization(
|
| physics_template: str,
|
| physics_code: str,
|
| param_text: str,
|
| csv_file,
|
| csv_text: str,
|
| objective_col: str,
|
| acq_fn: str,
|
| n_initial: int,
|
| n_iterations: int,
|
| batch_size: int,
|
| noise_var: float,
|
| maximize: bool,
|
| seed: int,
|
| ):
|
| """Run the full physics-informed BO campaign and return results."""
|
| try:
|
| torch.manual_seed(seed)
|
|
|
|
|
| code = physics_code.strip()
|
| if physics_template != "Custom (enter code below)" and not code:
|
| code = BUILTIN_PHYSICS[physics_template]["code"]
|
| physics_fn = _compile_physics_fn(code) if code else None
|
|
|
|
|
| if not param_text.strip():
|
| if physics_template != "Custom (enter code below)":
|
| param_text = BUILTIN_PHYSICS[physics_template]["params"]
|
| space, param_names = _parse_params(param_text)
|
|
|
|
|
| X_init, y_init = None, None
|
| df_init = None
|
|
|
| if csv_file is not None:
|
| df_init = pd.read_csv(csv_file.name)
|
| elif csv_text.strip():
|
| df_init = pd.read_csv(io.StringIO(csv_text.strip()))
|
|
|
| if df_init is not None:
|
| obj = objective_col.strip() or df_init.columns[-1]
|
| feature_cols = [c for c in df_init.columns if c != obj]
|
|
|
| if set(feature_cols) != set(param_names):
|
|
|
| feature_cols = [c for c in df_init.columns if c != obj][:len(param_names)]
|
| X_init = torch.tensor(df_init[feature_cols].values, dtype=torch.float64)
|
| y_init = torch.tensor(df_init[obj].values, dtype=torch.float64).unsqueeze(-1)
|
|
|
|
|
| from physics_informed_bo.config import OptimizationConfig, AcquisitionType
|
|
|
| acq_map = {
|
| "Expected Improvement (EI)": AcquisitionType.EXPECTED_IMPROVEMENT,
|
| "Upper Confidence Bound (UCB)": AcquisitionType.UPPER_CONFIDENCE_BOUND,
|
| "Probability of Improvement (PI)": AcquisitionType.PROBABILITY_OF_IMPROVEMENT,
|
| "Physics-Informed EI": AcquisitionType.PHYSICS_INFORMED_EI,
|
| }
|
|
|
| config = OptimizationConfig(
|
| acquisition_type=acq_map.get(acq_fn, AcquisitionType.EXPECTED_IMPROVEMENT),
|
| n_initial_samples=n_initial,
|
| max_iterations=n_iterations,
|
| batch_size=batch_size,
|
| noise_variance=noise_var,
|
| seed=seed,
|
| )
|
|
|
|
|
| from physics_informed_bo.experiment.campaign import OptimizationCampaign
|
|
|
| initial_data = (X_init, y_init) if X_init is not None else None
|
|
|
| campaign = OptimizationCampaign(
|
| name="hf_space_campaign",
|
| parameter_space=space,
|
| physics_fn=physics_fn,
|
| initial_data=initial_data,
|
| config=config,
|
| maximize=maximize,
|
| )
|
|
|
|
|
|
|
|
|
|
|
| def synthetic_objective(params: dict) -> float:
|
| vals = [params[n] for n in param_names]
|
| X = torch.tensor([vals], dtype=torch.float64)
|
| if physics_fn is not None:
|
| base = physics_fn(X).item()
|
| else:
|
| base = 0.0
|
| discrepancy = 0.15 * np.sin(3.0 * sum(vals))
|
| noise = noise_var**0.5 * np.random.randn()
|
| return base + discrepancy + noise
|
|
|
|
|
| log_lines = []
|
| best_vals = []
|
|
|
| for it in range(n_iterations):
|
| suggestions = campaign.suggest_next(batch_size)
|
| for params in suggestions:
|
| obj_val = synthetic_objective(params)
|
| campaign.report_result(params, obj_val)
|
| best = campaign.get_best() if maximize else campaign.get_best()
|
| best_vals.append(best["objective"])
|
| log_lines.append(
|
| f"Iter {it + 1:3d} | suggested {len(suggestions)} exp(s) | "
|
| f"best so far = {best['objective']:.4f}"
|
| )
|
|
|
|
|
| results_df = campaign.to_dataframe()
|
| best = campaign.get_best()
|
| summary = campaign.summary()
|
|
|
|
|
| fig_conv, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4.5))
|
|
|
| objs = results_df["objective"].values
|
| ax1.plot(objs, "o-", markersize=3, alpha=0.7)
|
| ax1.set_xlabel("Experiment #")
|
| ax1.set_ylabel("Objective")
|
| ax1.set_title("All Observations")
|
| ax1.grid(True, alpha=0.3)
|
|
|
| if maximize:
|
| bsf = np.maximum.accumulate(objs)
|
| else:
|
| bsf = np.minimum.accumulate(objs)
|
| ax2.plot(bsf, "s-", color="green", markersize=3)
|
| ax2.set_xlabel("Experiment #")
|
| ax2.set_ylabel("Best Objective")
|
| ax2.set_title("Convergence (Best So Far)")
|
| ax2.grid(True, alpha=0.3)
|
| fig_conv.tight_layout()
|
|
|
|
|
| fig_params = None
|
| if len(param_names) >= 2:
|
| fig_params, ax = plt.subplots(figsize=(7, 5))
|
| sc = ax.scatter(
|
| results_df[param_names[0]],
|
| results_df[param_names[1]],
|
| c=results_df["objective"],
|
| cmap="viridis",
|
| s=30,
|
| edgecolors="k",
|
| linewidths=0.5,
|
| )
|
| plt.colorbar(sc, ax=ax, label="Objective")
|
| ax.set_xlabel(param_names[0])
|
| ax.set_ylabel(param_names[1])
|
| ax.set_title("Parameter Exploration")
|
| fig_params.tight_layout()
|
|
|
|
|
| fig_surrogate = None
|
| if physics_fn is not None and campaign._designer._surrogate is not None:
|
| try:
|
| surrogate = campaign._designer._surrogate
|
| bounds = space.bounds
|
| n_grid = 150
|
|
|
|
|
| mid = (bounds[0] + bounds[1]) / 2
|
| x_range = torch.linspace(float(bounds[0, 0]), float(bounds[1, 0]), n_grid, dtype=torch.float64)
|
| X_grid = mid.unsqueeze(0).repeat(n_grid, 1)
|
| X_grid[:, 0] = x_range
|
|
|
| mean, var = surrogate.predict(X_grid)
|
| std = var.sqrt()
|
|
|
| fig_surrogate, ax = plt.subplots(figsize=(8, 5))
|
| x_np = x_range.numpy()
|
| m_np = mean.squeeze().detach().numpy()
|
| s_np = std.squeeze().detach().numpy()
|
|
|
| ax.plot(x_np, m_np, "b-", lw=2, label="Surrogate mean")
|
| ax.fill_between(x_np, m_np - 2 * s_np, m_np + 2 * s_np, alpha=0.2, color="blue", label="95% CI")
|
|
|
|
|
| phys_np = physics_fn(X_grid).detach().numpy()
|
| ax.plot(x_np, phys_np, "r--", lw=1.5, label="Physics model")
|
|
|
|
|
| if X_init is not None:
|
| ax.scatter(X_init[:, 0].numpy(), y_init.squeeze().numpy(), c="red", s=40, zorder=5, edgecolors="k", label="Initial data")
|
|
|
| ax.set_xlabel(param_names[0])
|
| ax.set_ylabel("Objective")
|
| ax.set_title(f"Surrogate vs Physics (slice along {param_names[0]})")
|
| ax.legend()
|
| ax.grid(True, alpha=0.3)
|
| fig_surrogate.tight_layout()
|
| except Exception:
|
| fig_surrogate = None
|
|
|
|
|
| log_text = "\n".join(log_lines)
|
| best_text = (
|
| f"**Best objective: {best['objective']:.4f}**\n\n"
|
| f"Parameters:\n"
|
| + "\n".join(f" - **{k}**: {v:.4f}" for k, v in best["parameters"].items())
|
| )
|
| summary_text = json.dumps(summary, indent=2, default=str)
|
|
|
| return (
|
| log_text,
|
| best_text,
|
| fig_conv,
|
| fig_params,
|
| fig_surrogate,
|
| results_df.round(4).to_string(index=False),
|
| summary_text,
|
| )
|
|
|
| except Exception as exc:
|
| tb = traceback.format_exc()
|
| err = f"**Error:** {exc}\n\n```\n{tb}\n```"
|
| return err, err, None, None, None, "", ""
|
|
|
|
|
|
|
|
|
|
|
|
|
| def on_template_change(template_name):
|
| """Populate code and params when a built-in template is selected."""
|
| info = BUILTIN_PHYSICS.get(template_name, {"code": "", "params": ""})
|
| return info["code"], info["params"]
|
|
|
|
|
| def build_app() -> gr.Blocks:
|
| with gr.Blocks(
|
| title="Physics-Informed Bayesian Optimization",
|
| theme=gr.themes.Soft(),
|
| ) as app:
|
| gr.Markdown(
|
| """
|
| # ⚗️ Physics-Informed Bayesian Optimization Platform
|
|
|
| Design experiments efficiently by combining **physics models** with
|
| **Gaussian Process surrogates**. The physics model acts as a structured prior
|
| (GP mean function), and the GP learns the residual — dramatically reducing
|
| the number of experiments needed.
|
|
|
| **Backends:** BoTorch · GPyTorch · AX · BoFire
|
| """
|
| )
|
|
|
| with gr.Tabs():
|
|
|
| with gr.TabItem("1 · Setup"):
|
| with gr.Row():
|
| with gr.Column(scale=1):
|
| gr.Markdown("### Physics Model")
|
| physics_template = gr.Dropdown(
|
| choices=list(BUILTIN_PHYSICS.keys()),
|
| value="Arrhenius Kinetics",
|
| label="Built-in template",
|
| )
|
| physics_code = gr.Code(
|
| value=BUILTIN_PHYSICS["Arrhenius Kinetics"]["code"],
|
| language="python",
|
| label="Physics model code (must define `physics_model(X)`)",
|
| lines=14,
|
| )
|
|
|
| with gr.Column(scale=1):
|
| gr.Markdown("### Parameter Space")
|
| param_text = gr.Textbox(
|
| value=BUILTIN_PHYSICS["Arrhenius Kinetics"]["params"],
|
| label="Parameters (name: lower-upper, one per line)",
|
| lines=6,
|
| )
|
|
|
| gr.Markdown("### Initial Data (optional)")
|
| csv_file = gr.File(label="Upload CSV", file_types=[".csv"])
|
| csv_text = gr.Textbox(
|
| value="",
|
| label="… or paste CSV text",
|
| lines=5,
|
| placeholder=DEMO_CSV,
|
| )
|
| objective_col = gr.Textbox(
|
| value="",
|
| label="Objective column name (leave blank → last column)",
|
| )
|
|
|
| physics_template.change(
|
| on_template_change,
|
| inputs=[physics_template],
|
| outputs=[physics_code, param_text],
|
| )
|
|
|
|
|
| with gr.TabItem("2 · Configure"):
|
| with gr.Row():
|
| acq_fn = gr.Dropdown(
|
| choices=[
|
| "Expected Improvement (EI)",
|
| "Upper Confidence Bound (UCB)",
|
| "Probability of Improvement (PI)",
|
| "Physics-Informed EI",
|
| ],
|
| value="Expected Improvement (EI)",
|
| label="Acquisition Function",
|
| )
|
| maximize = gr.Checkbox(value=True, label="Maximize objective")
|
| with gr.Row():
|
| n_initial = gr.Slider(3, 30, value=5, step=1, label="Initial samples (if no CSV)")
|
| n_iterations = gr.Slider(5, 100, value=20, step=1, label="BO iterations")
|
| batch_size = gr.Slider(1, 5, value=1, step=1, label="Batch size")
|
| with gr.Row():
|
| noise_var = gr.Slider(0.001, 1.0, value=0.01, step=0.001, label="Noise variance")
|
| seed = gr.Number(value=42, label="Random seed", precision=0)
|
|
|
|
|
| with gr.TabItem("3 · Run & Results"):
|
| run_btn = gr.Button("🚀 Run Optimization", variant="primary", size="lg")
|
|
|
| with gr.Row():
|
| best_md = gr.Markdown(label="Best Result")
|
|
|
| with gr.Row():
|
| convergence_plot = gr.Plot(label="Convergence")
|
| params_plot = gr.Plot(label="Parameter Exploration")
|
|
|
| with gr.Row():
|
| surrogate_plot = gr.Plot(label="Surrogate vs Physics")
|
|
|
| with gr.Accordion("Optimization log", open=False):
|
| log_box = gr.Textbox(label="Log", lines=15, interactive=False)
|
|
|
| with gr.Accordion("Full results table", open=False):
|
| results_box = gr.Textbox(label="Results", lines=12, interactive=False)
|
|
|
| with gr.Accordion("Campaign summary (JSON)", open=False):
|
| summary_box = gr.Textbox(label="Summary", lines=10, interactive=False)
|
|
|
| run_btn.click(
|
| run_optimization,
|
| inputs=[
|
| physics_template,
|
| physics_code,
|
| param_text,
|
| csv_file,
|
| csv_text,
|
| objective_col,
|
| acq_fn,
|
| n_initial,
|
| n_iterations,
|
| batch_size,
|
| noise_var,
|
| maximize,
|
| seed,
|
| ],
|
| outputs=[
|
| log_box,
|
| best_md,
|
| convergence_plot,
|
| params_plot,
|
| surrogate_plot,
|
| results_box,
|
| summary_box,
|
| ],
|
| )
|
|
|
|
|
| with gr.TabItem("About"):
|
| gr.Markdown(
|
| """
|
| ## How it works
|
|
|
| Traditional Bayesian optimisation uses a GP with a flat (constant) mean.
|
| This platform **replaces the mean with a physics model**:
|
|
|
| $$f(x) = \\phi(x) + \\varepsilon(x)$$
|
|
|
| where $\\phi(x)$ is the physics model and
|
| $\\varepsilon(x) \\sim \\mathcal{GP}(0,\\, k(x,x'))$ captures the
|
| residual (model discrepancy + noise).
|
|
|
| ### Benefits
|
| - **Sample efficiency** — physics captures the trend; the GP only
|
| learns small deviations.
|
| - **Extrapolation** — physics provides reasonable predictions
|
| outside observed data.
|
| - **Constraint awareness** — physical constraints steer the
|
| search toward feasible regions.
|
| - **Graceful degradation** — works physics-only (no data),
|
| hybrid, or pure GP.
|
|
|
| ### Surrogate mode selection
|
|
|
| | Data | Physics model | Mode |
|
| |------|--------------|------|
|
| | None | ✓ | `physics_only` |
|
| | < 20 | ✓ | `physics_as_mean` |
|
| | 20-50 | ✓ | `weighted_ensemble` |
|
| | Any | ✗ | `gp_only` |
|
|
|
| ### Stack
|
| **PyTorch** · **GPyTorch** · **BoTorch** · AX Platform · BoFire
|
|
|
| ---
|
| *Built by Plinity — infinite recyclable polymers*
|
| """
|
| )
|
|
|
| return app
|
|
|
|
|
|
|
|
|
|
|
|
|
| app = build_app()
|
|
|
| if __name__ == "__main__":
|
| app.launch()
|
|
|
