Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions apps/predbat/execute.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
control actions. Manages charge window programming, discharge/export scheduling,
reserve level adjustments, and multi-inverter balancing.
"""

# -----------------------------------------------------------------------------
# Predbat Home Battery System
# Copyright Trefor Southwell 2026 - All Rights Reserved
Expand Down Expand Up @@ -752,6 +753,9 @@ def fetch_inverter_data(self, create=True):
self.battery_charge_power_curve = self.battery_charge_power_curve_default
self.computed_charge_curve = True
self.log("Using default battery charge power curve")
elif not self.battery_charge_power_curve_auto:
# Stop retrying every cycle when not in auto mode and no curve found
self.computed_charge_curve = True

if id == 0 and (not self.computed_discharge_curve or self.battery_discharge_power_curve_auto) and not self.battery_discharge_power_curve:
curve = inverter.find_charge_curve(discharge=True)
Expand All @@ -764,6 +768,9 @@ def fetch_inverter_data(self, create=True):
self.battery_discharge_power_curve = self.battery_discharge_power_curve_default
self.computed_discharge_curve = True
self.log("Using default battery discharge power curve")
elif not self.battery_discharge_power_curve_auto:
# Stop retrying every cycle when not in auto mode and no curve found
self.computed_discharge_curve = True

# As the inverters will run in lockstep, we will initially look at the programming of the first enabled one for the current window setting
if not found_first:
Expand Down
4 changes: 2 additions & 2 deletions apps/predbat/inverter.py
Original file line number Diff line number Diff line change
Expand Up @@ -1166,8 +1166,8 @@ def find_charge_curve(self, discharge):
self.log("Warn: Found incomplete battery {} curve (no data points), maybe try again when you have more data.".format(curve_type))
else:
self.log(
"Warn: Cannot find battery {} curve (no full rate {} curve found for battery to {}), one of the required settings for {}, {}_rate, battery_power and predbat.status do not have history, check apps.yaml".format(
curve_type, curve_type, curve_label, soc_label, curve_type
"Info: Cannot find battery {} curve (no full rate {} cycle to {} found in history), battery may not have been fully charged/discharged at max rate recently - this is normal in cost-optimised operation".format(
curve_type, curve_type, curve_label
)
)
else:
Expand Down
19 changes: 11 additions & 8 deletions apps/predbat/output.py
Original file line number Diff line number Diff line change
Expand Up @@ -2872,16 +2872,19 @@ def calculate_yesterday(self):
cost_value += cost_yesterday
cost_yesterday_array[minute] = cost_value

# Get battery level yesterday
battery_today_data = self.get_history_wrapper(entity_id=self.prefix + ".soc_kw_h0", days=2, required=False)
if not battery_today_data:
self.log("Warn: Calculate yesterday: No soc_kw_h0 data for yesterday")
return
battery_data, _ = minute_data(battery_today_data[0], 2, self.now_utc, "state", "last_updated", backwards=True, clean_increment=False, smoothing=False, divide_by=1.0, scale=1.0)
battery_soc_yesterday = battery_data.get(minutes_back, 0.0)
# Get battery level yesterday - prefer the already-fetched in-memory history, fall back to HA query
battery_data = self.soc_kwh_history
if not battery_data:
battery_today_data = self.get_history_wrapper(entity_id=self.prefix + ".soc_kw_h0", days=2, required=False)
if battery_today_data:
battery_data, _ = minute_data(battery_today_data[0], 2, self.now_utc, "state", "last_updated", backwards=True, clean_increment=False, smoothing=False, divide_by=1.0, scale=1.0)
else:
self.log("Info: Calculate yesterday: No soc_kw_h0 history available, using saved soc value as fallback")
battery_data = {}
battery_soc_yesterday = battery_data.get(minutes_back, soc_yesterday)
battery_soc_yesterday_array = {}
for minute in range(0, end_record + self.minutes_now):
battery_soc_yesterday_array[minute] = battery_data.get(minutes_back + 24 * 60 - minute - 5, 0.0) # -5 gives 4 minutes into new data to allow for reset
battery_soc_yesterday_array[minute] = battery_data.get(minutes_back + 24 * 60 - minute - 5, soc_yesterday) # -5 gives 4 minutes into new data to allow for reset

# Get status history
predbat_status_data = self.get_history_wrapper(entity_id=self.prefix + ".status", days=2, required=False)
Expand Down
89 changes: 89 additions & 0 deletions apps/predbat/tests/test_calculate_yesterday.py
Original file line number Diff line number Diff line change
Expand Up @@ -964,6 +964,94 @@ def _mock_run_prediction(self_pb, charge_limit, charge_window, export_window, ex
return failed


def _make_counting_run_pred(counter_list):
"""Factory: create a run_prediction mock that increments counter_list[0] on each call."""

def _mock(self_pb, *args, **kwargs):
"""Mock run_prediction that counts invocations and clears soc arrays."""
counter_list[0] += 1
self_pb.predict_soc = {}
self_pb.predict_soc_best = {}
self_pb.predict_metric_best = {}
return (500, 1.0, 5.0, 0.5, 3.0, 5.0, 120, 2.0, 0, 0.0, 0)

return _mock


def _test_soc_kw_h0_fallback(my_predbat, failed):
"""Test: calculate_yesterday does NOT return early when soc_kw_h0 HA history is
unavailable; it falls back to soc_kwh_history (in-memory) if populated, and to
soc_yesterday (from savings_total_soc) when both sources are absent.

