From f86bce89271a2ba1b0a112519fb5f5ff436c723c Mon Sep 17 00:00:00 2001 From: Rik Allen Date: Wed, 19 Aug 2026 22:19:48 +0100 Subject: [PATCH 1/2] refactor(export): name the 99/100 export_limit sentinel values export_limits_best used 99.0 (freeze export) and 100.0 (idle/off) as undocumented magic numbers throughout plan.py, execute.py, prediction.py, output.py, gateway.py, inverter.py and enphase.py's schedule reconciler. Introduce EXPORT_LIMIT_FREEZE/EXPORT_LIMIT_IDLE in const.py and use them at every genuine sentinel comparison, leaving unrelated same-valued literals (the iboost gas-rate fallback, real charge-to-full/off values, percent-to-fraction conversions) untouched. Value-preserving only - prediction_kernel.cpp still hardcodes the same two literals independently and is annotated to point back here, but is otherwise unchanged, so no parity revision bump or binary rebuild is needed for this change. Co-Authored-By: Claude Sonnet 5 --- apps/predbat/const.py | 10 ++++ apps/predbat/enphase.py | 5 +- apps/predbat/execute.py | 18 +++--- apps/predbat/gateway.py | 9 +-- apps/predbat/inverter.py | 4 +- apps/predbat/output.py | 28 ++++----- apps/predbat/plan.py | 92 +++++++++++++++--------------- apps/predbat/prediction.py | 12 ++-- apps/predbat/prediction_kernel.cpp | 2 +- 9 files changed, 96 insertions(+), 84 deletions(-) diff --git a/apps/predbat/const.py b/apps/predbat/const.py index bf7a6eada..943597c62 100644 --- a/apps/predbat/const.py +++ b/apps/predbat/const.py @@ -64,6 +64,16 @@ INVERTER_TEST = False # Run inverter control self test +# Sentinel values for an export window's target SoC/limit (export_limits_best and friends). +# A real target is any value below EXPORT_LIMIT_FREEZE, expressed as a percentage 0-100 +# (see calc_percent_limit) with the fractional part sometimes encoding a low-power export rate. +# prediction_kernel.cpp hardcodes the same two literals independently (it can't import this +# file) - unlike PREDBAT_MAX_CARS/PK_MAX_CARS above, they aren't yet named there too, so a value +# change here needs the matching literals found and updated by hand, in lockstep with a parity +# revision bump and a rebuild of all platform binaries. +EXPORT_LIMIT_FREEZE = 99.0 # Hold SoC, export only genuine PV surplus - no forced discharge +EXPORT_LIMIT_IDLE = 100.0 # Export window disabled entirely + # Create an array of times in the day in 5-minute intervals BASE_TIME = datetime.strptime("00:00:00", "%H:%M:%S") OPTIONS_TIME = [((BASE_TIME + timedelta(seconds=minute * 60)).strftime("%H:%M:%S")) for minute in range(0, 24 * 60, 5)] diff --git a/apps/predbat/enphase.py b/apps/predbat/enphase.py index 6fb7222ca..42e14c5c7 100644 --- a/apps/predbat/enphase.py +++ b/apps/predbat/enphase.py @@ -28,6 +28,7 @@ import aiohttp from component_base import ComponentBase +from const import EXPORT_LIMIT_FREEZE from mock_base import MockBase from predbat_metrics import record_api_call @@ -1056,8 +1057,8 @@ def _desired_schedule_families(self, local): dtg_limit = max(export_soc, int(local.get("reserve", 0))) return { "cfg": {"enabled": bool(charge.get("enable")), "start": ha_time_to_enphase(charge.get("start_time", "00:00:00")), "end": ha_time_to_enphase(charge.get("end_time", "00:00:00")), "limit": charge.get("soc", 100)}, - "dtg": {"enabled": export_enabled and export_soc < 99, "start": export_start, "end": export_end, "limit": dtg_limit}, - "rbd": {"enabled": export_enabled and export_soc == 99, "start": export_start, "end": export_end, "limit": None}, + "dtg": {"enabled": export_enabled and export_soc < EXPORT_LIMIT_FREEZE, "start": export_start, "end": export_end, "limit": dtg_limit}, + "rbd": {"enabled": export_enabled and export_soc == EXPORT_LIMIT_FREEZE, "start": export_start, "end": export_end, "limit": None}, } async def _cleanup_family(self, site_id, family_key, target, force_recreate): diff --git a/apps/predbat/execute.py b/apps/predbat/execute.py index fe2e2a530..25ff18552 100644 --- a/apps/predbat/execute.py +++ b/apps/predbat/execute.py @@ -16,7 +16,7 @@ # pylint: disable=attribute-defined-outside-init from datetime import timedelta, datetime -from const import MINUTE_WATT +from const import MINUTE_WATT, EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE from utils import dp0, dp2, dp3, calc_percent_limit, find_charge_rate from predbat_metrics import metrics from inverter import Inverter @@ -436,8 +436,8 @@ def execute_plan(self): discharge_end_time = self.midnight_utc + timedelta(minutes=(minutes_end + export_adjust)) # Add in 1 minute margin to allow Predbat to restore demand mode discharge_soc = max((int(self.export_limits_best[0]) * self.soc_max) / 100.0, self.reserve, self.best_soc_min) self.log("Next export window will be: {} - {} at reserve {}".format(discharge_start_time, discharge_end_time, self.export_limits_best[0])) - if (self.minutes_now >= minutes_start) and (self.minutes_now < minutes_end) and (self.export_limits_best[0] < 100.0): - if not self.set_export_freeze_only and self.export_limits_best[0] < 99.0 and (self.soc_kw > discharge_soc): + if (self.minutes_now >= minutes_start) and (self.minutes_now < minutes_end) and (self.export_limits_best[0] < EXPORT_LIMIT_IDLE): + if not self.set_export_freeze_only and self.export_limits_best[0] < EXPORT_LIMIT_FREEZE and (self.soc_kw > discharge_soc): if self.set_export_low_power: export_rate_adjust = 1 - (self.export_limits_best[0] - int(self.export_limits_best[0])) else: @@ -462,7 +462,7 @@ def execute_plan(self): else: inverter.adjust_force_export(False) disabled_export = True - if self.set_export_freeze and self.export_limits_best[0] == 99: + if self.set_export_freeze and self.export_limits_best[0] == EXPORT_LIMIT_FREEZE: # In export freeze mode we disable charging during export slots if inverter.inv_charge_discharge_with_rate: inverter.adjust_charge_rate(0) @@ -490,7 +490,7 @@ def execute_plan(self): self.isExporting_Target = inverter.soc_percent self.log("Export Hold (Demand mode) as export is now at/below target or freeze only is set - current SoC {}kWh and target {}kWh".format(self.soc_kw, discharge_soc)) else: - if (self.minutes_now < minutes_end) and ((minutes_start - self.minutes_now) <= self.set_window_minutes) and (self.export_limits_best[0] < 99.0): + if (self.minutes_now < minutes_end) and ((minutes_start - self.minutes_now) <= self.set_window_minutes) and (self.export_limits_best[0] < EXPORT_LIMIT_FREEZE): # We can't schedule freeze export only full export # Don't turn off ECO mode for GE inverters except when we are within the export window as it will stop the battery being used ge_inverters = inverter.inv_has_ge_eco_toggle or inverter.inv_has_ge_inverter_mode @@ -606,12 +606,12 @@ def execute_plan(self): self.adjust_battery_target_multi(inverter, 0, isCharging, isExporting) # Immediate controls - if self.set_export_freeze and self.export_limits_best[0] == 99: + if self.set_export_freeze and self.export_limits_best[0] == EXPORT_LIMIT_FREEZE: inverter.adjust_export_immediate(inverter.soc_percent, freeze=True) elif not disabled_export: inverter.adjust_export_immediate(int(self.export_limits_best[0])) else: - inverter.adjust_export_immediate(100) # Dead code right, but kept in case other logic changes + inverter.adjust_export_immediate(int(EXPORT_LIMIT_IDLE)) # Dead code right, but kept in case other logic changes elif self.charge_limit_best and (self.minutes_now < inverter.charge_end_time_minutes) and ((inverter.charge_start_time_minutes - self.minutes_now) <= self.set_soc_minutes) and not (disabled_charge_window): if inverter.inv_has_charge_enable_time or isCharging: @@ -698,7 +698,7 @@ def execute_plan(self): else: inverter.adjust_charge_immediate(0) if not isExporting and self.set_export_window: - inverter.adjust_export_immediate(100) + inverter.adjust_export_immediate(int(EXPORT_LIMIT_IDLE)) # Reset reserve as discharge is enable but not running right now if self.set_reserve_enable and resetReserve: @@ -794,7 +794,7 @@ def reset_inverter(self): if self.set_export_window or (self.inverter_needs_reset_force in ["set_read_only", "mode"]): inverter.adjust_discharge_rate(inverter.battery_rate_max_discharge * MINUTE_WATT) inverter.adjust_force_export(False) - inverter.adjust_export_immediate(100) + inverter.adjust_export_immediate(int(EXPORT_LIMIT_IDLE)) self.isExporting = False self.inverter_needs_reset = False diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 0f6d354f5..7ca80ccce 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -16,6 +16,7 @@ import uuid import traceback from utils import calc_percent_limit +from const import EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE import pytz as _pytz from component_base import ComponentBase @@ -363,15 +364,15 @@ def _on_plan_executed(self, charge_windows=None, charge_limits=None, export_wind # Convert export/discharge windows to plan entries for i, window in enumerate(export_windows or []): limit = export_limits[i] if i < len(export_limits or []) else 0 - if limit >= 100: + if limit >= EXPORT_LIMIT_IDLE: continue target_soc = int(limit) export_power_w = discharge_rate_w - # Freeze export (export limit == 99): hold SoC and export only surplus PV + # Freeze export (export limit == EXPORT_LIMIT_FREEZE): hold SoC and export only surplus PV # rather than force-discharge. There is no freeze mode, so express it as a - # discharge entry with rate 0 and target = reserve. Match core's exact == 99 + # discharge entry with rate 0 and target = reserve. Match core's exact == # check — a fractional limit (e.g. 99.5) is a normal export, not a freeze. - if limit == 99: + if limit == EXPORT_LIMIT_FREEZE: target_soc = reserve_percent export_power_w = 0 else: diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 5c299389c..b1dc47028 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -24,7 +24,7 @@ import requests from datetime import datetime, timedelta from config import INVERTER_DEF, SOLAX_SOLIS_MODES_NEW, SOLAX_SOLIS_MODES -from const import MINUTE_WATT, TIME_FORMAT, TIME_FORMAT_OCTOPUS, INVERTER_TEST, TIME_FORMAT_SECONDS, INVERTER_MAX_RETRY, INVERTER_MAX_RETRY_REST, INVERTER_REST_TIMEOUT +from const import MINUTE_WATT, TIME_FORMAT, TIME_FORMAT_OCTOPUS, INVERTER_TEST, TIME_FORMAT_SECONDS, INVERTER_MAX_RETRY, INVERTER_MAX_RETRY_REST, INVERTER_REST_TIMEOUT, EXPORT_LIMIT_IDLE from utils import calc_percent_limit, compute_window_minutes, dp0, dp1, dp2, dp3, dp4, time_string_to_stamp, minute_data, minute_data_state, window2minutes TIME_FORMAT_HMS = "%H:%M:%S" @@ -1669,7 +1669,7 @@ def update_status(self, minutes_now, quiet=False): if self.discharge_enable_time: self.export_limits = [0.0 for i in range(len(self.export_window))] else: - self.export_limits = [100.0 for i in range(len(self.export_window))] + self.export_limits = [EXPORT_LIMIT_IDLE for i in range(len(self.export_window))] # Idle time? # Get previous idle start and end diff --git a/apps/predbat/output.py b/apps/predbat/output.py index 10b7be77a..8e90d4af1 100644 --- a/apps/predbat/output.py +++ b/apps/predbat/output.py @@ -20,7 +20,7 @@ import copy from datetime import datetime, timedelta from config import THIS_VERSION -from const import TIME_FORMAT, PREDICT_STEP +from const import TIME_FORMAT, PREDICT_STEP, EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE from utils import dp0, dp1, dp2, dp3, calc_percent_limit, minute_data, minute_data_state from prediction import Prediction @@ -762,7 +762,7 @@ def get_next_export_window(self, minutes_now): export_window_n = -1 for minute in range(minutes_now, self.forecast_minutes + minutes_now, PREDICT_STEP): export_window_n = self.in_charge_window(self.export_window_best, minute) - if export_window_n >= 0 and self.export_limits_best[export_window_n] == 100: + if export_window_n >= 0 and self.export_limits_best[export_window_n] == EXPORT_LIMIT_IDLE: export_window_n = -1 if export_window_n >= 0: break @@ -774,7 +774,7 @@ def get_charge_export_text(self, minutes_now, charge_window_n, export_window_n): """ if export_window_n >= 0: target_export = self.export_window_best[export_window_n].get("target", self.export_limits_best[export_window_n]) - if self.export_limits_best[export_window_n] == 99: + if self.export_limits_best[export_window_n] == EXPORT_LIMIT_FREEZE: text = "freeze exporting for the next {}".format(self.duration_string(self.export_window_best[export_window_n]["end"] - minutes_now)) # don't include target % for freeze exporting as (the 99%) is meaningless else: text = "force exporting to {}% for the next {}".format(target_export, self.duration_string(self.export_window_best[export_window_n]["end"] - minutes_now)) @@ -817,7 +817,7 @@ def get_export_type(self, export_limit, current=False): """ Get the export type for the given export limit """ - if export_limit == 99: + if export_limit == EXPORT_LIMIT_FREEZE: if current: return "freeze exporting" else: @@ -933,7 +933,7 @@ def short_textual_plan(self, soc_min, soc_min_minute, pv_forecast_minute_step, p charge_window_n = -1 export_window_n = self.in_charge_window(self.export_window_best, self.minutes_now) - if export_window_n >= 0 and self.export_limits_best[export_window_n] == 100: + if export_window_n >= 0 and self.export_limits_best[export_window_n] == EXPORT_LIMIT_IDLE: export_window_n = -1 charge_export_text = self.get_charge_export_text(self.minutes_now, charge_window_n, export_window_n) @@ -1088,7 +1088,7 @@ def publish_html_plan(self, pv_forecast_minute_step, pv_forecast_minute_step10, for try_minute in range(minute_start, minute_end, PREDICT_STEP): export_window_n = self.in_charge_window(self.export_window_best, try_minute) - if export_window_n >= 0 and self.export_limits_best[export_window_n] == 100: + if export_window_n >= 0 and self.export_limits_best[export_window_n] == EXPORT_LIMIT_IDLE: export_window_n = -1 if export_window_n >= 0: break @@ -1104,7 +1104,7 @@ def publish_html_plan(self, pv_forecast_minute_step, pv_forecast_minute_step10, discharge_intersect = -1 for try_minute in range(minute_start, charge_end_minute, PREDICT_STEP): discharge_intersect = self.in_charge_window(self.export_window_best, try_minute) - if discharge_intersect >= 0 and self.export_limits_best[discharge_intersect] == 100: + if discharge_intersect >= 0 and self.export_limits_best[discharge_intersect] == EXPORT_LIMIT_IDLE: discharge_intersect = -1 if discharge_intersect >= 0: break @@ -1361,7 +1361,7 @@ def publish_html_plan(self, pv_forecast_minute_step, pv_forecast_minute_step10, if "target" in self.export_window_best[export_window_n]: target = self.export_window_best[export_window_n]["target"] - if limit == 99: # freeze exporting + if limit == EXPORT_LIMIT_FREEZE: # freeze exporting if not had_state: state = "" if state: @@ -1373,7 +1373,7 @@ def publish_html_plan(self, pv_forecast_minute_step, pv_forecast_minute_step10, raw_state = "FrzExp" show_limit = "" # suppress displaying the limit (of 99) when freeze exporting as its a meaningless number reason_parts.append({"code": "freeze_export", "params": {}}) - elif limit < 100: + elif limit < EXPORT_LIMIT_IDLE: if not had_state: state = "" if state: @@ -2169,7 +2169,7 @@ def publish_export_limit(self, export_window, export_limits, best): export_limit_time_kw = {} export_limit_soc = self.soc_max - export_limit_percent = 100 + export_limit_percent = EXPORT_LIMIT_IDLE export_limit_first = False prev_limit = -1 @@ -2177,7 +2177,7 @@ def publish_export_limit(self, export_window, export_limits, best): window_n = self.in_charge_window(export_window, minute) minute_timestamp = self.midnight_utc + timedelta(minutes=minute) stamp = minute_timestamp.strftime(TIME_FORMAT) - if window_n >= 0 and (export_limits[window_n] < 100.0): + if window_n >= 0 and (export_limits[window_n] < EXPORT_LIMIT_IDLE): soc_perc = export_limits[window_n] soc_kw = (soc_perc * self.soc_max) / 100.0 if not export_limit_first: @@ -2185,7 +2185,7 @@ def publish_export_limit(self, export_window, export_limits, best): export_limit_percent = export_limits[window_n] export_limit_first = True else: - soc_perc = 100 + soc_perc = EXPORT_LIMIT_IDLE soc_kw = self.soc_max if prev_limit != soc_perc: export_limit_time[stamp] = soc_perc @@ -3231,7 +3231,7 @@ def calculate_yesterday(self): self.export_window_best.append({"start": export_start_minute, "end": export_end_minute}) if "freeze" in export_during_slot: # Assume freeze export - self.export_limits_best.append(99.0) + self.export_limits_best.append(EXPORT_LIMIT_FREEZE) else: soc_was = battery_soc_yesterday_array.get(export_end_minute, 0.0) soc_percent = calc_percent_limit(soc_was, self.soc_max) @@ -3540,7 +3540,7 @@ def window_as_text(self, windows, percents, ignore_min=False, ignore_max=False): if ignore_min and percent == 0.0: continue - if ignore_max and percent == 100.0: + if ignore_max and percent == EXPORT_LIMIT_IDLE: continue if not first_window: diff --git a/apps/predbat/plan.py b/apps/predbat/plan.py index 8d2635a44..443ba1f67 100644 --- a/apps/predbat/plan.py +++ b/apps/predbat/plan.py @@ -20,7 +20,7 @@ from datetime import datetime, timedelta from multiprocessing import cpu_count -from const import PREDICT_STEP, PV_SCENARIO_NOMINAL, PV_SCENARIO_PV10, PV_SCENARIO_PV90, TIME_FORMAT, MINUTE_WATT +from const import PREDICT_STEP, PV_SCENARIO_NOMINAL, PV_SCENARIO_PV10, PV_SCENARIO_PV90, TIME_FORMAT, MINUTE_WATT, EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE from utils import calc_percent_limit, clone_windows, dp0, dp1, dp2, dp3, dp4, remove_intersecting_windows, in_car_slot from prediction import Prediction @@ -282,7 +282,7 @@ def find_price_levels( elif typ == "d": if price == real_lowest_price_export: continue - if export_limits[window_n] < 99.0: + if export_limits[window_n] < EXPORT_LIMIT_FREEZE: if lowest_price_export is None: lowest_price_export = export_window[window_n]["average"] else: @@ -421,7 +421,7 @@ def optimise_charge_limit_price_threads( else: price_set_export.append([price, window_n, typ == "df"]) valid_export_windows[window_n] = True - best_export_limits_reset[window_n] = 100.0 + best_export_limits_reset[window_n] = EXPORT_LIMIT_IDLE FINE_SLOT_LENGTHS = [48, 32, 24, 16, 14, 12, 10, 8, 6, 5, 4, 3, 2, 1, 0] COARSE_SLOT_LENGTHS = [32, 16, 8, 4, 2, 1, 0] @@ -445,7 +445,7 @@ def optimise_charge_limit_price_threads( export_hash_delta = {} for window_n in valid_export_windows: reset_contribution = scenario_hash_entry(1, window_n, best_export_limits_reset[window_n]) - export_hash_delta[window_n] = {True: scenario_hash_entry(1, window_n, 99.0) - reset_contribution, False: scenario_hash_entry(1, window_n, min_freeze_percent) - reset_contribution} + export_hash_delta[window_n] = {True: scenario_hash_entry(1, window_n, EXPORT_LIMIT_FREEZE) - reset_contribution, False: scenario_hash_entry(1, window_n, min_freeze_percent) - reset_contribution} # Which charge window an export window collides with is a purely geometric question, and this # function only ever turns windows on and off - it never moves a window's start or end. So the @@ -604,7 +604,7 @@ def export_selection_for(max_slots, allow_freeze, loop_price=loop_price): try_charge_limit[window_n] = self.reserve if freeze else self.soc_max try_export = best_export_limits_reset.copy() for window_n, freeze in export_mods.items(): - try_export[window_n] = 99.0 if freeze else min_freeze_percent + try_export[window_n] = EXPORT_LIMIT_FREEZE if freeze else min_freeze_percent pred_item = {} pred_item["handle"] = self.launch_run_prediction_single(try_charge_limit, charge_window, export_window, try_export, PV_SCENARIO_NOMINAL, end_record=end_record, step=step) @@ -838,7 +838,7 @@ def scenario_summary_state(self, record_time): export_window_n = -1 for try_minute in range(this_minute_absolute, minute_absolute + self.plan_interval_minutes, 5): export_window_n = self.in_charge_window(self.export_window_best, try_minute) - if export_window_n >= 0 and self.export_limits_best[export_window_n] == 100.0: + if export_window_n >= 0 and self.export_limits_best[export_window_n] == EXPORT_LIMIT_IDLE: export_window_n = -1 if export_window_n >= 0: break @@ -859,7 +859,7 @@ def scenario_summary_state(self, record_time): elif export_window_n >= 0: export_target = self.export_limits_best[export_window_n] if export_target >= soc_percent_max: - if export_target == 99: + if export_target == EXPORT_LIMIT_FREEZE: value = "FrzExp" else: value = "HldExp" @@ -1066,7 +1066,7 @@ def plan_fragmentation(self, charge_window, charge_limit, export_window, export_ """ intervals = [] for window, limit in zip(export_window, export_limits): - if limit < 99: + if limit < EXPORT_LIMIT_FREEZE: intervals.append((window["start"], window["end"], "export")) for window, limit in zip(charge_window, charge_limit): if limit > self.reserve: @@ -1306,7 +1306,7 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True): self.charge_limit_best = [self.current_charge_limit * self.soc_max / 100.0 for i in range(len(self.charge_window_best))] # Pre-fill best export enable with Off - self.export_limits_best = [100.0 for i in range(len(self.export_window_best))] + self.export_limits_best = [EXPORT_LIMIT_IDLE for i in range(len(self.export_window_best))] self.end_record = self.forecast_minutes # Show best windows @@ -2160,7 +2160,7 @@ def optimise_export(self, window_n, record_charge_windows, try_charge_limit, cha best_cycle = 0 best_import = 0 best_carbon = 0 - this_export_limit = 100.0 + this_export_limit = EXPORT_LIMIT_IDLE window = export_window[window_n] # A shallow copy is enough: nothing here writes to a window dict, and the one write that does # happen downstream - the trial start - is applied copy-on-write by _prepare_export, which @@ -2179,14 +2179,14 @@ def optimise_export(self, window_n, record_charge_windows, try_charge_limit, cha # loop on each export option if allow_freeze and (freeze_only or self.set_export_freeze_only): - loop_options = [100.0, 99.0] + loop_options = [EXPORT_LIMIT_IDLE, EXPORT_LIMIT_FREEZE] elif allow_freeze and not self.set_export_freeze_only: # If we support freeze, try a 99% option which will freeze at any SoC level below this - loop_options = [100.0, 99.0, 0.0] + loop_options = [EXPORT_LIMIT_IDLE, EXPORT_LIMIT_FREEZE, 0.0] if self.set_export_low_power: loop_options.extend([0.3, 0.5, 0.7]) else: - loop_options = [100.0, 0.0] + loop_options = [EXPORT_LIMIT_IDLE, 0.0] if self.set_export_low_power: loop_options.extend([0.3, 0.5, 0.7]) @@ -2220,7 +2220,7 @@ def optimise_export(self, window_n, record_charge_windows, try_charge_limit, cha continue # Don't optimise start of disabled windows or freeze only windows, just for export ones - if (this_export_limit in [100.0, 99.0]) and (start != window["start"]): + if (this_export_limit in [EXPORT_LIMIT_IDLE, EXPORT_LIMIT_FREEZE]) and (start != window["start"]): continue # Never go below the minimum level @@ -2277,7 +2277,7 @@ def optimise_export(self, window_n, record_charge_windows, try_charge_limit, cha # caller checking whether the plan actually improved has to compare on this instead metric_plan = metric - if this_export_limit == 100.0: + if this_export_limit == EXPORT_LIMIT_IDLE: # Minor weighting to off metric -= 0.002 elif this_export_limit == 0: @@ -2286,7 +2286,7 @@ def optimise_export(self, window_n, record_charge_windows, try_charge_limit, cha # Adjust to try to keep existing windows keep_export = False - if window_n < 2 and this_export_limit < 99.0 and self.export_window and self.isExporting: + if window_n < 2 and this_export_limit < EXPORT_LIMIT_FREEZE and self.export_window and self.isExporting: pwindow = export_window[window_n] dwindow = self.export_window[0] if self.minutes_now >= pwindow["start"] and self.minutes_now < pwindow["end"] and ((self.minutes_now >= dwindow["start"] and self.minutes_now < dwindow["end"]) or (dwindow["end"] == pwindow["start"])): @@ -2336,7 +2336,7 @@ def optimise_export(self, window_n, record_charge_windows, try_charge_limit, cha # Scale back in the case of freeze export as improvements will be smaller rate_scale = 1 - (this_export_limit - int(this_export_limit)) - if this_export_limit == 99: + if this_export_limit == EXPORT_LIMIT_FREEZE: min_improvement_scaled = self.metric_min_improvement_export_freeze elif all_n: min_improvement_scaled = self.metric_min_improvement_export * rate_scale * len(all_n) @@ -2788,10 +2788,10 @@ def prune_dead_plan_slots(self): start_metric = None pruned = 0 trials = 0 - for typ, windows, limits, off_value in (("export", self.export_window_best, self.export_limits_best, 100.0), ("charge", self.charge_window_best, self.charge_limit_best, 0)): + for typ, windows, limits, off_value in (("export", self.export_window_best, self.export_limits_best, EXPORT_LIMIT_IDLE), ("charge", self.charge_window_best, self.charge_limit_best, 0)): for window_n, window in enumerate(windows): limit = limits[window_n] - active = (limit < 100.0) if typ == "export" else (limit > 0) + active = (limit < EXPORT_LIMIT_IDLE) if typ == "export" else (limit > 0) if not active: continue if window["end"] <= self.minutes_now or window["start"] >= record_limit: @@ -2909,7 +2909,7 @@ def clip_export_slots(self, minutes_now, predict_soc, export_window_best, export window_length = window_end - window_start window["target"] = limit - if limit == 100: + if limit == EXPORT_LIMIT_IDLE: # Ignore disabled windows pass elif window_length > 0: @@ -2935,7 +2935,7 @@ def clip_export_slots(self, minutes_now, predict_soc, export_window_best, export # no-SoC-above-reserve (#4171/#4434), phantom export (#4453/#4487) and target-unreachable. # That includes the window covering the current minute, so a dead slot is never left # commanding the inverter. - if limit != 99.0 and soc_min > limit_soc: + if limit != EXPORT_LIMIT_FREEZE and soc_min > limit_soc: # Give it 10 minute margin target_soc = max(limit_soc, soc_min) limit_soc = max(limit_soc, soc_min - 10 * self.battery_rate_max_discharge * self.battery_rate_max_scaling_discharge) @@ -2945,7 +2945,7 @@ def clip_export_slots(self, minutes_now, predict_soc, export_window_best, export self.log("Clip up export window {} from {} - {} from limit {} to new limit {} target set to {}".format(window_n, window_start, window_end, limit, export_limits_best[window_n], window["target"])) else: self.log("Warn: Clip export window {} as it's already passed".format(window_n)) - export_limits_best[window_n] = 100.0 + export_limits_best[window_n] = EXPORT_LIMIT_IDLE return export_window_best, export_limits_best def discard_unused_export_slots(self, export_limits_best, export_window_best): @@ -2955,7 +2955,7 @@ def discard_unused_export_slots(self, export_limits_best, export_window_best): new_best = [] new_enable = [] for window_n in range(len(export_limits_best)): - if export_limits_best[window_n] < 100.0: + if export_limits_best[window_n] < EXPORT_LIMIT_IDLE: # Also merge contiguous enabled windows if ( new_best @@ -3189,14 +3189,14 @@ def optimise_solar(self, best_metric, best_cost, best_keep, best_cycle, best_car # An existing freeze export slot may have been trimmed earlier (start moved later) - # restore it to its original full size so it covers the whole solar period - if self.export_limits_best[window_n] == 99.0: + if self.export_limits_best[window_n] == EXPORT_LIMIT_FREEZE: start_orig = self.export_window_best[window_n].get("start_orig", window_start) if start_orig < window_start: set_window_start(self.export_window_best[window_n], start_orig) continue # Only enable currently idle (disabled) export windows - if self.export_limits_best[window_n] != 100.0: + if self.export_limits_best[window_n] != EXPORT_LIMIT_IDLE: continue # Don't freeze export where a charge is already planned - we can't charge the battery @@ -3217,7 +3217,7 @@ def optimise_solar(self, best_metric, best_cost, best_keep, best_cycle, best_car if pv_period < 0.01: continue - self.export_limits_best[window_n] = 99.0 + self.export_limits_best[window_n] = EXPORT_LIMIT_FREEZE added += 1 if not added: @@ -3236,7 +3236,7 @@ def optimise_solar(self, best_metric, best_cost, best_keep, best_cycle, best_car continue if window_start in self.manual_all_times: continue - if self.export_limits_best[window_n] >= 99.0: + if self.export_limits_best[window_n] >= EXPORT_LIMIT_FREEZE: continue if window_start <= first_solar_minute: continue @@ -3317,8 +3317,8 @@ def optimise_swap_export(self, record_charge_windows, record_export_windows, dro continue # Try to drop the target - if drop and export_limit_target < 100: - self.export_limits_best[window_n_target] = 100.0 + if drop and export_limit_target < EXPORT_LIMIT_IDLE: + self.export_limits_best[window_n_target] = EXPORT_LIMIT_IDLE best_metric_drop, best_battery_value_drop, best_cost_drop, best_keep_drop, best_cycle_drop, best_carbon_drop, best_import_drop, best_export_drop = self.run_prediction_metric( self.charge_limit_best, self.charge_window_best, self.export_window_best, self.export_limits_best, end_record=self.end_record ) @@ -3348,7 +3348,7 @@ def optimise_swap_export(self, record_charge_windows, record_export_windows, dro selected_carbon = best_carbon_drop selected_import = best_import_drop swapped = True - export_limit_target = 100.0 + export_limit_target = EXPORT_LIMIT_IDLE else: self.export_limits_best[window_n_target] = export_limit_target @@ -3387,20 +3387,20 @@ def optimise_swap_export(self, record_charge_windows, record_export_windows, dro # Don't swap if the windows are the same continue - if export_limit < 99 and window_length <= orig_length_target: + if export_limit < EXPORT_LIMIT_FREEZE and window_length <= orig_length_target: # Don't optimise a charge window that hits an export window if this is disallowed if not self.allow_this_export_window(window_n_target): continue is_combined = False - if export_limit_target < 99 and (window_length_target + window_length) <= orig_length_target: + if export_limit_target < EXPORT_LIMIT_FREEZE and (window_length_target + window_length) <= orig_length_target: # Full combine - self.export_limits_best[window_n] = 100 + self.export_limits_best[window_n] = EXPORT_LIMIT_IDLE set_window_start(self.export_window_best[window_n], window_start_orig) self.export_limits_best[window_n_target] = export_limit set_window_start(self.export_window_best[window_n_target], self.export_window_best[window_n_target]["end"] - (window_length + window_length_target)) is_combined = True - elif export_limit_target < 99 and window_length_target < orig_length_target: + elif export_limit_target < EXPORT_LIMIT_FREEZE and window_length_target < orig_length_target: # Partial combine amount_to_move = min(orig_length_target - window_length_target, window_length) window_length_target_new = amount_to_move + window_length_target @@ -3412,7 +3412,7 @@ def optimise_swap_export(self, record_charge_windows, record_export_windows, dro is_combined = True else: # Swap - if export_limit_target < 100 and window_length < window_length_target: + if export_limit_target < EXPORT_LIMIT_IDLE and window_length < window_length_target: # Don't swap if we move a smaller window later continue @@ -3452,7 +3452,7 @@ def optimise_swap_export(self, record_charge_windows, record_export_windows, dro ) ) - if ((selected_metric - best_metric) >= self.metric_min_improvement_swap) and (best_metric <= selected_metric or ((export_limit_target == 100.0 or is_combined))): + if ((selected_metric - best_metric) >= self.metric_min_improvement_swap) and (best_metric <= selected_metric or ((export_limit_target == EXPORT_LIMIT_IDLE or is_combined))): if self.debug_enable: self.log( "Swap export window {} {}-{} limit {} with {} => {}-{} metric {}{}, selected_metric {}{}, min_improvement_swap {}, cost {}{}, keep {}kWh, cycle {}kWh, carbon {}kg, import {}kWh".format( @@ -3611,7 +3611,7 @@ def allow_this_charge_window(self, charge_window_n): if self.calculate_best_charge and (window_start not in self.manual_all_times): if not self.calculate_export_oncharge: hit_export = self.hit_charge_window(self.export_window_best, self.charge_window_best[charge_window_n]["start"], self.charge_window_best[charge_window_n]["end"]) - if hit_export >= 0 and self.export_limits_best[hit_export] < 100: + if hit_export >= 0 and self.export_limits_best[hit_export] < EXPORT_LIMIT_IDLE: return False return True return False @@ -3856,7 +3856,7 @@ def optimise_detailed_pass( continue # Don't trim a window that is already off - if pass_type in ["trim_export"] and (self.export_limits_best[window_n] == 100): + if pass_type in ["trim_export"] and (self.export_limits_best[window_n] == EXPORT_LIMIT_IDLE): continue # In normal don't do trimming of export @@ -3865,15 +3865,15 @@ def optimise_detailed_pass( # Do highest price first # Second pass to tune down any excess exports only - if pass_type == "low" and (self.export_limits_best[window_n] == 100): + if pass_type == "low" and (self.export_limits_best[window_n] == EXPORT_LIMIT_IDLE): continue # Don't trim freeze, that can be done in the freeze pass - if pass_type == "trim_export" and self.export_limits_best[window_n] == 99: + if pass_type == "trim_export" and self.export_limits_best[window_n] == EXPORT_LIMIT_FREEZE: continue # Ignore prices below the threshold if not already selected during levelling - if (price_key < best_price_export_level) and (self.export_limits_best[window_n] == 100): + if (price_key < best_price_export_level) and (self.export_limits_best[window_n] == EXPORT_LIMIT_IDLE): if self.debug_enable: self.log("Skip low window {} best limit {} price_set {} price {} level {}".format(window_n, self.export_limits_best[window_n], price_key, price, best_price_export_level)) continue @@ -3930,7 +3930,7 @@ def optimise_detailed_pass( # never a deeper discharge nor an earlier start (a bigger window exports more, even # when the SoC limit rises). Off/freeze (limit >= 99) export no battery and force the # start back to the full window, so they are exempt from the earlier-start check. - trim_export_ok = pass_type != "trim_export" or (n_best_soc >= self.export_limits_best[window_n] and (n_best_soc >= 99 or n_best_start >= keep_start)) + trim_export_ok = pass_type != "trim_export" or (n_best_soc >= self.export_limits_best[window_n] and (n_best_soc >= EXPORT_LIMIT_FREEZE or n_best_start >= keep_start)) if n_best_metric < best_metric and (n_best_soc != self.export_limits_best[window_n] or n_best_start != self.export_window_best[window_n]["start"]) and trim_export_ok: # Only a strict improvement drives another refinement iteration (see # the charge block above for why equal-metric flips must not). @@ -4241,11 +4241,11 @@ def optimise_charge_windows_manual(self): if self.export_window_best and self.calculate_best_export: for window_n in range(len(self.export_window_best)): if self.export_window_best[window_n]["start"] in self.manual_demand_times: - self.export_limits_best[window_n] = 100.0 + self.export_limits_best[window_n] = EXPORT_LIMIT_IDLE elif self.export_window_best[window_n]["start"] in self.manual_export_times: self.export_limits_best[window_n] = 0.0 elif self.export_window_best[window_n]["start"] in self.manual_freeze_export_times: - self.export_limits_best[window_n] = 99.0 + self.export_limits_best[window_n] = EXPORT_LIMIT_FREEZE def optimise_charge_windows_reset(self, reset_all): """ @@ -4271,9 +4271,9 @@ def optimise_charge_windows_reset(self, reset_all): for window_n in range(len(self.export_window_best)): if self.export_window_best[window_n]["start"] < (self.minutes_now + self.end_record): if reset_all: - self.export_limits_best[window_n] = 100.0 + self.export_limits_best[window_n] = EXPORT_LIMIT_IDLE else: - self.export_limits_best[window_n] = 100.0 + self.export_limits_best[window_n] = EXPORT_LIMIT_IDLE def run_prediction(self, charge_limit, charge_window, export_window, export_limits, pv_scenario, end_record, save=None, step=PREDICT_STEP): """ diff --git a/apps/predbat/prediction.py b/apps/predbat/prediction.py index a89d8d2f7..75a36d0f3 100644 --- a/apps/predbat/prediction.py +++ b/apps/predbat/prediction.py @@ -18,7 +18,7 @@ """ from datetime import timedelta -from const import PREDICT_STEP, PV_SCENARIO_PV10, PV_SCENARIO_PV90, RUN_EVERY, TIME_FORMAT +from const import PREDICT_STEP, PV_SCENARIO_PV10, PV_SCENARIO_PV90, RUN_EVERY, TIME_FORMAT, EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE from utils import remove_intersecting_windows, get_charge_rate_curve_cached, get_discharge_rate_curve_cached, find_charge_rate, calc_percent_limit, in_iboost_slot, in_car_slot, charge_curve_to_tuple from prediction_batch import PredictionBatch, prediction_cache_key @@ -446,7 +446,7 @@ def find_charge_window_optimised(self, charge_windows, charge_limit, is_export=F charge_window_optimised = {} for window_n in range(len(charge_windows)): for minute in range(charge_windows[window_n]["start"], charge_windows[window_n]["end"], PREDICT_STEP): - if is_export and charge_limit[window_n] < 100.0: + if is_export and charge_limit[window_n] < EXPORT_LIMIT_IDLE: charge_window_optimised[minute] = window_n elif not is_export and charge_limit[window_n] > 0.0: charge_window_optimised[minute] = window_n @@ -712,7 +712,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi export_window_n = export_window_optimised.get(minute_absolute, -1) charge_window_active = charge_window_n >= 0 export_window_active = export_window_n >= 0 - export_limit_now = export_limits[export_window_n] if export_window_active else 100.0 + export_limit_now = export_limits[export_window_n] if export_window_active else EXPORT_LIMIT_IDLE # Find charge limit charge_limit_n = 0 @@ -874,7 +874,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi # discharge freeze, reset charge rate by default if set_export_freeze: # Freeze mode - if (export_window_active) and export_limit_now < 100.0 and (set_export_freeze and (export_limit_now == 99.0 or set_export_freeze_only)): + if (export_window_active) and export_limit_now < EXPORT_LIMIT_IDLE and (set_export_freeze and (export_limit_now == EXPORT_LIMIT_FREEZE or set_export_freeze_only)): charge_rate_now = battery_rate_min # 0 # Set discharge during charge? @@ -902,7 +902,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi if export_window_active: discharge_min = max(soc_max * export_limit_now / 100.0, reserve, self.best_soc_min) - if not set_export_freeze_only and export_window_active and export_limit_now < 99.0 and (soc > discharge_min): + if not set_export_freeze_only and export_window_active and export_limit_now < EXPORT_LIMIT_FREEZE and (soc > discharge_min): # Discharge enable, capped at export limit if self.set_export_low_power: export_rate_adjust = 1 - (export_limit_now - int(export_limit_now)) @@ -1089,7 +1089,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi if inverter_hybrid: charge_rate_now_dc = battery_rate_max_charge_dc # Freeze mode - if set_export_freeze and export_window_active and export_limit_now < 100.0 and (export_limit_now == 99.0 or set_export_freeze_only): + if set_export_freeze and export_window_active and export_limit_now < EXPORT_LIMIT_IDLE and (export_limit_now == EXPORT_LIMIT_FREEZE or set_export_freeze_only): charge_rate_now_dc = battery_rate_min # 0 charge_rate_now_curve_dc = ( diff --git a/apps/predbat/prediction_kernel.cpp b/apps/predbat/prediction_kernel.cpp index ac1a0bc26..d79896bed 100644 --- a/apps/predbat/prediction_kernel.cpp +++ b/apps/predbat/prediction_kernel.cpp @@ -232,7 +232,7 @@ struct PkScenario { const double *charge_limit; // kWh target per charge window const int32_t *charge_start; // absolute minutes const int32_t *charge_end; - const double *export_limits; // percent per export window (99=freeze, 100=off) + const double *export_limits; // percent per export window (99=freeze, 100=off - see EXPORT_LIMIT_FREEZE/EXPORT_LIMIT_IDLE in const.py) const int32_t *export_start; const int32_t *export_end; double *soc_out; // caller-allocated, n_steps entries, filled with round(soc, 3) From 401cc7041f1e31b27cb81da71bcfa39cddf94a2f Mon Sep 17 00:00:00 2001 From: Trefor Southwell <48591903+springfall2008@users.noreply.github.com> Date: Fri, 21 Aug 2026 09:01:02 +0100 Subject: [PATCH 2/2] Update output.py --- apps/predbat/output.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/predbat/output.py b/apps/predbat/output.py index 848c41b5e..b78302fe5 100644 --- a/apps/predbat/output.py +++ b/apps/predbat/output.py @@ -20,7 +20,7 @@ import copy from datetime import datetime, timedelta from config import THIS_VERSION -from const import TIME_FORMAT, PREDICT_STEP, EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE +from const import TIME_FORMAT, PREDICT_STEP, EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE, MINUTE_WATT from utils import dp0, dp1, dp2, dp3, calc_percent_limit, minute_data, minute_data_state, find_charge_rate from prediction import Prediction