diff --git a/apps/predbat/prediction.py b/apps/predbat/prediction.py index 690b686be..708e8548e 100644 --- a/apps/predbat/prediction.py +++ b/apps/predbat/prediction.py @@ -1230,8 +1230,11 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi predict_state[stamp] = "g" + grid_state + "b" + battery_state predict_battery_power[stamp] = round(battery_draw * (60 / step), 3) predict_battery_cycle[stamp] = round(battery_cycle, 3) - # Use plan_interval_minutes instead of hardcoded 30 for scaling - predict_pv_power[stamp] = round((pv_forecast_minute_step[minute] + pv_forecast_minute_step.get(minute + step, 0)) * (self.plan_interval_minutes / step), 3) + # Two consecutive `step`-sized energy chunks cover 2*step minutes; convert to an + # instantaneous kW reading with 60/(2*step) - a constant derived from `step` (the + # simulation's fixed PREDICT_STEP), not plan_interval_minutes, which is unrelated + # to how many raw steps are being summed here. + predict_pv_power[stamp] = round((pv_forecast_minute_step[minute] + pv_forecast_minute_step.get(minute + step, 0)) * (60 / (2 * step)), 3) predict_grid_power[stamp] = round(diff * (60 / step), 3) predict_load_power[stamp] = round(load_yesterday * (60 / step), 3) if carbon_enable: diff --git a/apps/predbat/tests/test_predict_pv_power.py b/apps/predbat/tests/test_predict_pv_power.py new file mode 100644 index 000000000..8358030dc --- /dev/null +++ b/apps/predbat/tests/test_predict_pv_power.py @@ -0,0 +1,102 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# fmt off +# pylint: disable=consider-using-f-string +# pylint: disable=line-too-long +# pylint: disable=attribute-defined-outside-init + +from prediction import Prediction +from tests.test_infra import reset_inverter + + +def _run_flat_pv(my_predbat, plan_interval_minutes, flat_pv_kw): + """ + Run a real run_prediction(save="best") with a known, flat PV forecast and return the + resulting Prediction object so predict_pv_power can be inspected. + """ + my_predbat.plan_interval_minutes = plan_interval_minutes + reset_inverter(my_predbat) + my_predbat.forecast_minutes = 24 * 60 + my_predbat.end_record = my_predbat.forecast_minutes + + per_minute_kwh = flat_pv_kw / 60.0 + horizon = my_predbat.forecast_minutes + my_predbat.minutes_now + 10 + my_predbat.pv_forecast_minute = {m: per_minute_kwh for m in range(0, horizon)} + my_predbat.pv_forecast_minute10 = dict(my_predbat.pv_forecast_minute) + my_predbat.load_minutes_step = {m: 0.0 for m in range(0, horizon)} + my_predbat.load_minutes_step10 = dict(my_predbat.load_minutes_step) + + pv_step = {m: per_minute_kwh * 5 for m in range(0, horizon, 5)} + load_step = {m: 0.0 for m in range(0, horizon, 5)} + my_predbat.prediction = Prediction(my_predbat, pv_step, pv_step, load_step, load_step) + + my_predbat.run_prediction([], [], [], [], False, end_record=my_predbat.end_record, save="best") + return my_predbat.prediction + + +def test_predict_pv_power_matches_input_regardless_of_plan_interval(my_predbat): + """ + Regression test for issue #4361: predict_pv_power (which feeds pv_power_best, the source of + the "PV Power (Predicted)" trace on the LoadMLPower chart) must report the same instantaneous + kW as the underlying flat PV forecast, regardless of plan_interval_minutes. + + Root cause: the formula converts two summed PREDICT_STEP (5-min) energy chunks into an + instantaneous kW reading via a 60/(2*step) constant - PR #2865 (2025-11-07) accidentally + replaced that constant with self.plan_interval_minutes, which is unrelated to how many raw + 5-minute steps are being summed. That made the reported PV power scale with + plan_interval_minutes/30 - correct only for the 30-minute default, and exactly half for + anyone on 15-minute plan intervals (as in #4361), a third for 10-minute, etc. + """ + print("**** Running predict_pv_power plan-interval-independence test ****") + failed = False + + # my_predbat is a shared fixture reused by every test module in the suite - save everything + # this test mutates and restore it afterwards so later tests (e.g. model_kernel, which assumes + # the 30-minute default) aren't left running against a stale plan_interval_minutes. + original_plan_interval_minutes = my_predbat.plan_interval_minutes + original_forecast_minutes = my_predbat.forecast_minutes + original_end_record = my_predbat.end_record + original_pv_forecast_minute = getattr(my_predbat, "pv_forecast_minute", None) + original_pv_forecast_minute10 = getattr(my_predbat, "pv_forecast_minute10", None) + original_load_minutes_step = getattr(my_predbat, "load_minutes_step", None) + original_load_minutes_step10 = getattr(my_predbat, "load_minutes_step10", None) + original_prediction = getattr(my_predbat, "prediction", None) + + try: + flat_pv_kw = 4.0 + for plan_interval_minutes in [30, 15, 10]: + pred = _run_flat_pv(my_predbat, plan_interval_minutes, flat_pv_kw) + values = list(pred.predict_pv_power.values()) + # Skip the first/last couple of samples in case of edge alignment effects, check a mid-horizon one + sample = values[len(values) // 2] + if abs(sample - flat_pv_kw) > 0.01: + print("ERROR: plan_interval_minutes={} flat PV input={}kW but predict_pv_power sample={}kW".format(plan_interval_minutes, flat_pv_kw, sample)) + failed = True + else: + print(" plan_interval_minutes={}: predict_pv_power={}kW (expected {}kW) - OK".format(plan_interval_minutes, sample, flat_pv_kw)) + finally: + my_predbat.plan_interval_minutes = original_plan_interval_minutes + my_predbat.forecast_minutes = original_forecast_minutes + my_predbat.end_record = original_end_record + my_predbat.pv_forecast_minute = original_pv_forecast_minute + my_predbat.pv_forecast_minute10 = original_pv_forecast_minute10 + my_predbat.load_minutes_step = original_load_minutes_step + my_predbat.load_minutes_step10 = original_load_minutes_step10 + my_predbat.prediction = original_prediction + + if failed: + print("Test: predict_pv_power_matches_input_regardless_of_plan_interval FAILED") + else: + print("Test: predict_pv_power_matches_input_regardless_of_plan_interval PASSED") + + return failed + + +def run_predict_pv_power_tests(my_predbat): + """Run all predict_pv_power regression tests.""" + failed = False + failed |= test_predict_pv_power_matches_input_regardless_of_plan_interval(my_predbat) + return failed diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index 5dd1cb4b8..1ffef4661 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -19,6 +19,7 @@ from tests.test_compute_metric import run_compute_metric_tests from tests.test_perf import run_perf_test from tests.test_model import run_model_tests +from tests.test_predict_pv_power import run_predict_pv_power_tests from tests.test_kernel_parity import run_kernel_parity_tests, run_model_kernel_tests from tests.test_execute import run_execute_tests from tests.test_octopus_slots import run_load_octopus_slots_tests @@ -225,6 +226,7 @@ def main(): ("secrets", run_secrets_tests, "Secrets loading tests", False), ("perf", run_perf_test, "Performance tests", False), ("model", run_model_tests, "Model tests", False), + ("predict_pv_power", run_predict_pv_power_tests, "predict_pv_power plan-interval scaling tests", False), ("model_kernel", run_model_kernel_tests, "Model tests run with the C++ prediction kernel enabled", False), ("kernel_parity", run_kernel_parity_tests, "C++ prediction kernel vs Python engine parity tests", False), ("inverter", run_inverter_tests, "Inverter tests", False),