feat: add manual_soc_max, a discharge-direction ceiling sibling to manual_soc (#1578)

manual_soc already lets an automation set a minimum SoC floor for a
specific time - Predbat plans charging to reach it. There was no
equivalent for the opposite direction: forcing the battery down to a
ceiling by a deadline, which is what #1578 actually needs (a periodic
calibration discharge finishing right before a known cheap import slot,
e.g. an Octopus Intelligent Go midnight slot).

manual_soc_max mirrors manual_soc's existing mechanism exactly, just as
a ceiling: it merges into a new all_active_keep_max per-minute dict
(fetch.py), and the optimiser treats being above the ceiling as a cost
penalty priced at the export rate (prediction.py) - the same "keep"
scoring approach already used for the floor, just inverted. Because
it's a real cost signal integrated into the normal search rather than a
fixed-duration forced export, the optimiser is free to use the energy
against load or export depending on what's actually cheapest at each
moment, which should make better use of the energy than a blunt forced
discharge automation.

A ceiling below the floor at the same minute is a contradiction (most
likely a stale override left over from a previous automation run) - the
floor wins as the safety-relevant constraint, and the conflicting
ceiling is dropped with a warning rather than handing the optimiser two
penalties pulling opposite ways.

This touches the hot loop that has both a pure-Python and a C++ kernel
implementation (prediction_kernel.cpp) for performance - mirrored the
new ceiling logic there too and bumped both parity revisions per the
kernel's own documented rule, so a stale prebuilt kernel binary falls
back to the Python engine instead of silently disagreeing with it.
Verified locally: rebuilt the kernel and the full parity suite passes,
including a new randomized ceiling scenario added to its existing
seeded sweep (constructed to consume no additional RNG draws, so the
existing seeded scenarios are unaffected).

Also covered: the web plan page's SOC dropdown gained a second "Maximum
SOC" control (web.py/web_helper.py), plan-status markers mirroring the
existing floor indicator (execute.py/output.py), and the export-window
merge guard already applied to the floor is now applied to the ceiling
too (plan.py) so two export windows can't be silently combined across a
ceiling deadline.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: chalfontchubby

.cspell/custom-dictionary-workspace.txt

@@ -568,6 +568,7 @@ snakeviz
 snprintf
 socb
 socketloop
+socmax
 socs
 sofar
 SolarEdge

apps/predbat/config.py

@@ -1394,6 +1394,27 @@
         "icon": "mdi:battery-charging-100",
         "default": 100,
     },
