Speed up /api/calculate from 75s to 17s on h4a's small VM
- ProcessPoolExecutor with fork context: 4 workers each inherit the scenario df via copy-on-write. linprog/HiGHS doesn't release the GIL so threads were useless; processes get clean ~3.5× speedup. - Cross-scenario LP cache keyed on everything the LP depends on (demand, retail, PV size, saldering, η_rt) but NOT inflation or cost. Result: tweaking just inflation re-uses every LP — sub-100ms. Cold cold scenario went 75s → 17s; inflation tweak went 17s → 10ms.
This commit is contained in:
parent
fa7a205eb7
commit
1878b0a383
1 changed files with 69 additions and 12 deletions
|
|
@ -13,7 +13,10 @@ Deploy (h4a):
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from concurrent.futures import ProcessPoolExecutor
|
||||
from multiprocessing import get_context
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
|
@ -51,6 +54,31 @@ print(f" {len(_BASE_DF)} hourly rows, {len(_CATALOG)} batteries in catalog.", f
|
|||
# scenario hash → result dict
|
||||
_RESULT_CACHE: dict[tuple, dict] = {}
|
||||
|
||||
# (df_signature, cap, power) → year-1 LP savings. Survives across scenarios so
|
||||
# tweaking only inflation (which does not affect the LP) is instant.
|
||||
_LP_CACHE: dict[tuple, float] = {}
|
||||
|
||||
# Holds the per-scenario df so worker processes inherit it via fork.
|
||||
_WORKER_DF = None
|
||||
_WORKER_ETA = 0.88
|
||||
|
||||
|
||||
def _worker_init(df, eta):
|
||||
"""Set the module-level globals each forked worker uses."""
|
||||
global _WORKER_DF, _WORKER_ETA
|
||||
_WORKER_DF = df
|
||||
_WORKER_ETA = eta
|
||||
|
||||
|
||||
def _worker_run_lp(spec):
|
||||
cap, pw = spec
|
||||
bat = Battery(
|
||||
capacity_kwh=cap, max_charge_kw=pw, max_discharge_kw=pw,
|
||||
round_trip_eff=_WORKER_ETA, allows_export=False,
|
||||
)
|
||||
out = simulate(_WORKER_DF, bat, oracle_daily_schedule(_WORKER_DF, bat))
|
||||
return spec, float(out["savings"].sum())
|
||||
|
||||
|
||||
# ─── Core compute ─────────────────────────────────────────────────────
|
||||
def _build_df(demand_kwh: float, retail: float, pv_kwp: float, pv_yield: float,
|
||||
|
|
@ -91,8 +119,11 @@ def compute_leaderboard(
|
|||
avg_epex = float(df["epex_eur_per_kwh"].mean())
|
||||
store_params = StoreParams(avg_epex_eur_per_kwh=avg_epex)
|
||||
|
||||
lp_cache: dict[tuple[float, float], float] = {}
|
||||
rows = []
|
||||
# Collect every distinct (capacity, power) the catalog asks for, then run
|
||||
# the per-spec LPs in parallel. linprog/HiGHS release the GIL so threads
|
||||
# give a near-linear speedup on this VM (4 cores).
|
||||
valid_rows = []
|
||||
specs: set[tuple[float, float]] = set()
|
||||
for b in _CATALOG:
|
||||
try:
|
||||
cap = float(b["capacity"])
|
||||
|
|
@ -102,18 +133,44 @@ def compute_leaderboard(
|
|||
continue
|
||||
if cap <= 0 or pw <= 0 or price <= 0:
|
||||
continue
|
||||
valid_rows.append((b, cap, pw, price))
|
||||
specs.add((round(cap, 3), round(pw, 3)))
|
||||
|
||||
lp_key = (round(cap, 3), round(pw, 3))
|
||||
if lp_key in lp_cache:
|
||||
lp_year1 = lp_cache[lp_key]
|
||||
# Cache key for "what does this LP depend on?" — everything that goes
|
||||
# into the dataframe construction plus eta. Inflation isn't in here, so
|
||||
# tweaking only inflation hits the cache for every catalog entry.
|
||||
df_sig = (round(demand_kwh, 1), round(retail, 4), round(pv_kwp, 2),
|
||||
round(pv_yield, 1), fixed_rate, saldering, round(eta, 3))
|
||||
|
||||
cached_results = {}
|
||||
todo = []
|
||||
for spec in specs:
|
||||
key = (df_sig, *spec)
|
||||
if key in _LP_CACHE:
|
||||
cached_results[spec] = _LP_CACHE[key]
|
||||
else:
|
||||
bat = Battery(
|
||||
capacity_kwh=cap, max_charge_kw=pw, max_discharge_kw=pw,
|
||||
round_trip_eff=eta, allows_export=False,
|
||||
)
|
||||
out = simulate(df, bat, oracle_daily_schedule(df, bat))
|
||||
lp_year1 = float(out["savings"].sum())
|
||||
lp_cache[lp_key] = lp_year1
|
||||
todo.append(spec)
|
||||
|
||||
if todo:
|
||||
workers = min(len(todo), int(os.environ.get("PLUGINBATTERY_LP_WORKERS", "4")))
|
||||
# Fork pool: each worker inherits `df` via copy-on-write rather than
|
||||
# pickling it on every task — fast cold scenarios.
|
||||
ctx = get_context("fork")
|
||||
with ProcessPoolExecutor(
|
||||
max_workers=max(1, workers),
|
||||
mp_context=ctx,
|
||||
initializer=_worker_init,
|
||||
initargs=(df, eta),
|
||||
) as ex:
|
||||
for spec, year1 in ex.map(_worker_run_lp, todo):
|
||||
cached_results[spec] = year1
|
||||
_LP_CACHE[(df_sig, *spec)] = year1
|
||||
|
||||
lp_cache = cached_results
|
||||
|
||||
rows = []
|
||||
for b, cap, pw, price in valid_rows:
|
||||
lp_year1 = lp_cache[(round(cap, 3), round(pw, 3))]
|
||||
lp_payback = payback_years(lp_year1, price, inflation)
|
||||
|
||||
scn = Scenario(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue