Skip to content
Merged
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
78 changes: 40 additions & 38 deletions apps/predbat/inverter.py
Original file line number Diff line number Diff line change
Expand Up @@ -689,51 +689,53 @@ def find_battery_size(self, nominal_capacity=0):
if (soc_percent_sensor or soc_kw_sensor) and battery_power_sensor:
if soc_percent_sensor:
soc_percent_data = self.base.get_history_wrapper(entity_id=soc_percent_sensor, days=self.base.max_days_previous, required=False)
if soc_percent_data:
soc_percent, _ = minute_data(
soc_percent_data[0],
self.base.max_days_previous,
self.base.now_utc,
"state",
"last_updated",
backwards=True,
clean_increment=False,
smoothing=False,
divide_by=1.0,
scale=self.battery_scaling,
required_unit="%",
)
else:
soc_percent = {}
else:
soc_kw_data = self.base.get_history_wrapper(entity_id=soc_kw_sensor, days=self.base.max_days_previous, required=False)
# Compute soc_percent_data from soc_kw, we use find the maximum value in the history to assume that's soc_max
if nominal_capacity and nominal_capacity > 0:
soc_max = nominal_capacity
else:
soc_values = []
if soc_kw_data and len(soc_kw_data) > 0:
for state in soc_kw_data[0]:
try:
soc_values.append(float(dp0(float(state["state"]))))
except (ValueError, TypeError):
pass
soc_max = max(soc_values) if soc_values else None
if soc_max and soc_max > 0:
built_list = []
for state in soc_kw_data[0]:
try:
kw = float(state["state"])
percent = (kw / soc_max) * 100.0
built_list.append({"state": dp2(percent), "last_updated": state["last_updated"], "attributes": {"unit_of_measurement": "%"}})
except (ValueError, TypeError, KeyError):
continue
soc_percent_data = [built_list] # wrap to match get_history_wrapper format
else:
soc_percent_data = None
soc_percent = {}
if soc_kw_data:
# Parse kWh history into a clean minute dict then convert to percent
soc_kw_minute, _ = minute_data(
soc_kw_data[0],
self.base.max_days_previous,
self.base.now_utc,
"state",
"last_updated",
backwards=True,
clean_increment=False,
smoothing=False,
divide_by=1.0,
scale=1.0,
required_unit="kWh",
)
Comment on lines +713 to +725
# Determine soc_max from nominal_capacity or the observed maximum
if nominal_capacity and nominal_capacity > 0:
soc_max = nominal_capacity
else:
soc_max = max(soc_kw_minute.values()) if soc_kw_minute else 0
if soc_max > 0:
soc_percent = {minute: (kw / soc_max) * 100.0 for minute, kw in soc_kw_minute.items()}
battery_power_data = self.base.get_history_wrapper(entity_id=battery_power_sensor, days=self.base.max_days_previous, required=False)

if not soc_percent_data or not battery_power_data:
if not soc_percent or not battery_power_data:
self.log("Warn: Unable to estimate battery size - no history data available")
return None

soc_percent, _ = minute_data(
soc_percent_data[0],
self.base.max_days_previous,
self.base.now_utc,
"state",
"last_updated",
backwards=True,
clean_increment=False,
smoothing=False,
divide_by=1.0,
scale=self.battery_scaling,
required_unit="%",
)
battery_power, _ = minute_data(
battery_power_data[0],
self.base.max_days_previous,
Expand Down
21 changes: 14 additions & 7 deletions apps/predbat/tests/test_find_battery_size.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,9 @@ def mock_get_history(entity_id, now=None, days=30):
def create_test_history_data_soc_kw(my_predbat, num_days=2, battery_size_kwh=10.0):
"""
Like create_test_history_data but exposes soc_kw (kWh absolute) instead of soc_percent.
The find_battery_size code must detect the maximum kWh value in the history and derive
percentages from it.
The find_battery_size code infers soc_max from the observed maximum kWh value, so the
charging cycles must reach 100% (battery_size_kwh) at least once so that soc_max is
accurately calibrated and percentage-based estimates are correct.
"""
ha = my_predbat.ha_interface
base_time = my_predbat.midnight_utc - timedelta(days=num_days)
Expand All @@ -111,25 +112,31 @@ def create_test_history_data_soc_kw(my_predbat, num_days=2, battery_size_kwh=10.
max_power_w = 2600
charge_power_w = max_power_w * 0.94 # above 90% threshold

current_soc_kwh = battery_size_kwh * 0.20 # Start at 20%
# Start near-empty so the 5-hour charge window reaches 100%, giving an accurate
# observed soc_max for percentage derivation.
current_soc_kwh = battery_size_kwh * 0.05 # Start at 5%

for minutes in range(0, total_minutes, 5):
timestamp = base_time + timedelta(minutes=minutes)
timestamp_str = timestamp.strftime("%Y-%m-%dT%H:%M:%S%z")
hour = timestamp.hour

if 2 <= hour < 4:
# Morning charge: 00:00–05:00 (5 hours). At 2444 W that adds ~12.2 kWh,
# which is more than enough to top up a 10 kWh battery from 5%.
if 0 <= hour < 5:
battery_power = -charge_power_w
energy_added_kwh = charge_power_w * 5 / 60.0 / 1000.0
current_soc_kwh = min(battery_size_kwh, current_soc_kwh + energy_added_kwh)
elif 18 <= hour < 20:
# Evening charge: 18:00–23:00 (same logic)
elif 18 <= hour < 23:
battery_power = -charge_power_w
energy_added_kwh = charge_power_w * 5 / 60.0 / 1000.0
current_soc_kwh = min(battery_size_kwh, current_soc_kwh + energy_added_kwh)
else:
battery_power = 0
if hour == 1 or hour == 17:
current_soc_kwh = battery_size_kwh * 0.20
# Reset to near-empty before each charge session
if hour == 17:
current_soc_kwh = battery_size_kwh * 0.05
Comment on lines 135 to +139

history_dict["sensor.soc_kw"].append({"state": str(round(current_soc_kwh, 3)), "last_updated": timestamp_str, "attributes": {"unit_of_measurement": "kWh"}})
history_dict["sensor.battery_power"].append({"state": round(battery_power, 1), "last_updated": timestamp_str, "attributes": {"unit_of_measurement": "W"}})
Expand Down
Loading