Covers the regression introduced in v8.39.4 where missing soc_kw_h0 HA history
(possible in cost-optimised operation or after a fresh install) caused the entire
calculate_yesterday run to be skipped every cycle, leaving savings_* entities
stuck in 'unavailable'.
"""
print("calculate_yesterday: Test – soc_kw_h0 fallback when HA history unavailable")
now_utc = _setup_base(my_predbat)
prefix = my_predbat.prefix

# History mock that provides cost_today but NOT soc_kw_h0
def _history_no_soc(entity_id, days=30, required=True, tracked=True):
if entity_id == prefix + ".cost_today":
return _make_constant_history(100.0, now_utc)
return None # soc_kw_h0 returns None

# --- sub-case A: soc_kwh_history empty, soc_kw_h0 HA history missing ---
# Ensure soc_kwh_history is empty (the default after reset)
my_predbat.soc_kwh_history = {}

my_predbat.get_history_wrapper = _history_no_soc
my_predbat.step_data_history = _make_mock_step_data(my_predbat.pv_today)
my_predbat.plan_write_debug = lambda *a, **kw: ("", "{}")
my_predbat.publish_html_plan = lambda *a, **kw: ("", "{}")
original_run_pred = my_predbat.run_prediction

ran_count = [0]
my_predbat.run_prediction = lambda *a, **kw: _make_counting_run_pred(ran_count)(my_predbat, *a, **kw)

my_predbat.calculate_yesterday()

if ran_count[0] == 0:
print("ERROR: calculate_yesterday returned early (ran_count=0); should continue with fallback soc values")
failed = True
else:
print("Sub-case A passed: calculate_yesterday ran {} prediction(s) despite missing soc_kw_h0 HA history".format(ran_count[0]))

_restore_methods(my_predbat, original_run_pred)
my_predbat.savings_last_updated = None

# --- sub-case B: soc_kwh_history populated, soc_kw_h0 HA history missing ---
# Populate soc_kwh_history directly (as fetch.py would have done)
minutes_back = my_predbat.minutes_now + 1
my_predbat.soc_kwh_history = {minutes_back: 7.5}

my_predbat.get_history_wrapper = _history_no_soc # still returns None for soc_kw_h0
my_predbat.step_data_history = _make_mock_step_data(my_predbat.pv_today)
my_predbat.plan_write_debug = lambda *a, **kw: ("", "{}")
my_predbat.publish_html_plan = lambda *a, **kw: ("", "{}")
original_run_pred = my_predbat.run_prediction
ran_count_b = [0]
my_predbat.run_prediction = lambda *a, **kw: _make_counting_run_pred(ran_count_b)(my_predbat, *a, **kw)

my_predbat.calculate_yesterday()

if ran_count_b[0] == 0:
print("ERROR: calculate_yesterday returned early in sub-case B (ran_count=0)")
failed = True
elif not my_predbat.savings_last_updated:
print("ERROR: sub-case B: savings_last_updated was not set after calculate_yesterday")
failed = True
else:
print("Sub-case B passed: calculate_yesterday ran {} prediction(s) using in-memory soc_kwh_history".format(ran_count_b[0]))

_restore_methods(my_predbat, original_run_pred)
my_predbat.savings_last_updated = None
my_predbat.soc_kwh_history = {}
return failed


# ---------------------------------------------------------------------------
# Entry point registered in TEST_REGISTRY
# ---------------------------------------------------------------------------
Expand All @@ -984,5 +1072,6 @@ def test_calculate_yesterday(my_predbat):
failed = _test_early_exit_respects_day_rollover(my_predbat, failed)
failed = _test_reconstruct_car_slots(my_predbat, failed)
failed = _test_soc_not_mutated_and_override_passed(my_predbat, failed)
failed = _test_soc_kw_h0_fallback(my_predbat, failed)

return failed
Loading