From 2ec958a7dd680e78673b882c8d6c5e5c61243bf8 Mon Sep 17 00:00:00 2001 From: Lars Valentin Date: Wed, 22 Jul 2026 08:02:53 +0200 Subject: [PATCH 1/4] feat: add charge_profile for SoC-dependent charge power limits Add CC-CV taper modeling to the optimizer. Batteries can now specify a charge_profile (e.g. lifepo4_conservative) that limits charge power as a function of SoC, matching real-world BMS behavior. The MILP solver uses iterative constraint tightening: solve, extract SoC trajectory, compute per-slot P_max(SoC) from the profile, add upper bound constraints, re-solve until convergence (max 5 iterations). New API fields in BatteryConfig: - charge_profile: named profile (e.g. lifepo4_conservative) - charge_knee: SoC% where taper begins (override) - charge_k: exponential decay constant (override) - charge_c_rate_float: minimum C-rate at float (override) New module: charging_profiles.py with profile definitions and max_charge_power() calculation. --- openapi.yaml | 33 +++++++ src/optimizer/app.py | 41 ++++++++- src/optimizer/charging_profiles.py | 133 +++++++++++++++++++++++++++++ src/optimizer/optimizer.py | 85 +++++++++++++++++- 4 files changed, 288 insertions(+), 4 deletions(-) create mode 100644 src/optimizer/charging_profiles.py diff --git a/openapi.yaml b/openapi.yaml index 7447c542e..3e416c342 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -269,6 +269,39 @@ components: maximum: 2 default: 0 description: Charging and discharging priority 0..2 compared to other batteries. 2 = highest priority. + charge_profile: + type: string + description: | + Named charge profile for SoC-dependent charge power limiting + (CC-CV taper). When set, the MILP solver uses iterative constraint + tightening to enforce realistic charge power limits at high SoC. + Available profiles: lifepo4_conservative, lifepo4_moderate. + Individual parameters can be overridden with charge_knee, + charge_k, charge_c_rate_float. + example: lifepo4_conservative + charge_knee: + type: number + minimum: 0 + maximum: 100 + description: | + SoC percentage where charge power tapering begins. + With charge_profile: overrides the profile default. + Without charge_profile: requires charge_k to build a custom profile. + example: 65 + charge_k: + type: number + minimum: 0 + description: | + Exponential decay constant for the CC-CV taper curve. + Higher values mean steeper power reduction above the knee. + example: 0.052 + charge_c_rate_float: + type: number + minimum: 0 + description: | + Minimum C-rate at the tail end of charging (float/balancing phase). + Default: 0.01 (C/100). + example: 0.01 TimeSeries: type: object diff --git a/src/optimizer/app.py b/src/optimizer/app.py index 0efdd660a..adbeca9d6 100644 --- a/src/optimizer/app.py +++ b/src/optimizer/app.py @@ -5,6 +5,7 @@ from flask_restx import Api, Resource, fields from werkzeug.exceptions import BadRequest +from .charging_profiles import get_profile, CHEMISTRY_PROFILES from .optimizer import BatteryConfig, GridConfig, OptimizationStrategy, Optimizer, TimeSeriesData app = Flask(__name__) @@ -80,7 +81,11 @@ def handle_validation_error(error): 'c_max': fields.Float(required=True, description='Maximum charge power (W)'), 'd_max': fields.Float(required=True, description='Maximum discharge power (W)'), 'p_a': fields.Float(required=True, description='Monetary value per Wh at end of the optimization horizon'), - 'c_priority': fields.Integer(required=False, description='Charging and discharging priority compared to other batteries. 2 = highest priority.') + 'c_priority': fields.Integer(required=False, description='Charging and discharging priority compared to other batteries. 2 = highest priority.'), + 'charge_profile': fields.String(required=False, description='Named charge profile for SoC-dependent charge power limiting (CC-CV taper). Enables iterative constraint tightening in the MILP solver. Individual parameters can be overridden with charge_knee, charge_k, charge_c_rate_float.'), + 'charge_knee': fields.Float(required=False, description='SoC (%) where CC-CV taper begins. With charge_profile: overrides the profile value. Without: requires charge_k to build a custom profile.'), + 'charge_k': fields.Float(required=False, description='Exponential decay constant for the CC-CV taper curve.'), + 'charge_c_rate_float': fields.Float(required=False, description='Minimum C-rate at float / tail end of charging (default 0.01).'), }) time_series_model = api.model('TimeSeries', { @@ -157,20 +162,50 @@ def post(self): # Parse battery configurations batteries = [] for bat_data in data['batteries']: + # Build optional CC-CV taper profile + profile = None + profile_name = bat_data.get('charge_profile') + charge_knee = bat_data.get('charge_knee') + charge_k = bat_data.get('charge_k') + charge_c_rate_float = bat_data.get('charge_c_rate_float') + + s_capacity = bat_data.get('s_capacity', bat_data['s_max']) + c_max = bat_data['c_max'] + + if profile_name: + # Named profile with optional overrides + overrides = {} + if charge_knee is not None: + overrides['knee'] = charge_knee + if charge_k is not None: + overrides['k'] = charge_k + if charge_c_rate_float is not None: + overrides['c_rate_float'] = charge_c_rate_float + profile = get_profile(profile_name, **overrides) + elif charge_knee is not None and charge_k is not None: + # Custom profile: derive c_rate_max from c_max / s_capacity + profile = { + 'c_rate_max': c_max / s_capacity if s_capacity > 0 else 0.25, + 'knee': charge_knee, + 'k': charge_k, + 'c_rate_float': charge_c_rate_float if charge_c_rate_float is not None else 0.01, + } + batteries.append(BatteryConfig( charge_from_grid=bat_data.get('charge_from_grid', False), discharge_to_grid=bat_data.get('discharge_to_grid', False), - s_capacity=bat_data.get('s_capacity', bat_data['s_max']), + s_capacity=s_capacity, s_min=bat_data['s_min'], s_max=bat_data['s_max'], s_initial=bat_data['s_initial'], p_demand=bat_data.get('p_demand'), s_goal=bat_data.get('s_goal'), c_min=bat_data['c_min'], - c_max=bat_data['c_max'], + c_max=c_max, d_max=bat_data['d_max'], p_a=bat_data['p_a'], c_priority=bat_data.get('c_priority', 0), + charge_profile=profile, )) # Parse time series data diff --git a/src/optimizer/charging_profiles.py b/src/optimizer/charging_profiles.py new file mode 100644 index 000000000..f303e7d84 --- /dev/null +++ b/src/optimizer/charging_profiles.py @@ -0,0 +1,133 @@ +""" +Battery charging profiles by cell chemistry. + +Each profile defines how the maximum charge power decreases as a function +of State of Charge (SoC). This models the real-world CC-CV charging behavior +where the Battery Management System (BMS) reduces the charge current at +higher SoC to protect cells and minimize resistive losses. + +Usage: + from charging_profiles import get_profile, max_charge_power + + profile = get_profile("lifepo4_conservative") + watts = max_charge_power(profile, capacity_wh=31000, soc_percent=75) + +To find values for your own system: + - Set c_rate_max to the highest C-rate your battery sustains + without the BMS reducing the charge current + - Set knee to the SoC percentage where the BMS starts + reducing the charge current below c_rate_max + - Leave k and c_rate_float at their defaults unless you have + your own measurements (fit with Wolfram Alpha: + "exponential fit {SoC1,I1}, {SoC2,I2}, ...") + +General advice: always stay slightly below your measured values. +If your BMS starts tapering at 66%, set knee to 65%. If your +inverter sustains 160A, set c_rate_max to match C/4 or lower. +This avoids triggering BMS intervention, reduces resistive losses, +and extends cell life. +""" + +import math + + +# --------------------------------------------------------------------------- +# Chemistry profiles +# --------------------------------------------------------------------------- + +CHEMISTRY_PROFILES = { + "lifepo4_conservative": { + # Charging profile for LiFePO4 cells optimized for longevity and + # minimal resistive losses. Derived from real-world measurements + # on a mixed environment of aged LiYFePO4 cells (200Ah + 90Ah) + # and newer LiFePO4 cells (314Ah) with Victron inverters. + # + # The charging curve has two phases: + # 1. Below 'knee': constant power at 'c_rate_max' rate + # 2. Above 'knee': power drops exponentially toward 'c_rate_float' + # + # Formula above knee: + # P = max(c_rate_float, c_rate_max * e^(-k * (SoC - knee))) + + "c_rate_max": 0.25, + # Maximum charge rate below knee, as fraction of capacity. + # C/4 means a 31 kWh battery charges at up to 7750 W. + + "knee": 65, + # SoC percentage where charge current begins to taper. + + "k": 0.052, + # Exponential decay constant (R^2 = 0.996 from BMS measurements). + + "c_rate_float": 0.01, + # Minimum charge rate (maintenance/balancing phase). + # C/100 means a 31 kWh battery trickle-charges at 310 W. + }, + + "lifepo4_moderate": { + # Moderate profile for newer LiFePO4 cells in good condition. + + "c_rate_max": 0.33, # C/3 + "knee": 75, + "k": 0.045, + "c_rate_float": 0.02, # C/50 + }, +} + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def list_profiles(): + """Return list of available profile names.""" + return list(CHEMISTRY_PROFILES.keys()) + + +def get_profile(name, **overrides): + """ + Get a charging profile by name with optional overrides. + + Args: + name: Profile name (e.g. "lifepo4_conservative") + **overrides: Override individual values (e.g. knee=70, c_rate_max=0.2) + + Returns: + dict with keys: c_rate_max, knee, k, c_rate_float + """ + if name not in CHEMISTRY_PROFILES: + available = ", ".join(CHEMISTRY_PROFILES.keys()) + raise ValueError(f"Unknown profile '{name}'. Available: {available}") + + profile = dict(CHEMISTRY_PROFILES[name]) + for key, value in overrides.items(): + if key not in profile: + raise ValueError(f"Unknown parameter '{key}'. Valid: {', '.join(profile.keys())}") + profile[key] = value + + return profile + + +def max_charge_power(profile, capacity_wh, soc_percent): + """ + Calculate maximum charge power in watts for a given SoC. + + Args: + profile: Charging profile dict from get_profile() + capacity_wh: Battery capacity in Wh + soc_percent: Current state of charge (0-100) + + Returns: + Maximum charge power in watts + """ + c_rate_max = profile["c_rate_max"] + knee = profile["knee"] + k = profile["k"] + c_rate_float = profile["c_rate_float"] + + if soc_percent <= knee: + c_rate = c_rate_max + else: + c_rate = max(c_rate_float, c_rate_max * math.exp(-k * (soc_percent - knee))) + + return capacity_wh * c_rate diff --git a/src/optimizer/optimizer.py b/src/optimizer/optimizer.py index b878ec8e6..029706e9d 100644 --- a/src/optimizer/optimizer.py +++ b/src/optimizer/optimizer.py @@ -5,6 +5,7 @@ import numpy as np import pulp +from .charging_profiles import max_charge_power from .settings import OptimizerSettings @@ -36,6 +37,7 @@ class BatteryConfig: p_demand: Optional[List[float]] = None # Minimum charge demand (Wh) s_goal: Optional[List[float]] = None # Goal state of charge (Wh) c_priority: int = 0 + charge_profile: Optional[dict] = None # CC-CV taper profile (charging_profiles format) @dataclass @@ -491,10 +493,75 @@ def _add_battery_constraints(self): # Charge constraint self.problem += self.variables['c'][i][t] <= self.M * (1 - self.variables['z_cd'][i][t]) + def _has_charge_profiles(self) -> bool: + """True if any battery has a CC-CV taper profile.""" + return any(bat.charge_profile is not None for bat in self.batteries) + + def _soc_percent(self, bat_index: int, soc_wh: float) -> float: + """Convert SoC from Wh to percent for a battery.""" + cap = self.batteries[bat_index].s_capacity + if cap <= 0: + return 0.0 + return (soc_wh / cap) * 100.0 + + def _add_taper_constraints(self, iteration: int) -> int: + """Read SoC trajectory from current solution and inject per-slot + charge power upper bounds from the taper profile. + + Returns the number of binding constraints added. + """ + n_added = 0 + for i, bat in enumerate(self.batteries): + if bat.charge_profile is None: + continue + for t in self.time_steps: + soc_wh = pulp.value(self.variables['s'][i][t]) + if soc_wh is None: + continue + soc_pct = self._soc_percent(i, soc_wh) + p_limit = max_charge_power(bat.charge_profile, bat.s_capacity, soc_pct) + e_limit = p_limit * self.time_series.dt[t] / 3600.0 + e_full = bat.c_max * self.time_series.dt[t] / 3600.0 + if e_limit < e_full - 1.0: # 1 Wh tolerance + self.problem += ( + self.variables['c'][i][t] <= e_limit, + f"taper_{i}_{t}_iter{iteration}", + ) + n_added += 1 + return n_added + + def _soc_trajectory(self) -> Dict[int, List[float]]: + """Extract SoC trajectory (Wh) per battery from current solution.""" + result = {} + for i in range(len(self.batteries)): + result[i] = [ + pulp.value(self.variables['s'][i][t]) + for t in self.time_steps + ] + return result + + @staticmethod + def _trajectory_converged(prev, curr, tol_wh=50.0): + """Check whether SoC trajectories converged between iterations.""" + for i in prev: + for t in range(len(prev[i])): + p = prev[i][t] + c = curr[i][t] + if p is None or c is None: + continue + if abs(p - c) > tol_wh: + return False + return True + def solve(self) -> Dict: """ Creates the MILP model if none exists and solves the optimization problem. - Returns a dictionary with the optimization results + + When batteries have a charge_profile (CC-CV taper), iterative constraint + tightening is used: solve, extract SoC trajectory, compute per-slot + P_max(SoC), add constraints, re-solve until convergence. + + Returns a dictionary with the optimization results. """ if self.problem is None: @@ -506,10 +573,26 @@ def solve(self) -> Dict: threads=self.settings.num_threads, timeLimit=self.settings.time_limit, ) + max_taper_iterations = 5 with TemporaryDirectory() as tmpdir: solver.tmpDir = tmpdir self.problem.solve(solver) + # Iterative tightening for SoC-dependent charge limits + if self._has_charge_profiles() and pulp.LpStatus[self.problem.status] == 'Optimal': + prev_trajectory = self._soc_trajectory() + for iteration in range(1, max_taper_iterations + 1): + n_added = self._add_taper_constraints(iteration) + if n_added == 0: + break + self.problem.solve(solver) + if pulp.LpStatus[self.problem.status] != 'Optimal': + break + curr_trajectory = self._soc_trajectory() + if self._trajectory_converged(prev_trajectory, curr_trajectory): + break + prev_trajectory = curr_trajectory + # Extract results status = pulp.LpStatus[self.problem.status] From 6fff2b2180e16e066a1ef4ed7f37588d52b81dbf Mon Sep 17 00:00:00 2001 From: Lars Valentin Date: Wed, 22 Jul 2026 08:14:29 +0200 Subject: [PATCH 2/4] fix: shorten description strings to pass lint (E501, I001) --- src/optimizer/app.py | 6 +++--- src/optimizer/charging_profiles.py | 1 - 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/src/optimizer/app.py b/src/optimizer/app.py index adbeca9d6..16180b513 100644 --- a/src/optimizer/app.py +++ b/src/optimizer/app.py @@ -5,7 +5,7 @@ from flask_restx import Api, Resource, fields from werkzeug.exceptions import BadRequest -from .charging_profiles import get_profile, CHEMISTRY_PROFILES +from .charging_profiles import get_profile from .optimizer import BatteryConfig, GridConfig, OptimizationStrategy, Optimizer, TimeSeriesData app = Flask(__name__) @@ -82,8 +82,8 @@ def handle_validation_error(error): 'd_max': fields.Float(required=True, description='Maximum discharge power (W)'), 'p_a': fields.Float(required=True, description='Monetary value per Wh at end of the optimization horizon'), 'c_priority': fields.Integer(required=False, description='Charging and discharging priority compared to other batteries. 2 = highest priority.'), - 'charge_profile': fields.String(required=False, description='Named charge profile for SoC-dependent charge power limiting (CC-CV taper). Enables iterative constraint tightening in the MILP solver. Individual parameters can be overridden with charge_knee, charge_k, charge_c_rate_float.'), - 'charge_knee': fields.Float(required=False, description='SoC (%) where CC-CV taper begins. With charge_profile: overrides the profile value. Without: requires charge_k to build a custom profile.'), + 'charge_profile': fields.String(required=False, description='Named CC-CV taper profile (e.g. lifepo4_conservative).'), + 'charge_knee': fields.Float(required=False, description='SoC% where taper begins. Overrides profile default.'), 'charge_k': fields.Float(required=False, description='Exponential decay constant for the CC-CV taper curve.'), 'charge_c_rate_float': fields.Float(required=False, description='Minimum C-rate at float / tail end of charging (default 0.01).'), }) diff --git a/src/optimizer/charging_profiles.py b/src/optimizer/charging_profiles.py index f303e7d84..960ea4c6b 100644 --- a/src/optimizer/charging_profiles.py +++ b/src/optimizer/charging_profiles.py @@ -30,7 +30,6 @@ import math - # --------------------------------------------------------------------------- # Chemistry profiles # --------------------------------------------------------------------------- From 18c52db57524b64d12927d6acfce43febc5e6f25 Mon Sep 17 00:00:00 2001 From: Lars Valentin Date: Wed, 22 Jul 2026 14:17:39 +0200 Subject: [PATCH 3/4] refactor: replace iterative taper loop with static tangent constraints Replace the iterative constraint tightening (3-6 re-solves) with piecewise-linear upper bounds added at model build time. The CC-CV taper curve is convex, so tangent lines at 4 breakpoints (knee to 100% SoC) provide valid linear constraints on charge power as a function of s[i][t-1]. One solve, no stale bounds, bounded wall clock. --- src/optimizer/optimizer.py | 135 +++++++++++++++++-------------------- 1 file changed, 63 insertions(+), 72 deletions(-) diff --git a/src/optimizer/optimizer.py b/src/optimizer/optimizer.py index 029706e9d..f8007baf0 100644 --- a/src/optimizer/optimizer.py +++ b/src/optimizer/optimizer.py @@ -1,3 +1,4 @@ +import math from dataclasses import dataclass from tempfile import TemporaryDirectory from typing import Dict, List, Optional @@ -117,6 +118,7 @@ def create_model(self): self._setup_target_function() self._add_energy_balance_constraints() self._add_battery_constraints() + self._add_charge_profile_constraints() def _setup_variables(self): """ @@ -493,73 +495,78 @@ def _add_battery_constraints(self): # Charge constraint self.problem += self.variables['c'][i][t] <= self.M * (1 - self.variables['z_cd'][i][t]) - def _has_charge_profiles(self) -> bool: - """True if any battery has a CC-CV taper profile.""" - return any(bat.charge_profile is not None for bat in self.batteries) + def _add_charge_profile_constraints(self): + """Add piecewise-linear upper bounds on charge power from CC-CV taper profiles. - def _soc_percent(self, bat_index: int, soc_wh: float) -> float: - """Convert SoC from Wh to percent for a battery.""" - cap = self.batteries[bat_index].s_capacity - if cap <= 0: - return 0.0 - return (soc_wh / cap) * 100.0 + For each battery with a charge_profile, the taper curve P_max(SoC) is + a convex decreasing function above the knee. We approximate it with + tangent lines at 4 breakpoints. Each tangent is a valid linear upper + bound (convexity guarantees the tangent never exceeds the curve), so + the MILP enforces realistic charge power limits in a single solve. - def _add_taper_constraints(self, iteration: int) -> int: - """Read SoC trajectory from current solution and inject per-slot - charge power upper bounds from the taper profile. - - Returns the number of binding constraints added. + The constraint for slot t uses s[i][t-1] (SoC at slot start). For + t=0, s_initial is a constant so the bound is applied directly to + the variable's upper bound. """ - n_added = 0 for i, bat in enumerate(self.batteries): if bat.charge_profile is None: continue - for t in self.time_steps: - soc_wh = pulp.value(self.variables['s'][i][t]) - if soc_wh is None: - continue - soc_pct = self._soc_percent(i, soc_wh) - p_limit = max_charge_power(bat.charge_profile, bat.s_capacity, soc_pct) - e_limit = p_limit * self.time_series.dt[t] / 3600.0 - e_full = bat.c_max * self.time_series.dt[t] / 3600.0 - if e_limit < e_full - 1.0: # 1 Wh tolerance - self.problem += ( - self.variables['c'][i][t] <= e_limit, - f"taper_{i}_{t}_iter{iteration}", - ) - n_added += 1 - return n_added - - def _soc_trajectory(self) -> Dict[int, List[float]]: - """Extract SoC trajectory (Wh) per battery from current solution.""" - result = {} - for i in range(len(self.batteries)): - result[i] = [ - pulp.value(self.variables['s'][i][t]) - for t in self.time_steps - ] - return result - - @staticmethod - def _trajectory_converged(prev, curr, tol_wh=50.0): - """Check whether SoC trajectories converged between iterations.""" - for i in prev: - for t in range(len(prev[i])): - p = prev[i][t] - c = curr[i][t] - if p is None or c is None: - continue - if abs(p - c) > tol_wh: - return False - return True + + profile = bat.charge_profile + cap = bat.s_capacity + if cap <= 0: + continue + + knee = profile['knee'] + k = profile['k'] + c_rate_max = profile['c_rate_max'] + + # 4 breakpoints from knee to 100% SoC + n_points = 4 + breakpoints = [] + for j in range(n_points): + soc_pct = knee + (100.0 - knee) * j / (n_points - 1) + s_wh = soc_pct / 100.0 * cap + p_w = max_charge_power(profile, cap, soc_pct) + # derivative of P_max w.r.t. SoC in Wh: + # dP/ds = dP/d(soc%) * d(soc%)/ds + # = (-k * c_rate_max * cap * e^(-k*(soc%-knee))) * (100/cap) + # = -k * c_rate_max * 100 * e^(-k*(soc%-knee)) + dp_ds = -k * c_rate_max * 100.0 * math.exp(-k * (soc_pct - knee)) + breakpoints.append((s_wh, p_w, dp_ds)) + + for s_b, p_b, dp_ds in breakpoints: + for t in self.time_steps: + dt_h = self.time_series.dt[t] / 3600.0 + + if t == 0: + # s_initial is a constant: compute a fixed upper bound + p_limit = p_b + dp_ds * (bat.s_initial - s_b) + e_limit = p_limit * dt_h + e_var_ub = bat.c_max * dt_h + # only add if tighter than the existing variable bound + if e_limit < e_var_ub - 1.0: + self.problem += ( + self.variables['c'][i][t] <= e_limit, + f"taper_{i}_{t}_bp{breakpoints.index((s_b, p_b, dp_ds))}", + ) + else: + # linear constraint: c[i][t] <= (p_b - dp_ds*s_b)*dt_h + dp_ds*dt_h * s[i][t-1] + s_prev = self.variables['s'][i][t - 1] + intercept = (p_b - dp_ds * s_b) * dt_h + slope = dp_ds * dt_h + self.problem += ( + self.variables['c'][i][t] <= intercept + slope * s_prev, + f"taper_{i}_{t}_bp{breakpoints.index((s_b, p_b, dp_ds))}", + ) def solve(self) -> Dict: """ Creates the MILP model if none exists and solves the optimization problem. - When batteries have a charge_profile (CC-CV taper), iterative constraint - tightening is used: solve, extract SoC trajectory, compute per-slot - P_max(SoC), add constraints, re-solve until convergence. + When batteries have a charge_profile (CC-CV taper), piecewise-linear + upper bounds on charge power are part of the model, so a single solve + enforces realistic SoC-dependent charge limits. Returns a dictionary with the optimization results. """ @@ -573,26 +580,10 @@ def solve(self) -> Dict: threads=self.settings.num_threads, timeLimit=self.settings.time_limit, ) - max_taper_iterations = 5 with TemporaryDirectory() as tmpdir: solver.tmpDir = tmpdir self.problem.solve(solver) - # Iterative tightening for SoC-dependent charge limits - if self._has_charge_profiles() and pulp.LpStatus[self.problem.status] == 'Optimal': - prev_trajectory = self._soc_trajectory() - for iteration in range(1, max_taper_iterations + 1): - n_added = self._add_taper_constraints(iteration) - if n_added == 0: - break - self.problem.solve(solver) - if pulp.LpStatus[self.problem.status] != 'Optimal': - break - curr_trajectory = self._soc_trajectory() - if self._trajectory_converged(prev_trajectory, curr_trajectory): - break - prev_trajectory = curr_trajectory - # Extract results status = pulp.LpStatus[self.problem.status] From ac5f77e7357d8aa9755af7a1eebe02d9b525c151 Mon Sep 17 00:00:00 2001 From: Lars Valentin Date: Wed, 22 Jul 2026 08:22:46 +0200 Subject: [PATCH 4/4] feat: add /optimize/level-schedule endpoint for solar-first strategy Add a new optimization endpoint that finds the optimal flat export ceiling for charging a battery from solar surplus. Instead of minimizing cost (MILP), this maximizes self-use by flattening the grid export curve. The algorithm finds the export level where all active charging slots reduce the forecasted feed-in power by the same amount. Charge power per slot is variable and respects the CC-CV taper profile. The response uses the same format as /optimize/charge-schedule (batteries[].charging_power/discharging_power/state_of_charge), so consumers can process both endpoints with identical code. New endpoint: POST /optimize/level-schedule New function: find_optimal_level() in charging_profiles.py New function: _level_to_milp_format() in app.py --- openapi.yaml | 109 +++++++++++++++++++++ src/optimizer/app.py | 151 ++++++++++++++++++++++++++++- src/optimizer/charging_profiles.py | 141 +++++++++++++++++++++++++++ 3 files changed, 400 insertions(+), 1 deletion(-) diff --git a/openapi.yaml b/openapi.yaml index 3e416c342..735055eca 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -125,6 +125,53 @@ paths: example: message: "Optimization failed: Infeasible problem" + /optimize/level-schedule: + post: + tags: + - optimization + summary: Optimize battery charging using the Level algorithm + description: | + Finds the optimal flat export ceiling that charges a battery from + soc_start to soc_target using solar surplus, while minimizing peak + grid export. Respects CC-CV taper profiles for SoC-dependent charge + power limits. + + Returns the same response format as /optimize/charge-schedule so + consumers can process both endpoints with identical code. + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/LevelInput" + example: + charge_profile: lifepo4_conservative + capacity_wh: 42500 + soc_start: 60 + soc_target: 95 + solar_w: [0, 0, 200, 500, 1000, 1800, 2800, 4000, 5500, 7000, 8500, 9500, 10000, 10200, 10000, 9500, 8500, 7000, 5500, 4000, 2800, 1800, 1000, 500] + consumption_w: 800 + dt_seconds: 900 + responses: + "200": + description: Optimization completed successfully + content: + application/json: + schema: + $ref: "#/components/schemas/OptimizationResult" + "400": + description: Bad request - Invalid input data + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "500": + description: Internal server error - Optimization failed + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + /optimize/health: get: tags: @@ -348,6 +395,68 @@ components: description: Grid export remuneration per Wh at each time step (currency units/Wh) example: [0.15, 0.12, 0.10, 0.11, 0.14, 0.16] + LevelInput: + type: object + required: + - capacity_wh + - soc_start + - soc_target + - solar_w + properties: + charge_profile: + type: string + description: Named CC-CV taper profile (e.g. lifepo4_conservative). + example: lifepo4_conservative + charge_knee: + type: number + minimum: 0 + maximum: 100 + description: SoC% where taper begins. Overrides profile default. + charge_k: + type: number + minimum: 0 + description: Exponential decay constant for the taper curve. + charge_c_rate_float: + type: number + minimum: 0 + description: Minimum C-rate at float (default 0.01). + capacity_wh: + type: number + minimum: 0 + description: Battery capacity in Wh + example: 42500 + soc_start: + type: number + minimum: 0 + maximum: 100 + description: Starting state of charge (0-100) + example: 60 + soc_target: + type: number + minimum: 0 + maximum: 100 + description: Target state of charge (0-100) + example: 95 + solar_w: + type: array + items: + type: number + minimum: 0 + description: Solar power forecast per time slot (watts) + example: [0, 500, 1000, 2000, 4000, 6000, 8000, 6000, 4000, 2000] + consumption_w: + type: number + minimum: 0 + default: 800 + description: Household consumption in watts + example: 800 + dt_seconds: + type: number + minimum: 0 + default: 900 + description: Slot duration in seconds + example: 900 + OptimizationInput: type: object required: diff --git a/src/optimizer/app.py b/src/optimizer/app.py index 16180b513..05856aa12 100644 --- a/src/optimizer/app.py +++ b/src/optimizer/app.py @@ -5,7 +5,7 @@ from flask_restx import Api, Resource, fields from werkzeug.exceptions import BadRequest -from .charging_profiles import get_profile +from .charging_profiles import find_optimal_level, get_profile from .optimizer import BatteryConfig, GridConfig, OptimizationStrategy, Optimizer, TimeSeriesData app = Flask(__name__) @@ -130,6 +130,54 @@ def handle_validation_error(error): }) +def _level_to_milp_format(level_result, capacity_wh, dt_seconds): + """Convert level algorithm output to MILP-compatible response format. + + Maps the level schedule (per-slot watts + SoC%) into the same structure + the MILP endpoint returns (per-slot Wh + SoC in Wh absolute), so + consumers can process both endpoints with identical code. + """ + schedule = level_result.get('schedule', []) + dt_hours = dt_seconds / 3600.0 + + charging_power = [] + discharging_power = [] + state_of_charge = [] + grid_export = [] + grid_import = [] + flow_direction = [] + + for slot in schedule: + charge_wh = slot['charge_w'] * dt_hours + export_wh = slot['export_w'] * dt_hours + + charging_power.append(round(charge_wh, 2)) + discharging_power.append(0.0) + state_of_charge.append(round(slot['soc'] / 100.0 * capacity_wh, 1)) + grid_export.append(round(export_wh, 2)) + grid_import.append(0.0) + flow_direction.append(1 if export_wh > 0 else 0) + + return { + 'status': level_result.get('status', 'Optimal'), + 'objective_value': 0.0, + 'limit_violations': { + 'grid_import_limit_exceeded': False, + 'grid_export_limit_hit': False, + }, + 'batteries': [{ + 'charging_power': charging_power, + 'discharging_power': discharging_power, + 'state_of_charge': state_of_charge, + }], + 'grid_import': grid_import, + 'grid_export': grid_export, + 'flow_direction': flow_direction, + 'grid_import_overshoot': [], + 'grid_export_overshoot': [], + } + + @ns.route('/charge-schedule') class OptimizeCharging(Resource): @api.expect(optimization_input_model, validate=True) @@ -256,6 +304,107 @@ def post(self): api.abort(500, f"Optimization failed: {str(e)}") +# Input model for level schedule +level_input_model = api.model('LevelInput', { + 'charge_profile': fields.String(required=False, description='Named CC-CV taper profile.'), + 'charge_knee': fields.Float(required=False, description='SoC% where taper begins.'), + 'charge_k': fields.Float(required=False, description='Exponential decay constant.'), + 'charge_c_rate_float': fields.Float(required=False, description='Minimum C-rate at float.'), + 'capacity_wh': fields.Float(required=True, description='Battery capacity in Wh'), + 'soc_start': fields.Float(required=True, description='Starting SoC (0-100)'), + 'soc_target': fields.Float(required=True, description='Target SoC (0-100)'), + 'solar_w': fields.List(fields.Float, required=True, description='Solar power per slot (W)'), + 'consumption_w': fields.Float(required=False, default=800, description='Consumption (W)'), + 'dt_seconds': fields.Float(required=False, default=900, description='Slot duration (s)'), +}) + + +@ns.route('/level-schedule') +class OptimizeLevel(Resource): + @api.expect(level_input_model, validate=True) + @api.marshal_with(optimization_result_model) + def post(self): + """ + Compute an optimal charge schedule using the Level algorithm. + + Finds the flat export ceiling that charges the battery from soc_start + to soc_target using solar surplus, while minimizing peak grid export. + Respects the CC-CV taper profile for SoC-dependent charge power limits. + + Returns the same format as /optimize/charge-schedule so consumers can + process both endpoints with identical code. + """ + try: + data = api.payload + + # Build profile + profile = None + profile_name = data.get('charge_profile') + charge_knee = data.get('charge_knee') + charge_k = data.get('charge_k') + charge_c_rate_float = data.get('charge_c_rate_float') + capacity_wh = data['capacity_wh'] + + if profile_name: + overrides = {} + if charge_knee is not None: + overrides['knee'] = charge_knee + if charge_k is not None: + overrides['k'] = charge_k + if charge_c_rate_float is not None: + overrides['c_rate_float'] = charge_c_rate_float + profile = get_profile(profile_name, **overrides) + elif charge_knee is not None and charge_k is not None: + profile = { + 'c_rate_max': 0.25, + 'knee': charge_knee, + 'k': charge_k, + 'c_rate_float': charge_c_rate_float or 0.01, + } + else: + profile = get_profile('lifepo4_conservative') + + dt_seconds = data.get('dt_seconds', 900) + + result = find_optimal_level( + profile=profile, + capacity_wh=capacity_wh, + soc_start=data['soc_start'], + soc_target=data['soc_target'], + solar_w=data['solar_w'], + consumption_w=data.get('consumption_w', 800), + dt_seconds=dt_seconds, + ) + + if result is None: + return { + 'status': 'Optimal', + 'objective_value': 0.0, + 'limit_violations': { + 'grid_import_limit_exceeded': False, + 'grid_export_limit_hit': False, + }, + 'batteries': [{ + 'charging_power': [], + 'discharging_power': [], + 'state_of_charge': [], + }], + 'grid_import': [], + 'grid_export': [], + 'flow_direction': [], + 'grid_import_overshoot': [], + 'grid_export_overshoot': [], + } + + result['status'] = 'Optimal' + return _level_to_milp_format(result, capacity_wh, dt_seconds) + + except ValueError as e: + api.abort(400, str(e)) + except Exception as e: + api.abort(500, f"Level optimization failed: {str(e)}") + + @ns.route('/health') class Health(Resource): def get(self): diff --git a/src/optimizer/charging_profiles.py b/src/optimizer/charging_profiles.py index 960ea4c6b..4cda63fb0 100644 --- a/src/optimizer/charging_profiles.py +++ b/src/optimizer/charging_profiles.py @@ -130,3 +130,144 @@ def max_charge_power(profile, capacity_wh, soc_percent): c_rate = max(c_rate_float, c_rate_max * math.exp(-k * (soc_percent - knee))) return capacity_wh * c_rate + + +# --------------------------------------------------------------------------- +# Level algorithm +# --------------------------------------------------------------------------- + + +def find_optimal_level( + profile, + capacity_wh, + soc_start, + soc_target, + solar_w, + consumption_w=800, + dt_seconds=900, +): + """ + Find the optimal flat export level that charges the battery to + target SoC while minimizing the peak grid export. + + Instead of sliding a fixed charge block, this algorithm finds the + horizontal line (export ceiling) such that charging with the surplus + above this line fills the battery exactly to the target SoC. + + In each slot, the charge power is: + min(PV - consumption - level, profile_max(SoC)) + + This allows variable charge power per slot: slow charging at the + edges of the solar bell curve, faster in the middle -- but always + limited by the SoC-dependent profile. + + The result is a flatter export curve compared to the fixed-block + approach, because the battery absorbs more during peak hours + (when SoC is still low) instead of ramping up early. + + Args: + profile: Charging profile dict (c_rate_max, knee, k, c_rate_float) + capacity_wh: Battery capacity in Wh + soc_start: Starting SoC (0-100) + soc_target: Target SoC (0-100) + solar_w: List of solar power values per slot (watts) + consumption_w: Household consumption in watts (default: 800) + dt_seconds: Slot duration in seconds (default: 900) + + Returns: + dict with schedule and metadata, or None if target unreachable. + Keys: n_slots, peak_export_w, peak_without_charge_w, + total_charged_kwh, export_level_w, schedule + schedule entries: {solar_w, charge_w, export_w, soc} + """ + if soc_start >= soc_target: + return None + + n_solar = len(solar_w) + if n_solar == 0: + return None + + dt_h = dt_seconds / 3600.0 + energy_needed_wh = capacity_wh * (soc_target - soc_start) / 100 + + # Peak export without any charging + peak_without = max(max(0, s - consumption_w) for s in solar_w) + + # Find the target peak that minimizes the actual peak export. + # The relationship between target peak and actual peak is not + # monotonic: too aggressive (low target) fills the battery early, + # leaving peak solar slots unprotected. Too conservative (high + # target) doesn't charge enough. We sweep through candidates + # and simulate each one to find the true minimum. + n_steps = 200 + best_target = peak_without + best_actual_peak = peak_without + + for step in range(n_steps + 1): + candidate = peak_without * step / n_steps + + # Simulate charging with this target + soc = soc_start + charged_wh = 0 + actual_peak = 0 + for i in range(n_solar): + surplus = max(0, solar_w[i] - consumption_w) + available = max(0, surplus - candidate) + p_max = max_charge_power(profile, capacity_wh, soc) + charge_w = min(available, p_max) + energy_wh = charge_w * dt_h + charged_wh += energy_wh + soc += energy_wh / capacity_wh * 100 + soc = min(soc, soc_target) + export_w = max(0, surplus - charge_w) + actual_peak = max(actual_peak, export_w) + + # Only consider if enough energy is charged + if charged_wh >= energy_needed_wh * 0.99: + if actual_peak < best_actual_peak: + best_actual_peak = actual_peak + best_target = candidate + + target_peak = best_target + + # Final simulation at the found target peak + soc = soc_start + total_charged_wh = 0 + schedule = [] + peak_export = 0 + + for i in range(n_solar): + surplus = max(0, solar_w[i] - consumption_w) + available = max(0, surplus - target_peak) + p_max = max_charge_power(profile, capacity_wh, soc) + charge_w = min(available, p_max) + remaining_wh = capacity_wh * (soc_target - soc) / 100 + if charge_w * dt_h > remaining_wh: + charge_w = remaining_wh / dt_h + energy_wh = charge_w * dt_h + total_charged_wh += energy_wh + soc += energy_wh / capacity_wh * 100 + soc = min(soc, soc_target) + + export_w = max(0, surplus - charge_w) + peak_export = max(peak_export, export_w) + + schedule.append({ + "solar_w": solar_w[i], + "charge_w": charge_w, + "export_w": export_w, + "soc": soc, + }) + + n_charge_slots = sum(1 for s in schedule if s["charge_w"] > 0) + if n_charge_slots == 0: + return None + + return { + "n_slots": n_charge_slots, + "peak_export_w": peak_export, + "peak_without_charge_w": peak_without, + "total_charged_kwh": total_charged_wh / 1000, + "export_level_w": target_peak, + "schedule": schedule, + }