"""MILP Energy Lab - interactive experiment lab (AGILAB native React). Derived from the PyPSA 'Modular Expansion with Unit Commitment' tutorial (CC BY 4.0, PyPSA contributors). Attribution and original-source link below. """ from __future__ import annotations import json import os import signal import subprocess import sys import tempfile import time import altair as alt import pandas as pd from agi_web import python_ui as st sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from energy_core import ( default_settings, make_batch, model_text, validate_settings, ) from energy_runner import cpu_limits SOURCE_URL = ( "https://github.com/PyPSA/PyPSA/blob/" "c838aa498557cc8e27a9d3ed10d45e35c4b0b442/docs/examples/modular-committable.ipynb" ) st.title("MILP Energy Lab") st.caption( "Interactive lab for modular gas-generator expansion with unit commitment, " "solved with HiGHS (one thread)." ) st.markdown( "**How to use:** set parameters in *Experiment*, press **Run analysis**, " "then inspect, compare and scale. Results are labelled with the inputs " "that produced them." ) # ---- session state --------------------------------------------------------- for key, val in { "milp_energy_settings": default_settings(), "milp_energy_result": None, "milp_energy_saved": [], "milp_energy_benchmark": None, "milp_energy_busy": False, }.items(): st.session_state.setdefault(key, val) def _child_env() -> dict: env = dict(os.environ) for var in ( "OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS", "NUMEXPR_NUM_THREADS", "VECLIB_MAXIMUM_THREADS", ): env[var] = "1" project_root = str(os.path.dirname(os.path.abspath(__file__))) if project_root not in env.get("PYTHONPATH", ""): env["PYTHONPATH"] = project_root + os.pathsep + env.get("PYTHONPATH", "") return env def _kill_group(proc: subprocess.Popen) -> None: try: os.killpg(os.getpgid(proc.pid), signal.SIGKILL) except Exception: try: proc.kill() except Exception: pass def _run_supervised(code: str, args: list[str], timeout: float = 150.0) -> dict: """Run code in an isolated child process group with a bounded deadline.""" with tempfile.TemporaryDirectory(prefix="milp-ui-") as tmp: proc = subprocess.Popen( [sys.executable, "-c", code] + args, cwd=tmp, env=_child_env(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True, ) try: out, err = proc.communicate(timeout=timeout) except subprocess.TimeoutExpired: _kill_group(proc) proc.communicate() return {"status": "error", "error": "timeout"} if proc.returncode != 0: return {"status": "error", "error": err.decode()[-2000:]} try: return json.loads(out.decode()) except json.JSONDecodeError: return {"status": "error", "error": "bad output"} def _apply_controls(): s = st.session_state.milp_energy_settings st.selectbox("Hours", [4, 12, 24], index=[4, 12, 24].index(s["hours"]), key="milp_energy_hours") st.number_input("Demand multiplier", 0.0, 10.0, s["demand_multiplier"], 0.1, key="milp_energy_dm") st.number_input("Module MW", 1.0, 10000.0, s["module_mw"], 50.0, key="milp_energy_mod") st.number_input("Max modules", 1, 200, s["max_modules"], 1, key="milp_energy_max") st.number_input("Investment cost", 0.0, 1e6, s["investment_cost"], 1.0, key="milp_energy_inv") st.number_input("Marginal cost", 0.0, 1e6, s["marginal_cost"], 1.0, key="milp_energy_marg") st.number_input("Startup cost", 0.0, 1e6, s["startup_cost"], 10.0, key="milp_energy_start") st.number_input("Standby cost", 0.0, 1e6, s["standby_cost"], 1.0, key="milp_energy_stand") st.slider("Minimum loading", 0.0, 1.0, s["min_loading"], 0.05, key="milp_energy_min") st.number_input("Solar capacity MW", 0.0, 1e6, s["solar_capacity"], 100.0, key="milp_energy_solar") st.toggle("Allow load shedding", s["allow_shedding"], key="milp_energy_shed") def _collect_settings() -> dict: return { "hours": st.session_state.milp_energy_hours, "demand_multiplier": float(st.session_state.milp_energy_dm), "module_mw": float(st.session_state.milp_energy_mod), "max_modules": int(st.session_state.milp_energy_max), "investment_cost": float(st.session_state.milp_energy_inv), "marginal_cost": float(st.session_state.milp_energy_marg), "startup_cost": float(st.session_state.milp_energy_start), "standby_cost": float(st.session_state.milp_energy_stand), "min_loading": float(st.session_state.milp_energy_min), "solar_capacity": float(st.session_state.milp_energy_solar), "allow_shedding": bool(st.session_state.milp_energy_shed), } def _df(result): return pd.DataFrame({ "Hour": list(range(len(result["demand"]))), "Demand": result["demand"], "Dispatch": result["dispatch"], "Solar": result["solar"], "Shed": result["shed"], "Active": result["active_modules"], }) def _charts(result): df = _df(result) long = df.melt(id_vars="Hour", value_vars=["Demand", "Dispatch", "Solar", "Shed"], var_name="Series", value_name="MW") c1 = alt.Chart(long).mark_bar().encode( x=alt.X("Hour:O"), y=alt.Y("MW:Q"), color=alt.Color("Series:N", scale=alt.Scale( domain=["Demand", "Dispatch", "Solar", "Shed"], range=["#e45756", "#4c78a8", "#f2ae72", "#b279a2"])), tooltip=["Hour", "Series", "MW"], ).properties(title="Hourly power [MW]", height=260) c2 = alt.Chart(df).mark_bar(color="#54a24b").encode( x=alt.X("Hour:O"), y=alt.Y("Active:Q"), tooltip=["Hour", "Active"], ).properties(title="Committed modules", height=200) return c1, c2 tab_exp, tab_ins, tab_cmp, tab_scale, tab_rep = st.tabs( ["Experiment", "Inspect", "Compare", "Scale", "Reproduce"] ) # ---- Experiment ------------------------------------------------------------ with tab_exp: st.subheader("Experiment") _apply_controls() if st.session_state.milp_energy_busy: st.warning("Analysis is already running. Please wait.") if st.button("Run analysis", type="primary", width="stretch", disabled=st.session_state.milp_energy_busy): st.session_state.milp_energy_busy = True try: settings = validate_settings(_collect_settings()) with st.spinner("Solving with HiGHS in isolated child..."): code = ( "import json,sys;" "from energy_core import solve_scenario;" "print(json.dumps(solve_scenario(json.loads(sys.argv[1])), allow_nan=False))" ) result = _run_supervised(code, [json.dumps(settings)]) if result.get("status") == "error": st.error(f"Solver error: {result.get('error')}") else: st.session_state.milp_energy_result = result st.session_state.milp_energy_settings = settings except Exception as exc: # noqa: BLE001 st.error(f"Invalid settings: {exc}") finally: st.session_state.milp_energy_busy = False res = st.session_state.milp_energy_result if res: st.markdown("*Results from the inputs at the time of the last run.*") st.json(res["settings"]) m1, m2, m3, m4 = st.columns(4) m1.metric("Status", res["status"]) m2.metric("Objective", res["objective"] if res["objective"] is not None else "n/a") m3.metric("Modules", res["modules"] if res["modules"] is not None else "n/a") m4.metric("Capacity MW", res["capacity_mw"] if res["capacity_mw"] is not None else "n/a") if res["status"] in ("optimal", "feasible"): c1, c2 = _charts(res) st.altair_chart(c1, width="stretch") st.altair_chart(c2, width="stretch") st.subheader("Cost breakdown") st.bar_chart(pd.DataFrame([res["costs"]]).T.rename(columns={0: "Cost"})) elif res["status"] == "infeasible": st.warning("Infeasible: no incumbent solution. Enable shedding or raise max modules.") else: st.error(res.get("error", "Solver error")) # ---- Inspect --------------------------------------------------------------- with tab_ins: st.subheader("Inspect") res = st.session_state.milp_energy_result if not res: st.info("Run an analysis first.") else: st.markdown("**Variables / constraints (units):** n_mod (integer modules), " "status[t] (committed modules), p[t] (MW), start_up/shut_down " "(modules), balance p+solar+shed=demand.") st.json(res["solver"]) st.markdown("**Residuals (independent check):**") st.json(res["residuals"]) if res["status"] == "optimal": st.success("Optimal solution verified against reconstructed objective.") elif res["status"] == "feasible": st.warning("Feasible but not proven optimal (gap may be unavailable).") elif res["status"] == "infeasible": st.error("Infeasible: physical shortage without shedding.") else: st.error("Solver error.") st.dataframe(_df(res), width="stretch") # ---- Compare --------------------------------------------------------------- with tab_cmp: st.subheader("Compare") res = st.session_state.milp_energy_result name = st.text_input("Scenario name", key="milp_energy_name", max_chars=40) if st.button("Save scenario", width="stretch"): if res and res["status"] in ("optimal", "feasible"): saved = st.session_state.milp_energy_saved if len(saved) >= 10: st.warning("Bounded to 10 saved scenarios.") else: saved.append({ "name": name or f"scenario-{len(saved)+1}", "settings": res["settings"], "objective": res["objective"], "modules": res["modules"], "capacity_mw": res["capacity_mw"], }) else: st.info("Run a feasible analysis first.") if st.session_state.milp_energy_saved: st.dataframe(pd.DataFrame(st.session_state.milp_energy_saved), width="stretch") # ---- Scale ----------------------------------------------------------------- with tab_scale: st.subheader("Scale") env = cpu_limits() st.json(env) if env["effective_cpus"] <= 1: st.info("Scaling unavailable: only one effective CPU detected.") n_cases = st.slider("Cases", 4, 12, 4, step=4, key="milp_energy_cases") n_workers = st.slider("Workers", 1, env["max_workers"], min(2, env["max_workers"]), key="milp_energy_workers") if st.session_state.milp_energy_busy: st.warning("Analysis is already running. Please wait.") if st.button("Run scaling experiment", width="stretch", disabled=st.session_state.milp_energy_busy): st.session_state.milp_energy_busy = True try: batch = make_batch(st.session_state.milp_energy_settings, n_cases) with st.spinner("Running AGILAB benchmark in isolated child..."): code = ( "import json,sys;" "from energy_runner import run_benchmark;" "batch=json.loads(sys.argv[1]);" "workers=int(sys.argv[2]);" "print(json.dumps(run_benchmark(batch, workers), allow_nan=False))" ) bench = _run_supervised(code, [json.dumps(batch), str(n_workers)]) if bench.get("status") == "error": st.error(f"Benchmark error: {bench.get('error')}") else: st.session_state.milp_energy_benchmark = bench except Exception as exc: # noqa: BLE001 st.error(f"Benchmark failed: {exc}") finally: st.session_state.milp_energy_busy = False bench = st.session_state.milp_energy_benchmark if bench: st.markdown("**One worker vs N workers (same batch):**") st.json(bench["comparison"]) rows = bench["parallel"]["rows"] st.dataframe(pd.DataFrame([ {"case": r["case"], "pid": r["pid"], "status": r["status"], "objective": r["objective"], "dur_s": round(r["end_monotonic"] - r["start_monotonic"], 3)} for r in rows ]), width="stretch") st.caption("No guaranteed speedup; public measurements may be slower in parallel.") # ---- Reproduce ------------------------------------------------------------- with tab_rep: st.subheader("Reproduce") res = st.session_state.milp_energy_settings st.markdown( "Source notebook: [PyPSA modular-committable tutorial](%s) — " "PyPSA contributors, CC BY 4.0. Library PyPSA is MIT; docs notebooks " "are CC BY 4.0. Derived portions: interactive lab, HiGHS solver, " "parameters, solver checks, AGILAB batch measurement." % SOURCE_URL ) st.download_button( "Download settings JSON", json.dumps(res, indent=2), "settings.json", "application/json", width="stretch", ) if st.session_state.milp_energy_result: st.download_button( "Download results JSON", json.dumps(st.session_state.milp_energy_result, indent=2, allow_nan=False), "results.json", "application/json", width="stretch", ) st.download_button( "Download model artifact (text)", model_text(res, st.session_state.milp_energy_result), "model.txt", "text/plain", width="stretch", ) if st.session_state.milp_energy_benchmark: st.download_button( "Download benchmark JSON", json.dumps(st.session_state.milp_energy_benchmark, indent=2, default=str), "benchmark.json", "application/json", width="stretch", )