Skip to content
Draft
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
8 changes: 4 additions & 4 deletions apps/predbat/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -617,7 +617,7 @@
"icon": "mdi:ev-station",
"enable": "car_charging_manual_soc",
"default": 0.0,
"restore": False,
"restore": True,
},
{
"name": "car_charging_manual_soc_1",
Expand All @@ -638,7 +638,7 @@
"icon": "mdi:ev-station",
"enable": "car_charging_manual_soc_1",
"default": 0.0,
"restore": False,
"restore": True,
},
{
"name": "car_charging_manual_soc_2",
Expand All @@ -659,7 +659,7 @@
"icon": "mdi:ev-station",
"enable": "car_charging_manual_soc_2",
"default": 0.0,
"restore": False,
"restore": True,
},
{
"name": "car_charging_manual_soc_3",
Expand All @@ -680,7 +680,7 @@
"icon": "mdi:ev-station",
"enable": "car_charging_manual_soc_3",
"default": 0.0,
"restore": False,
"restore": True,
},
{
"name": "octopus_intelligent_charging",
Expand Down
8 changes: 7 additions & 1 deletion apps/predbat/fetch.py
Original file line number Diff line number Diff line change
Expand Up @@ -901,12 +901,18 @@ def fetch_sensor_data(self, save=True):
self.octopus_intelligent_charging = False

# Work out car SoC and reset next
# Store previous values to detect unexpected resets
prev_car_charging_soc = self.car_charging_soc[:] if hasattr(self, 'car_charging_soc') and self.car_charging_soc else []
self.car_charging_soc = [0.0 for car_n in range(self.num_cars)]
self.car_charging_soc_next = [None for car_n in range(self.num_cars)]
for car_n in range(self.num_cars):
if car_n < len(self.car_charging_manual_soc) and self.car_charging_manual_soc[car_n]:
car_postfix = "" if car_n == 0 else "_" + str(car_n)
self.car_charging_soc[car_n] = self.get_arg("car_charging_manual_soc_kwh" + car_postfix, 0.0)
manual_soc_kwh = self.get_arg("car_charging_manual_soc_kwh" + car_postfix, 0.0)
self.car_charging_soc[car_n] = manual_soc_kwh
# Log if manual SOC is unexpectedly low
if manual_soc_kwh < 0.1 and car_n < len(prev_car_charging_soc) and prev_car_charging_soc[car_n] > 1.0:
self.log("Warn: Car {} manual SOC has dropped from {:.2f} kWh to {:.2f} kWh - this may indicate the entity was reset".format(car_n, prev_car_charging_soc[car_n], manual_soc_kwh))
else:
self.car_charging_soc[car_n] = (self.get_arg("car_charging_soc", 0.0, index=car_n) * self.car_charging_battery_size[car_n]) / 100.0
if self.num_cars:
Expand Down
8 changes: 7 additions & 1 deletion apps/predbat/predbat.py
Original file line number Diff line number Diff line change
Expand Up @@ -948,7 +948,13 @@ def update_pred(self, scheduled=True):
car_postfix = "" if car_n == 0 else "_" + str(car_n)
self.log("Car {} charging Manual SoC current is {} next is {}".format(car_n, self.car_charging_soc[car_n], self.car_charging_soc_next[car_n]))
if self.car_charging_soc_next[car_n] is not None:
self.expose_config("car_charging_manual_soc_kwh" + car_postfix, dp3(self.car_charging_soc_next[car_n]))
# Prevent manual SOC from being incorrectly reset to zero
# Only update if the new value is reasonable (allows small decreases for rounding)
if self.car_charging_soc_next[car_n] < 0.1 and self.car_charging_soc[car_n] > 1.0:
self.log("Warn: Car {} manual SOC would be reset from {:.2f} kWh to {:.2f} kWh - skipping update to prevent data loss".format(
car_n, self.car_charging_soc[car_n], self.car_charging_soc_next[car_n]))
else:
self.expose_config("car_charging_manual_soc_kwh" + car_postfix, dp3(self.car_charging_soc_next[car_n]))

# Holiday days left countdown, subtract a day at midnight every day
if scheduled and self.holiday_days_left > 0 and self.minutes_now < RUN_EVERY:
Expand Down
89 changes: 89 additions & 0 deletions apps/predbat/tests/test_car_charging_manual_soc.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
# -----------------------------------------------------------------------------
# Predbat Home Battery System
# Copyright Trefor Southwell 2024 - 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 tests.test_infra import reset_rates2, reset_inverter


def run_car_charging_manual_soc_test(my_predbat):
"""
Test car charging manual SOC protection against unexpected resets
"""
failed = False

print("**** Running Car Charging Manual SOC tests ****")

# Initialize car charging setup
my_predbat.num_cars = 1
my_predbat.car_charging_battery_size = [75.0] # 75 kWh battery
my_predbat.car_charging_limit = [70.0] # Charge to 70 kWh
my_predbat.car_charging_soc = [30.0] # Currently at 30 kWh
my_predbat.car_charging_soc_next = [None]
my_predbat.car_charging_manual_soc = [True] # Manual SOC mode enabled
my_predbat.car_charging_rate = [7.4] # 7.4 kW charging rate
my_predbat.car_charging_loss = 1.0 # No loss for simplicity
my_predbat.car_charging_plan_max_price = [99.0]
my_predbat.car_charging_plan_smart = [True]
my_predbat.car_charging_plan_time = ["07:00:00"]
my_predbat.car_charging_slots = [[]]

# Test 1: Verify manual SOC is preserved when value is sensible
print("Test 1: Manual SOC should be updated when soc_next is reasonable")
my_predbat.car_charging_soc[0] = 30.0
my_predbat.car_charging_soc_next[0] = 35.0 # Increased after charging

# Simulate the update check (from predbat.py line 944-951)
car_n = 0
if my_predbat.car_charging_soc_next[car_n] is not None:
# Check protection logic
if my_predbat.car_charging_soc_next[car_n] < 0.1 and my_predbat.car_charging_soc[car_n] > 1.0:
print("ERROR: Test 1 failed - protection blocked valid update")
failed = True
else:
print("PASS: Test 1 - Manual SOC would be updated from {:.2f} to {:.2f}".format(
my_predbat.car_charging_soc[car_n], my_predbat.car_charging_soc_next[car_n]))

# Test 2: Verify protection prevents reset from high value to zero
print("Test 2: Manual SOC should NOT be reset from high value to near-zero")
my_predbat.car_charging_soc[0] = 30.0
my_predbat.car_charging_soc_next[0] = 0.05 # Unexpectedly low value

if my_predbat.car_charging_soc_next[car_n] is not None:
if my_predbat.car_charging_soc_next[car_n] < 0.1 and my_predbat.car_charging_soc[car_n] > 1.0:
print("PASS: Test 2 - Protection correctly prevented reset from {:.2f} to {:.2f}".format(
my_predbat.car_charging_soc[car_n], my_predbat.car_charging_soc_next[car_n]))
else:
print("ERROR: Test 2 failed - protection did NOT block invalid reset")
failed = True

# Test 3: Verify zero to zero is allowed (initial state)
print("Test 3: Manual SOC at zero can stay at zero")
my_predbat.car_charging_soc[0] = 0.0
my_predbat.car_charging_soc_next[0] = 0.0

if my_predbat.car_charging_soc_next[car_n] is not None:
if my_predbat.car_charging_soc_next[car_n] < 0.1 and my_predbat.car_charging_soc[car_n] > 1.0:
print("ERROR: Test 3 failed - protection incorrectly blocked zero to zero")
failed = True
else:
print("PASS: Test 3 - Zero to zero transition allowed")

# Test 4: Verify small decreases are allowed (within threshold)
print("Test 4: Small manual SOC values (< 1 kWh) can be updated")
my_predbat.car_charging_soc[0] = 0.5
my_predbat.car_charging_soc_next[0] = 0.05

if my_predbat.car_charging_soc_next[car_n] is not None:
if my_predbat.car_charging_soc_next[car_n] < 0.1 and my_predbat.car_charging_soc[car_n] > 1.0:
print("ERROR: Test 4 failed - protection incorrectly blocked small value update")
failed = True
else:
print("PASS: Test 4 - Small values can be updated")

return failed
2 changes: 2 additions & 0 deletions apps/predbat/unit_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
from tests.test_optimise_all_windows import run_optimise_all_windows_tests
from tests.test_nordpool import run_nordpool_test
from tests.test_car_charging_smart import run_car_charging_smart_tests
from tests.test_car_charging_manual_soc import run_car_charging_manual_soc_test
from tests.test_plugin_startup import test_plugin_startup_order
from tests.test_optimise_levels import run_optimise_levels_tests
from tests.test_energydataservice import test_energydataservice
Expand Down Expand Up @@ -207,6 +208,7 @@ def main():
("solax", run_solax_tests, "SolaX API tests", False),
("iboost_smart", run_iboost_smart_tests, "iBoost smart tests", False),
("car_charging_smart", run_car_charging_smart_tests, "Car charging smart tests", False),
("car_charging_manual_soc", run_car_charging_manual_soc_test, "Car charging manual SOC protection tests", False),
("intersect_window", run_intersect_window_tests, "Intersect window tests", False),
("inverter_multi", run_inverter_multi_tests, "Inverter multi tests", False),
("octopus_free", test_octopus_free, "Octopus free electricity tests", False),
Expand Down