+    {
+        "name": "manual_soc_max",
+        "friendly_name": "Manual SOC maximum target",
+        "type": "select",
+        "options": ["off"],
+        "icon": "mdi:battery-arrow-down",
+        "default": "off",
+        "restore": False,
+        "manual_rate": True,
+    },
+    {
+        "name": "manual_soc_max_value",
+        "friendly_name": "Manual SOC maximum target value",
+        "type": "input_number",
+        "min": 0,
+        "max": 100,
+        "step": 1,
+        "unit": "%",
+        "icon": "mdi:battery-arrow-down-outline",
+        "default": 0,
+    },
     {
         "name": "manual_api",
         "friendly_name": "Manual API controls",

apps/predbat/execute.py

@@ -137,6 +137,7 @@ def execute_plan(self):
 
         in_alert = self.alert_active_keep.get(self.minutes_now, 0) > 0
         in_manual_soc = self.manual_soc_keep.get(self.minutes_now, 0) > 0
+        in_manual_soc_max = self.manual_soc_max_keep.get(self.minutes_now, 0) > 0
 
         # Safeguard for set_charge_freeze_only: the planner never selects a charge target above the
         # reserve while the switch is on, but a plan computed before it was turned on can still be
@@ -776,6 +777,8 @@ def execute_plan(self):
             status += " [Alert]"
         if in_manual_soc:
             status += " [Manual SoC]"
+        if in_manual_soc_max:
+            status += " [Manual SoC Max]"
 
         return status, status_extra
 

apps/predbat/fetch.py

@@ -961,6 +961,25 @@ def fetch_sensor_data(self, save=True):
                 else:
                     self.all_active_keep[minute] = soc_value
 
+        # Manual SOC max is a ceiling rather than a floor - combine separately, taking the
+        # tightest (lowest) ceiling if more than one source ever applies to the same minute.
+        self.all_active_keep_max = {}
+        if self.manual_soc_max_keep:
+            for minute, soc_value in self.manual_soc_max_keep.items():
+                if minute in self.all_active_keep_max:
+                    self.all_active_keep_max[minute] = min(self.all_active_keep_max[minute], soc_value)
+                else:
+                    self.all_active_keep_max[minute] = soc_value
+
+        # A ceiling below the floor at the same minute is a contradiction (e.g. a leftover manual_soc
+        # override never cleared) - the floor wins as the safety-relevant constraint, so drop the
+        # conflicting ceiling rather than hand the optimiser two penalties pulling opposite ways.
+        for minute in list(self.all_active_keep_max.keys()):
+            floor_value = self.all_active_keep.get(minute, 0)
+            if floor_value > self.all_active_keep_max[minute]:
+                self.log("Warn: manual_soc_max target {}% at minute {} is below the manual_soc/alert floor {}% for the same minute - ignoring the ceiling there".format(self.all_active_keep_max[minute], minute, floor_value))
+                del self.all_active_keep_max[minute]
+
         # iBoost load data
         if "iboost_energy_today" in self.args:
             self.iboost_energy_today, iboost_energy_age = self.minute_data_load(self.now_utc, "iboost_energy_today", self.max_days_previous, required_unit="kWh", load_scaling=1.0)
@@ -3216,6 +3235,7 @@ def fetch_config_options(self):
         self.manual_export_rates = self.manual_rates("manual_export_rates", default_rate=self.get_arg("manual_export_value"))
         self.manual_load_adjust = self.manual_rates("manual_load_adjust", default_rate=self.get_arg("manual_load_value"))
         self.manual_soc_keep = self.manual_rates("manual_soc", default_rate=self.get_arg("manual_soc_value"))
+        self.manual_soc_max_keep = self.manual_rates("manual_soc_max", default_rate=self.get_arg("manual_soc_max_value"))
 
         # Update list of config options to save/restore to
         self.update_save_restore_list()

apps/predbat/output.py

@@ -1232,6 +1232,7 @@ def import_rate_color(rate):
 
             in_alert = self.alert_active_keep.get(minute, 0) > 0
             in_manual_soc = self.manual_soc_keep.get(minute, 0) > 0
+            in_manual_soc_max = self.manual_soc_max_keep.get(minute, 0) > 0
 
             pv_forecast = 0
             load_forecast = 0
@@ -1519,6 +1520,8 @@ def import_rate_color(rate):
                 soc_sym = "⚠ " + soc_sym
             if in_manual_soc:
                 soc_sym = "✎ " + soc_sym
+            if in_manual_soc_max:
+                soc_sym = "⬇ " + soc_sym
 
             # Import and export rates -> to string
             adjust_type = self.rate_import_replicated.get(minute, None)

apps/predbat/plan.py

@@ -3099,6 +3099,8 @@ def discard_unused_export_slots(self, export_limits_best, export_window_best):
                     and (export_limits_best[window_n] == new_enable[-1])
                     and (export_window_best[window_n]["start"] not in self.manual_all_times)
                     and (new_best[-1]["start"] not in self.manual_all_times)
+                    and (export_window_best[window_n]["start"] not in self.all_active_keep_max)
+                    and (new_best[-1]["start"] not in self.all_active_keep_max)
                 ):
                     new_best[-1]["end"] = export_window_best[window_n]["end"]
                     new_best[-1]["target"] = export_window_best[window_n].get("target", export_limits_best[window_n])

apps/predbat/predbat.py

