"""Tab 1 — Single-cycle simulation with 3-engine dropdown and diagnostics. Builds all Gradio components (left control panel + right results panel) and wires up the click handler. Must be called inside an active ``gr.TabItem`` context. Parameter surface mirrors the original DR MURPHY dashboard: • Operating Conditions (always visible) • ICV Parameters accordion (9 inputs) • DCV Parameters accordion (9 inputs) • Pump Geometry accordion (12 inputs) • Process / Losses accordion (8 + flash_eff) • Downstream / Fill Mode (fill_type, num_cycles, snubber, CHSS, AOV, RO dia) """ from __future__ import annotations import os import tempfile import traceback from datetime import datetime, timezone import gradio as gr import numpy as np import plotly.graph_objects as go from murphy_unified.engine_bridge import cycle_sim, ENGINES from murphy_unified.diagnostics import evaluate from murphy_unified.engine_params import build_engine_kwargs from murphy_unified.params_io import params_to_csv, csv_to_params from murphy_unified.sim_defaults import SIM_DEFAULTS from murphy_unified.theme import ( metric_html, TRACE_COLORS, TEXT_DIM, TEXT_BRIGHT, ACCENT, ACCENT2, WARN, DANGER, FONT_MONO, VALVE_ICV, VALVE_DCV, ENERGY_POS, ENERGY_NEG, TEXT, ) from murphy_unified.tab_help import tagline_html # ── Engine name mapping ────────────────────────────────────────────────────── _ENGINE_MAP = { "Euler": "euler", "Lightspeed": "lightspeed", "General-ODE": "general_ode", } _FLUID_MAP = {"H2": "h2", "N2": "n2"} # ── Extra plot definitions ────────────────────────────────────────────────── # Each entry: (hist_key, y-axis label, trace color index) # "Energy Balance" is a special case (bar chart, not time-series). _EXTRA_PLOT_DEFS: dict[str, dict] = { "Energy Balance": {"type": "bar"}, "Chamber Density": {"key": "den", "unit": "kg/m³", "color": 2}, "Chamber Mass": {"key": "mc_g", "unit": "g", "color": 3}, "Piston Position": {"key": "yp", "unit": "norm", "color": 4}, "Specific Enthalpy": {"key": "hc", "unit": "kJ/kg", "color": 5}, "Mass Flow Rate": {"key": "dm_tot_kgps", "unit": "kg/s", "color": 0}, } EXTRA_PLOT_CHOICES = list(_EXTRA_PLOT_DEFS.keys()) def _build_energy_balance_fig(hist: dict) -> go.Figure: """Build the energy-balance horizontal bar chart.""" eb = hist.get("energy_breakdown", {}) engine = hist.get("engine", "") # Energy breakdown only available from Euler engine if not eb or all(eb.get(k, 0.0) == 0.0 for k in ( "Qig_kJ", "Qf_kJ", "work_extend_kJ", "dh_DCV_kJ", )): fig = go.Figure() fig.add_annotation( text=f"Energy breakdown not available for {engine} engine.
" "Switch to Euler to see energy balance.", xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False, font=dict(family=FONT_MONO, size=12, color=TEXT), ) fig.update_layout(height=300) return fig eb_keys = [ "Qig_kJ", "Qf_kJ", "Qtmass_kJ", "work_extend_kJ", "work_retract_kJ", "dh_DCV_kJ", "dh_ICV_kJ", "dh_bb_kJ", ] eb_labels = [ "Heat Ingress", "Friction", "Thermal Mass", "pV Work (Extend)", "pV Work (Retract)", "DCV Flow", "ICV Flow", "Blowby", ] eb_values = [eb.get(k, 0.0) for k in eb_keys] eb_colors = [ENERGY_POS if v >= 0 else ENERGY_NEG for v in eb_values] fig = go.Figure(go.Bar( y=eb_labels, x=eb_values, orientation="h", marker_color=eb_colors, text=[f"{v:.4f}" for v in eb_values], textposition="outside", textfont=dict(family=FONT_MONO, size=10, color=TEXT_BRIGHT), )) fig.update_layout( title="Energy Balance Breakdown — per cycle", xaxis_title="Energy [kJ]", height=300, xaxis=dict(tickfont=dict(family=FONT_MONO, size=10, color=TEXT)), yaxis=dict(tickfont=dict(family=FONT_MONO, size=10, color=TEXT)), margin=dict(l=130), ) return fig def _build_timeseries_fig( name: str, hist: dict, engine: str, Pexit: float, ) -> go.Figure: """Build a single time-series plot for an extra variable.""" defn = _EXTRA_PLOT_DEFS[name] key = defn["key"] unit = defn["unit"] color = TRACE_COLORS[defn["color"]] angle = hist.get("angle_deg", np.array([])) data = hist.get(key, []) fig = go.Figure() fig.add_trace(go.Scatter( x=angle, y=data, name=name, line=dict(color=color, width=1.5), )) fig.update_layout( title=f"{name} — {engine} @ {Pexit:.0f} barg", xaxis_title="Crank angle [deg]", yaxis_title=f"{name} [{unit}]", height=300, xaxis=dict(tickfont=dict(family=FONT_MONO, size=10, color=TEXT)), yaxis=dict(tickfont=dict(family=FONT_MONO, size=10, color=TEXT)), legend=dict(x=0.01, y=0.99, bgcolor="rgba(0,0,0,0)"), ) return fig def _empty_fig(height: int = 200) -> go.Figure: fig = go.Figure() fig.update_layout(height=height) return fig # ── Parameter summary HTML ────────────────────────────────────────────────── def _fmt_val(v): """Format a parameter value compactly.""" if isinstance(v, float) and v != 0 and (abs(v) < 0.001 or abs(v) >= 1e6): return f"{v:.4g}" if isinstance(v, float): return f"{v:g}" return str(v) def _build_params_summary_html( Pexit, speed_f, Ptank, Psat, fluid, engine, icv_port, icv_mass, icv_Fs, icv_SC, icv_leakKv, dcv_port, dcv_mass, dcv_Fs, dcv_SC, dcv_leakKv, bore, stroke, dvf, cpm, KvBB, fric, Exp, flash_eff, fill_type, vsnubber, vchss, aov140f, rodia, n_cyc, ) -> str: """Compact 5-line parameter summary (mirrors original scenario_gui layout).""" _f = _fmt_val return ( '
' 'Base Parameters
' f'Operating: engine={engine}, Pexit={_f(Pexit)} barg, ' f'speed={_f(speed_f)}, Ptank={_f(Ptank)}, Psat={_f(Psat)}, fluid={fluid}
' f'ICV: port={_f(icv_port)}mm, mass={_f(icv_mass)}g, ' f'Fs={_f(icv_Fs)}N, SC={_f(icv_SC)}, leakKv={_f(icv_leakKv)}
' f'DCV: port={_f(dcv_port)}mm, mass={_f(dcv_mass)}g, ' f'Fs={_f(dcv_Fs)}N, SC={_f(dcv_SC)}, leakKv={_f(dcv_leakKv)}
' f'Pump: bore={_f(bore)}mm, stroke={_f(stroke)}mm, ' f'dvf={_f(dvf)}, cpm={_f(cpm)}
' f'Process: Kv_BB={_f(KvBB)}, fric={_f(fric)}, ' f'Exp_eff={_f(Exp)}, flash_eff={_f(flash_eff)}
' f'Downstream: mode={fill_type}, snubber={_f(vsnubber)}L, ' f'CHSS={_f(vchss)}L, AOV140={_f(aov140f)}, RO_dia={_f(rodia)}mm, ' f'n_cycles={_f(n_cyc)}' '
' ) # ── Click handler ──────────────────────────────────────────────────────────── def run_simulation( # Operating conditions engine_name: str, fluid: str, Pexit: float, speed: float, Ptank: float, Psat: float, # ICV (9) icv_port: float, icv_mass: float, icv_travel: float, icv_dp_area: float, icv_Fs: float, icv_SC: float, icv_leakKv: float, icv_comp_eff: float, icv_npts: float, # DCV (9) dcv_port: float, dcv_mass: float, dcv_travel: float, dcv_dp_area: float, dcv_Fs: float, dcv_SC: float, dcv_leakKv: float, dcv_comp_eff: float, dcv_npts: float, # Pump geometry (12) bore: float, stroke: float, hOD: float, cLen: float, emH: float, emS: float, kvoid: float, kH: float, Vfv: float, vac: float, cpm: float, dvf: float, # Process / losses (8 + flash_eff) Tamb: float, htc: float, drive: float, Fmult: float, KvBB: float, Pbb: float, fric: float, Exp: float, flash_eff: float, # Downstream / Fill mode fill_type: str, num_cycles: float, vsnubber: float, vchss: float, aov140f: float, rodia: float, # Output selection extra_plots: list[str], ): """Run one pump cycle with the full parameter set and return plots + diagnostics.""" # ── Snapshot submitted params BEFORE the engine call so sim_params_state # reflects user inputs even if cycle_sim raises. sim_state: dict = { "engine_name": engine_name, "fluid": fluid, "Pexit": Pexit, "speed": speed, "Ptank": Ptank, "Psat": Psat, "icv_port": icv_port, "icv_mass": icv_mass, "icv_travel": icv_travel, "icv_dp_area": icv_dp_area, "icv_Fs": icv_Fs, "icv_SC": icv_SC, "icv_leakKv": icv_leakKv, "icv_comp_eff": icv_comp_eff, "icv_npts": icv_npts, "dcv_port": dcv_port, "dcv_mass": dcv_mass, "dcv_travel": dcv_travel, "dcv_dp_area": dcv_dp_area, "dcv_Fs": dcv_Fs, "dcv_SC": dcv_SC, "dcv_leakKv": dcv_leakKv, "dcv_comp_eff": dcv_comp_eff, "dcv_npts": dcv_npts, "bore": bore, "stroke": stroke, "hOD": hOD, "cLen": cLen, "emH": emH, "emS": emS, "kvoid": kvoid, "kH": kH, "Vfv": Vfv, "vac": vac, "cpm": cpm, "dvf": dvf, "Tamb": Tamb, "htc": htc, "drive": drive, "Fmult": Fmult, "KvBB": KvBB, "Pbb": Pbb, "fric": fric, "Exp": Exp, "flash_eff": flash_eff, "fill_type": fill_type, "num_cycles": num_cycles, "vsnubber": vsnubber, "vchss": vchss, "aov140f": aov140f, "rodia": rodia, "extra_plots": list(extra_plots or []), "_written_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "_engine_ok": True, } engine_id = _ENGINE_MAP.get(engine_name, "euler") fluid_id = _FLUID_MAP.get(fluid, "h2") n_cyc = int(num_cycles) if num_cycles else 1 # ── Parameter summary (shown above plots on every run) ────────────── params_html = _build_params_summary_html( Pexit, speed, Ptank, Psat, fluid_id, engine_id, icv_port, icv_mass, icv_Fs, icv_SC, icv_leakKv, dcv_port, dcv_mass, dcv_Fs, dcv_SC, dcv_leakKv, bore, stroke, dvf, cpm, KvBB, fric, Exp, flash_eff, fill_type, vsnubber, vchss, aov140f, rodia, n_cyc, ) # ── Call engine bridge via shared helper ───────────────────────────── engine_kwargs = build_engine_kwargs(sim_state) try: out, hist = cycle_sim(engine=engine_id, fluid=fluid_id, **engine_kwargs) except Exception: err = traceback.format_exc() err_html = ( f'
' f'Simulation failed:\n{err}
' ) ef = _empty_fig() sim_state["_engine_ok"] = False return err_html, params_html, ef, ef, ef, ef, ef, err_html, sim_state, None # ── Extract scalars (prefer hist dict) ─────────────────────────────── mdot = hist.get("mdot_kgpm", 0.0) mass_eff = hist.get("mass_eff", 0.0) Tc_peak = ( hist["Tc_peak_K"] if hist.get("Tc_peak_K") is not None else (float(np.max(hist["Tc_K"])) if "Tc_K" in hist else 0.0) ) pc_peak = ( hist["pc_peak_barg"] if hist.get("pc_peak_barg") is not None else (float(np.max(hist["pc"])) if "pc" in hist else 0.0) ) kWh_ext = hist.get("kWh_extend") cpu_s = hist.get("cpu_s", 0.0) steps = int(hist.get("steps", len(hist.get("angle_deg", [])))) actual_engine = hist.get("engine", engine_id) fallback = hist.get("fallback") # Scalars for diagnostics scalars = { "mdot_kgpm": mdot, "mass_eff": mass_eff, "Tc_peak_K": Tc_peak, "pc_peak_barg": pc_peak, } if "ICV_open_frac" in hist: scalars["ICVmax_open_frac"] = float(np.max(hist["ICV_open_frac"])) if "DCV_ct_s" in hist: scalars["DCV_ct_s"] = hist["DCV_ct_s"] findings = evaluate(scalars) # ── Metric cards HTML ──────────────────────────────────────────────── eff_accent = "green" if mass_eff > 0.5 else "amber" kWh_str = f"{kWh_ext:.3f}" if kWh_ext is not None else "N/A" bb_kgpm = hist.get("bb_kgpm") bb_str = f"{bb_kgpm:.4f}" if bb_kgpm is not None else "N/A" metrics_html = ( '
' + metric_html("MASS FLOW", f"{mdot:.3f}", "kg/min") + metric_html("EFFICIENCY", f"{mass_eff:.1%}", "", eff_accent) + metric_html("PEAK TEMP", f"{Tc_peak:.1f}", "K") + "
" + '
' + metric_html("PEAK PRESSURE", f"{pc_peak:.0f}", "barg") + metric_html("SPECIFIC ENERGY", kWh_str, "kWh/kg") + metric_html("BLOWBY", bb_str, "kg/min") + metric_html("CPU TIME", f"{cpu_s:.2f}", "s") + "
" ) # ── Pressure plot (issue #1: separate plots, not cluttered overlay) ── angle = hist.get("angle_deg", np.array([])) fig_pressure = go.Figure() fig_pressure.add_trace(go.Scatter( x=angle, y=hist.get("pc", []), name="Chamber P", line=dict(color=TRACE_COLORS[0], width=1.5), )) fig_pressure.update_layout( title=f"Chamber Pressure — {actual_engine} @ {Pexit:.0f} barg", xaxis_title="Crank angle [deg]", yaxis_title="Pressure [barg]", height=320, ) # ── Temperature plot ──────────────────────────────────────────────── fig_temp = go.Figure() fig_temp.add_trace(go.Scatter( x=angle, y=hist.get("Tc_K", []), name="Chamber T", line=dict(color=TRACE_COLORS[1], width=1.5), )) fig_temp.update_layout( title=f"Chamber Temperature — {actual_engine} @ {Pexit:.0f} barg", xaxis_title="Crank angle [deg]", yaxis_title="Temperature [K]", height=320, ) # ── Valve dynamics plot ────────────────────────────────────────────── fig_valves = go.Figure() fig_valves.add_trace(go.Scatter( x=angle, y=hist.get("ICV_open_frac", []), name="ICV open frac", line=dict(color=VALVE_ICV, width=1.5), )) fig_valves.add_trace(go.Scatter( x=angle, y=hist.get("DCV_open_frac", []), name="DCV open frac", line=dict(color=VALVE_DCV, width=1.5), )) if "ICV_leak_kgpm" in hist: fig_valves.add_trace(go.Scatter( x=angle, y=hist["ICV_leak_kgpm"], name="ICV leak [kg/min]", yaxis="y2", line=dict(color=VALVE_ICV, width=1, dash="dash"), )) if "DCV_leak_kgpm" in hist: fig_valves.add_trace(go.Scatter( x=angle, y=hist["DCV_leak_kgpm"], name="DCV leak [kg/min]", yaxis="y2", line=dict(color=VALVE_DCV, width=1, dash="dash"), )) fig_valves.update_layout( title=f"Valve Dynamics — {actual_engine} @ {Pexit:.0f} barg", xaxis_title="Crank angle [deg]", yaxis_title="Open fraction", yaxis2=dict( title="Leak flow [kg/min]", overlaying="y", side="right", tickfont=dict(family=FONT_MONO, size=10, color=TRACE_COLORS[3]), ), height=300, legend=dict(x=0.01, y=0.99, bgcolor="rgba(0,0,0,0)"), ) # ── Extra plots (up to 2) ─────────────────────────────────────────── selected = (extra_plots or [])[:2] extra_figs: list[go.Figure] = [] for name in selected: defn = _EXTRA_PLOT_DEFS.get(name) if defn is None: extra_figs.append(_empty_fig(300)) elif defn.get("type") == "bar": extra_figs.append(_build_energy_balance_fig(hist)) else: extra_figs.append( _build_timeseries_fig(name, hist, actual_engine, Pexit) ) # Pad to exactly 2 while len(extra_figs) < 2: extra_figs.append(_empty_fig(300)) # ── Status bar ─────────────────────────────────────────────────────── parts = [ f"ENGINE: {actual_engine}", f"{Pexit:.0f} barg", f"speed {speed:.2f}", f"{steps:,} steps in {cpu_s:.2f}s", ] if fill_type != "Fixed Pexit": parts.append(f"MODE: {fill_type}") if n_cyc > 1: parts.append(f"n_cycles={n_cyc}") if fallback: parts.append(f"FALLBACK: {fallback}") for f in findings: sev = f["severity"] pill_color = WARN if sev == "WARNING" else DANGER parts.append( f'' f'{sev}: {f["name"]} ({f["field"]}={f["actual"]:.3g})' f"" ) status_html = f'
{" | ".join(parts)}
' # Store results for export (issue #2) results_bundle = { "scalars": { "mdot_kgpm": mdot, "mass_eff": mass_eff, "Tc_peak_K": Tc_peak, "pc_peak_barg": pc_peak, "kWh_extend": kWh_ext, "bb_kgpm": hist.get("bb_kgpm"), "cpu_s": cpu_s, "steps": steps, "engine": actual_engine, }, "time_series": {k: v for k, v in hist.items() if isinstance(v, np.ndarray)}, "params": {k: v for k, v in sim_state.items() if not k.startswith("_")}, } return (metrics_html, params_html, fig_pressure, fig_temp, fig_valves, extra_figs[0], extra_figs[1], status_html, sim_state, results_bundle) # ── Builder ────────────────────────────────────────────────────────────────── _WIDGET_KEYS_IN_ORDER = ( "engine_name", "fluid", "Pexit", "speed", "Ptank", "Psat", "icv_port", "icv_mass", "icv_travel", "icv_dp_area", "icv_Fs", "icv_SC", "icv_leakKv", "icv_comp_eff", "icv_npts", "dcv_port", "dcv_mass", "dcv_travel", "dcv_dp_area", "dcv_Fs", "dcv_SC", "dcv_leakKv", "dcv_comp_eff", "dcv_npts", "bore", "stroke", "hOD", "cLen", "emH", "emS", "kvoid", "kH", "Vfv", "vac", "cpm", "dvf", "Tamb", "htc", "drive", "Fmult", "KvBB", "Pbb", "fric", "Exp", "flash_eff", "fill_type", "num_cycles", "vsnubber", "vchss", "aov140f", "rodia", "extra_plots", ) def _export_params(*widget_values) -> str: """Write current widget state to a tempfile CSV; return path for gr.File.""" state = dict(zip(_WIDGET_KEYS_IN_ORDER, widget_values)) state["extra_plots"] = list(state.get("extra_plots") or []) csv_text = params_to_csv(state) stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") path = os.path.join(tempfile.gettempdir(), f"murphy_run_{stamp}.csv") with open(path, "w", encoding="utf-8", newline="") as f: f.write(csv_text) return path def _import_params(file_obj): """Load a CSV and return widget updates in _WIDGET_KEYS_IN_ORDER + info HTML.""" if file_obj is None: return (*([gr.update()] * len(_WIDGET_KEYS_IN_ORDER)), "") path = file_obj if isinstance(file_obj, str) else file_obj.name with open(path, "r", encoding="utf-8") as f: csv_text = f.read() parsed, skipped = csv_to_params(csv_text) updates = tuple( gr.update(value=parsed[key]) if key in parsed else gr.update() for key in _WIDGET_KEYS_IN_ORDER ) loaded = len(parsed) total = len(SIM_DEFAULTS) skipped_msg = f" Skipped: {', '.join(skipped)}" if skipped else "" info_html = ( f'
' f'Loaded {loaded}/{total} params.{skipped_msg}
' ) return (*updates, info_html) def build_simulation_tab(sim_params_state: gr.State): """Create all Gradio components for the Simulation tab and wire events. Must be called inside an active ``gr.TabItem(...)`` context manager. """ gr.HTML(tagline_html("simulation")) with gr.Row(): # ── Left panel — controls ──────────────────────────────────────── with gr.Column(scale=1, min_width=300): # ── Operating conditions (always visible) ─────────────────── engine_dd = gr.Dropdown( choices=["Euler", "Lightspeed", "General-ODE"], value=SIM_DEFAULTS["engine_name"], label="Engine", ) fluid_dd = gr.Dropdown( choices=["H2", "N2"], value=SIM_DEFAULTS["fluid"], label="Fluid", ) pexit_sl = gr.Slider( minimum=50, maximum=1100, value=SIM_DEFAULTS["Pexit"], step=10, label="Exit Pressure [barg]", ) speed_sl = gr.Slider( minimum=0.1, maximum=1.0, value=SIM_DEFAULTS["speed"], step=0.05, label="Speed Fraction", ) with gr.Row(): ptank_num = gr.Number(value=SIM_DEFAULTS["Ptank"], label="Tank P [barg]", step=0.1) psat_num = gr.Number(value=SIM_DEFAULTS["Psat"], label="Sat P [barg]", step=0.1) run_btn = gr.Button("RUN CYCLE", variant="primary") # ── Save / Load run params ────────────────────────────────── with gr.Row(): save_btn = gr.DownloadButton("Save params to CSV") load_file = gr.File( label="Load params from CSV", file_types=[".csv"], type="filepath", ) load_info_html = gr.HTML() # ── ICV Parameters ────────────────────────────────────────── with gr.Accordion("ICV Parameters", open=False): with gr.Row(): icv_port = gr.Number(value=SIM_DEFAULTS['icv_port'], label="Port Dia (mm)", step=0.1) icv_mass = gr.Number(value=SIM_DEFAULTS['icv_mass'], label="Mass (g)", step=0.1) icv_travel = gr.Number(value=SIM_DEFAULTS['icv_travel'], label="Travel (mm)", step=0.1) with gr.Row(): icv_dparea = gr.Number(value=SIM_DEFAULTS['icv_dp_area'], label="dP Area (mm²)", step=0.1) icv_Fs = gr.Number(value=SIM_DEFAULTS['icv_Fs'], label="Spring F (N)", step=0.1) icv_SC = gr.Number(value=SIM_DEFAULTS['icv_SC'], label="Spring K (N/mm)", step=0.01) with gr.Row(): icv_leak = gr.Number(value=SIM_DEFAULTS['icv_leakKv'], label="Leak Kv (m³/hr)", step=1e-7) icv_ceff = gr.Number(value=SIM_DEFAULTS['icv_comp_eff'], label="Comp Eff", step=0.01) icv_npts = gr.Number(value=SIM_DEFAULTS['icv_npts'], label="Npts", step=1, precision=0) # ── DCV Parameters ────────────────────────────────────────── with gr.Accordion("DCV Parameters", open=False): with gr.Row(): dcv_port = gr.Number(value=SIM_DEFAULTS['dcv_port'], label="Port Dia (mm)", step=0.1) dcv_mass = gr.Number(value=SIM_DEFAULTS['dcv_mass'], label="Mass (g)", step=0.1) dcv_travel = gr.Number(value=SIM_DEFAULTS['dcv_travel'], label="Travel (mm)", step=0.01) with gr.Row(): dcv_dparea = gr.Number(value=SIM_DEFAULTS['dcv_dp_area'], label="dP Area (mm²)", step=0.01) dcv_Fs = gr.Number(value=SIM_DEFAULTS['dcv_Fs'], label="Spring F (N)", step=0.1) dcv_SC = gr.Number(value=SIM_DEFAULTS['dcv_SC'], label="Spring K (N/mm)", step=0.001) with gr.Row(): dcv_leak = gr.Number(value=SIM_DEFAULTS['dcv_leakKv'], label="Leak Kv (m³/hr)", step=1e-7) dcv_ceff = gr.Number(value=SIM_DEFAULTS['dcv_comp_eff'], label="Comp Eff", step=0.01) dcv_npts = gr.Number(value=SIM_DEFAULTS['dcv_npts'], label="Npts", step=1, precision=0) # ── Pump Geometry ─────────────────────────────────────────── with gr.Accordion("Pump Geometry", open=False): with gr.Row(): bore = gr.Number(value=SIM_DEFAULTS['bore'], label="Bore (mm)", step=0.1) stroke = gr.Number(value=SIM_DEFAULTS['stroke'], label="Stroke (mm)", step=0.1) with gr.Row(): cpm = gr.Number(value=SIM_DEFAULTS['cpm'], label="Speed (cpm)", step=1) dvf = gr.Number(value=SIM_DEFAULTS['dvf'], label="Dead Vol Frac", step=0.001) with gr.Row(): hOD = gr.Number(value=SIM_DEFAULTS['hOD'], label="Housing OD (mm)", step=1) cLen = gr.Number(value=SIM_DEFAULTS['cLen'], label="Chamber Len (mm)", step=1) with gr.Row(): emH = gr.Number(value=SIM_DEFAULTS['emH'], label="ε Housing", step=0.01) emS = gr.Number(value=SIM_DEFAULTS['emS'], label="ε Shield", step=0.01) with gr.Row(): kvoid = gr.Number(value=SIM_DEFAULTS['kvoid'], label="k void (W/m/K)", step=0.001) kH = gr.Number(value=SIM_DEFAULTS['kH'], label="k housing (W/m/K)", step=0.1) with gr.Row(): Vfv = gr.Number(value=SIM_DEFAULTS['Vfv'], label="Void Vol Frac", step=0.1) vac = gr.Number(value=SIM_DEFAULTS['vac'], label="Vacuum (µHg)", step=1000) # ── Process / Losses ──────────────────────────────────────── with gr.Accordion("Process / Losses", open=False): with gr.Row(): Tamb = gr.Number(value=SIM_DEFAULTS['Tamb'], label="T_amb (K)", step=1) htc = gr.Number(value=SIM_DEFAULTS['htc'], label="HTC (W/m²/K)", step=0.1) with gr.Row(): drive = gr.Number(value=SIM_DEFAULTS['drive'], label="Drive (kgf)", step=0.1) Fmult = gr.Number(value=SIM_DEFAULTS['Fmult'], label="F Multiplier", step=1) with gr.Row(): fric = gr.Number(value=SIM_DEFAULTS['fric'], label="Friction (0-1)", step=0.01) KvBB = gr.Number(value=SIM_DEFAULTS['KvBB'], label="BB Kv (m³/hr)", step=0.001) with gr.Row(): Pbb = gr.Number(value=SIM_DEFAULTS['Pbb'], label="BB Exit P (barg)", step=0.1) Exp = gr.Number(value=SIM_DEFAULTS['Exp'], label="Exp Eff (0-1)", step=0.01) flash_eff = gr.Number(value=SIM_DEFAULTS["flash_eff"], label="Thermal Mass Eff (0-1)", step=0.01) # ── Downstream / Fill mode ────────────────────────────────── with gr.Accordion("Downstream / Fill Mode", open=False): with gr.Row(): fill_type = gr.Dropdown( choices=["Fixed Pexit", "CHSS Fill", "RO Vent"], value=SIM_DEFAULTS["fill_type"], label="Fill Type", ) num_cycles = gr.Number( value=SIM_DEFAULTS["num_cycles"], label="No. cycles", step=1, minimum=1, maximum=50, precision=0, ) with gr.Row(): vsnubber = gr.Number(value=SIM_DEFAULTS["vsnubber"], label="Snubber (L)", step=0.1) vchss = gr.Number(value=SIM_DEFAULTS["vchss"], label="CHSS (L)", step=1) with gr.Row(): aov140f = gr.Number(value=SIM_DEFAULTS["aov140f"], label="AOV140 (0-1)", step=0.01, minimum=0, maximum=1) rodia = gr.Number(value=SIM_DEFAULTS["rodia"], label="RO Dia (mm)", step=0.01) # ── Extra plots selector ──────────────────────────────────── gr.Markdown("**Additional plots** (select up to 2):") extras_cb = gr.CheckboxGroup( choices=EXTRA_PLOT_CHOICES, value=SIM_DEFAULTS["extra_plots"], label="Extra Graphs", ) # ── Right panel — results ──────────────────────────────────────── with gr.Column(scale=3): metrics_out = gr.HTML(label="Metrics") params_out = gr.HTML(label="Parameter Summary") with gr.Row(): fig_pressure_out = gr.Plot(label="Chamber Pressure") fig_temp_out = gr.Plot(label="Chamber Temperature") fig_valves_out = gr.Plot(label="Valve Dynamics") with gr.Row(): fig_extra1_out = gr.Plot(label="Extra Plot 1") fig_extra2_out = gr.Plot(label="Extra Plot 2") status_out = gr.HTML(label="Status") # ── Results export (issue #2) ─────────────────────────────── with gr.Accordion("Export Results", open=False): export_sections_cb = gr.CheckboxGroup( choices=["Key Results", "Time Series", "Parameters"], value=["Key Results", "Time Series"], label="Sections to include", ) export_results_btn = gr.DownloadButton( "Export selected sections (CSV)", size="sm", ) # ── Hidden state for last run results ─────────────────────────────── sim_results_state = gr.State(None) # ── Wire click event ───────────────────────────────────────────────── run_btn.click( fn=run_simulation, inputs=[ # Operating (6) engine_dd, fluid_dd, pexit_sl, speed_sl, ptank_num, psat_num, # ICV (9) icv_port, icv_mass, icv_travel, icv_dparea, icv_Fs, icv_SC, icv_leak, icv_ceff, icv_npts, # DCV (9) dcv_port, dcv_mass, dcv_travel, dcv_dparea, dcv_Fs, dcv_SC, dcv_leak, dcv_ceff, dcv_npts, # Pump geometry (12) bore, stroke, hOD, cLen, emH, emS, kvoid, kH, Vfv, vac, cpm, dvf, # Process / losses (9) Tamb, htc, drive, Fmult, KvBB, Pbb, fric, Exp, flash_eff, # Downstream (6) fill_type, num_cycles, vsnubber, vchss, aov140f, rodia, # Output selector extras_cb, ], outputs=[metrics_out, params_out, fig_pressure_out, fig_temp_out, fig_valves_out, fig_extra1_out, fig_extra2_out, status_out, sim_params_state, sim_results_state], ) # ── Widget list (order must match _WIDGET_KEYS_IN_ORDER) ───────────── _all_widgets = [ engine_dd, fluid_dd, pexit_sl, speed_sl, ptank_num, psat_num, icv_port, icv_mass, icv_travel, icv_dparea, icv_Fs, icv_SC, icv_leak, icv_ceff, icv_npts, dcv_port, dcv_mass, dcv_travel, dcv_dparea, dcv_Fs, dcv_SC, dcv_leak, dcv_ceff, dcv_npts, bore, stroke, hOD, cLen, emH, emS, kvoid, kH, Vfv, vac, cpm, dvf, Tamb, htc, drive, Fmult, KvBB, Pbb, fric, Exp, flash_eff, fill_type, num_cycles, vsnubber, vchss, aov140f, rodia, extras_cb, ] save_btn.click( fn=_export_params, inputs=_all_widgets, outputs=[save_btn], ) load_file.change( fn=_import_params, inputs=[load_file], outputs=[*_all_widgets, load_info_html], ) # ── Live sync: any widget edit → update sim_params_state ───────────── # Keeps the Sweep-tab baseline panel in sync with current widget values # without requiring the user to click RUN first (issue #11). def _snapshot_state(*widget_values) -> dict: state = dict(zip(_WIDGET_KEYS_IN_ORDER, widget_values)) state["extra_plots"] = list(state.get("extra_plots") or []) state["_written_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds") + " (widgets)" state["_engine_ok"] = None return state gr.on( triggers=[w.change for w in _all_widgets], fn=_snapshot_state, inputs=_all_widgets, outputs=[sim_params_state], ) # ── Results export handler (issue #2) ──────────────────────────────── import pandas as pd def _export_results(results: dict | None, sections: list[str]) -> str | None: if results is None: return None stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") path = os.path.join(tempfile.gettempdir(), f"murphy_results_{stamp}.xlsx") with pd.ExcelWriter(path, engine="openpyxl") as writer: if "Key Results" in sections and "scalars" in results: s = results["scalars"] df = pd.DataFrame([ {"Metric": k, "Value": v} for k, v in s.items() if v is not None ]) df.to_excel(writer, sheet_name="Key Results", index=False) if "Time Series" in sections and "time_series" in results: ts = results["time_series"] max_len = max((len(v) for v in ts.values()), default=0) ts_df = pd.DataFrame({ k: np.pad(v, (0, max_len - len(v)), constant_values=np.nan) for k, v in ts.items() }) ts_df.to_excel(writer, sheet_name="Time Series", index=False) if "Parameters" in sections and "params" in results: p = results["params"] df = pd.DataFrame([ {"Parameter": k, "Value": str(v)} for k, v in p.items() ]) df.to_excel(writer, sheet_name="Parameters", index=False) return path export_results_btn.click( fn=_export_results, inputs=[sim_results_state, export_sections_cb], outputs=[export_results_btn], )