@@ -615,7 +615,9 @@ def reset(self):
         self.alerts = []
         self.alert_active_keep = {}
         self.manual_soc_keep = {}
+        self.manual_soc_max_keep = {}
         self.all_active_keep = {}
+        self.all_active_keep_max = {}
         self.set_charge_low_power = False
         self.set_export_low_power = False
         self.config_root = "./"

apps/predbat/prediction.py

@@ -169,6 +169,7 @@ def __init__(
             self.load_minutes_step90 = load_minutes_step90 if load_minutes_step90 is not None else load_minutes_step
             self.carbon_intensity = base.carbon_intensity
             self.all_active_keep = base.all_active_keep
+            self.all_active_keep_max = base.all_active_keep_max
             self.iboost_running = False
             self.iboost_running_solar = False
             self.iboost_running_full = False
@@ -628,6 +629,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
         battery_loss_discharge = self.battery_loss_discharge
         battery_temperature_prediction = self.battery_temperature_prediction
         all_active_keep = self.all_active_keep
+        all_active_keep_max = self.all_active_keep_max
         best_soc_keep_weight = self.best_soc_keep_weight
         best_soc_keep_orig = self.best_soc_keep
         debug_enable = self.debug_enable
@@ -697,6 +699,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
 
             # Alert?
             alert_keep = all_active_keep.get(minute_absolute, 0)
+            alert_keep_max = all_active_keep_max.get(minute_absolute, 0)
 
             # Project battery temperature
             battery_temperature = battery_temperature_prediction.get(minute, self.battery_temperature)
@@ -715,6 +718,12 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
                 keep_minute_scaling = max(keep_minute_scaling, 10.0)
                 best_soc_keep = max(best_soc_keep, min(alert_keep / 100.0 * soc_max, soc_max))
 
+            # Soc max keep is a ceiling rather than a floor (e.g. manual_soc_max) - 0 means no ceiling
+            best_soc_max = 0
+            if alert_keep_max > 0:
+                keep_minute_scaling = max(keep_minute_scaling, 10.0)
+                best_soc_max = min(alert_keep_max / 100.0 * soc_max, soc_max)
+
             # Find charge & discharge windows
             charge_window_n = charge_window_optimised.get(minute_absolute, -1)
             export_window_n = export_window_optimised.get(minute_absolute, -1)
@@ -1275,6 +1284,10 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
             if best_soc_keep > 0 and soc <= best_soc_keep:
                 metric_keep += (best_soc_keep - soc) * import_rate * keep_minute_scaling * step / 60.0
 
+            # Metric keep max - pretend the excess above the ceiling should have been exported instead of held
+            if best_soc_max > 0 and soc >= best_soc_max:
+                metric_keep += (soc - best_soc_max) * export_rate * keep_minute_scaling * step / 60.0
+
             if diff > 0:
                 # Import
                 # All imports must go to home (no inverter loss) or to the battery (inverter loss accounted before above)

apps/predbat/prediction_kernel.cpp

@@ -43,7 +43,7 @@
 // falling back. Bumping makes the loader reject it and use the Python engine, which is the whole
 // point of the check.
 #define PK_ABI_VERSION 5
-#define PK_PARITY_REVISION 10
+#define PK_PARITY_REVISION 11
 #define PK_MAX_CARS 8
 #define PK_RUN_EVERY 5 // const.py RUN_EVERY
 #define PK_EXPORT_LIMIT_FREEZE 99.0 // const.py EXPORT_LIMIT_FREEZE
@@ -145,6 +145,7 @@ struct PkContext {
     const double *rate_import;        // import rate per step
     const double *rate_export;        // export rate per step
     const double *alert_keep;         // alert keep value per step
+    const double *alert_keep_max;     // soc ceiling keep value per step (manual_soc_max), 0 = no ceiling
     const double *pv;                 // PV forecast kWh per step (central)
     const double *load;               // load kWh per step (central)
     const double *pv10;               // PV forecast kWh per step (PV10)
@@ -360,7 +361,7 @@ static PkScratch &thread_scratch()
 
 // Deep-copied context storage so Python-side buffers can be freed after create
 struct ContextStore {
-    std::vector<double> rate_import, rate_export, alert_keep;
+    std::vector<double> rate_import, rate_export, alert_keep, alert_keep_max;
     std::vector<double> pv, load, pv10, load10, pv90, load90;
     std::vector<double> temp_charge_cap, temp_discharge_cap;
     std::vector<int32_t> io_flag;
@@ -550,6 +551,7 @@ int64_t pk_context_create(const PkContext *in)
     store->rate_import.assign(in->rate_import, in->rate_import + n);
     store->rate_export.assign(in->rate_export, in->rate_export + n);
     store->alert_keep.assign(in->alert_keep, in->alert_keep + n);
+    store->alert_keep_max.assign(in->alert_keep_max, in->alert_keep_max + n);
     store->pv.assign(in->pv, in->pv + n);
     store->load.assign(in->load, in->load + n);
     store->pv10.assign(in->pv10, in->pv10 + n);
@@ -573,6 +575,7 @@ int64_t pk_context_create(const PkContext *in)
     store->ctx.rate_import = store->rate_import.data();
     store->ctx.rate_export = store->rate_export.data();
     store->ctx.alert_keep = store->alert_keep.data();
+    store->ctx.alert_keep_max = store->alert_keep_max.data();
     store->ctx.pv = store->pv.data();
     store->ctx.load = store->load.data();
     store->ctx.pv10 = store->pv10.data();
@@ -740,6 +743,7 @@ static int32_t pk_run_one(const ContextStore *store, const PkScenario *s, PkResu
 
         // Alert - prediction.py:583
         const double alert_keep = c->alert_keep[k];
+        const double alert_keep_max = c->alert_keep_max[k];
 
         // Four hour rule scaling - prediction.py:589-592
         double keep_minute_scaling = four_hour_rule ? std::min(minute / 240.0, 1.0) * best_soc_keep_weight : best_soc_keep_weight;
@@ -751,6 +755,14 @@ static int32_t pk_run_one(const ContextStore *store, const PkScenario *s, PkResu
             best_soc_keep = std::max(best_soc_keep, std::min(alert_keep / 100.0 * soc_max, soc_max));
         }
 
+        // Soc max keep is a ceiling rather than a floor (manual_soc_max) - mirrors prediction.py's
+        // best_soc_max block right after the alert keep floor. 0 = no ceiling.
+        double best_soc_max = 0;
+        if (alert_keep_max > 0) {
+            keep_minute_scaling = std::max(keep_minute_scaling, 10.0);
+            best_soc_max = std::min(alert_keep_max / 100.0 * soc_max, soc_max);
+        }
+
         // Find charge & discharge windows - prediction.py:602-607
         const int32_t charge_window_n = charge_window_optimised[k];
         const int32_t export_window_n = export_window_optimised[k];
@@ -1262,6 +1274,12 @@ static int32_t pk_run_one(const ContextStore *store, const PkScenario *s, PkResu
             metric_keep += (best_soc_keep - soc) * import_rate * keep_minute_scaling * step / 60.0;
         }
 
+        // Metric keep max - pretend the excess above the ceiling should have been exported instead
+        // of held - mirrors prediction.py's best_soc_max block right after the floor.
+        if (best_soc_max > 0 && soc >= best_soc_max) {
+            metric_keep += (soc - best_soc_max) * export_rate * keep_minute_scaling * step / 60.0;
+        }
+
         // Import/export accounting - prediction.py:1104-1143
         if (diff > 0) {
             // Import

apps/predbat/prediction_kernel.py

@@ -32,7 +32,7 @@
 
 # Expected ABI/parity revisions of the shared library (see prediction_kernel.cpp)
 KERNEL_ABI_VERSION = 5
-KERNEL_PARITY_REVISION = 10
+KERNEL_PARITY_REVISION = 11
 
 # Maximum number of cars supported by the kernel (PK_MAX_CARS in prediction_kernel.cpp)
 KERNEL_MAX_CARS = PREDBAT_MAX_CARS
@@ -53,6 +53,7 @@ class PkContext(ctypes.Structure):
         ("rate_import", ctypes.POINTER(ctypes.c_double)),
         ("rate_export", ctypes.POINTER(ctypes.c_double)),
         ("alert_keep", ctypes.POINTER(ctypes.c_double)),
+        ("alert_keep_max", ctypes.POINTER(ctypes.c_double)),
         ("pv", ctypes.POINTER(ctypes.c_double)),
         ("load", ctypes.POINTER(ctypes.c_double)),
         ("pv10", ctypes.POINTER(ctypes.c_double)),
@@ -527,6 +528,7 @@ def build_static_context_arrays(pred, n_steps, minutes_now, num_cars):
     rate_import = []
     rate_export = []
     alert_keep = []
+    alert_keep_max = []
     io_flag = []
     pv = []
     pv10 = []
@@ -549,6 +551,7 @@ def build_static_context_arrays(pred, n_steps, minutes_now, num_cars):
         rate_import.append(pred.rate_import.get(minute_absolute, 0))
         rate_export.append(pred.rate_export.get(minute_absolute, 0))
         alert_keep.append(pred.all_active_keep.get(minute_absolute, 0))
+        alert_keep_max.append(pred.all_active_keep_max.get(minute_absolute, 0))
         io_flag.append(1 if pred.io_adjusted.get(minute_absolute, 0) else 0)
         pv.append(pred.pv_forecast_minute_step[minute])
         pv10.append(pred.pv_forecast_minute10_step[minute])
@@ -579,7 +582,7 @@ def build_static_context_arrays(pred, n_steps, minutes_now, num_cars):
     charge_curve = [get_curve_value(pred.battery_charge_power_curve, percent, 1.0) for percent in range(101)]
     discharge_curve = [get_curve_value(pred.battery_discharge_power_curve, percent, 1.0) for percent in range(101)]
 
-    return (rate_import, rate_export, alert_keep, io_flag, pv, pv10, pv90, temp_charge_cap, temp_discharge_cap, carbon, gas_rate, iboost_plan_load, car_load_flat, car_rate_flat, charge_curve, discharge_curve)
+    return (rate_import, rate_export, alert_keep, alert_keep_max, io_flag, pv, pv10, pv90, temp_charge_cap, temp_discharge_cap, carbon, gas_rate, iboost_plan_load, car_load_flat, car_rate_flat, charge_curve, discharge_curve)
 
 
 def create_kernel_context(pred, static_cache=None):
@@ -628,12 +631,13 @@ def create_kernel_context(pred, static_cache=None):
             static = (shape, build_static_context_arrays(pred, n_steps, minutes_now, num_cars))
             if static_cache is not None:
                 static_cache["arrays"] = static
-        (rate_import, rate_export, alert_keep, io_flag, pv, pv10, pv90, temp_charge_cap, temp_discharge_cap, carbon, gas_rate, iboost_plan_load, car_load_flat, car_rate_flat, charge_curve, discharge_curve) = static[1]
+        (rate_import, rate_export, alert_keep, alert_keep_max, io_flag, pv, pv10, pv90, temp_charge_cap, temp_discharge_cap, carbon, gas_rate, iboost_plan_load, car_load_flat, car_rate_flat, charge_curve, discharge_curve) = static[1]
 
         ctx = PkContext()
         ctx.rate_import = double_array(rate_import)
         ctx.rate_export = double_array(rate_export)
         ctx.alert_keep = double_array(alert_keep)
+        ctx.alert_keep_max = double_array(alert_keep_max)
         ctx.pv = double_array(pv)
         ctx.load = double_array(load)
         ctx.pv10 = double_array(pv10)