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
1 change: 1 addition & 0 deletions apps/predbat/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -1609,6 +1609,7 @@
"can_span_midnight": False,
"charge_discharge_with_rate": False,
"target_soc_used_for_discharge": False,
"clear_slot_on_disable": True,
},
"SE": {
"name": "SolarEdge",
Expand Down
67 changes: 63 additions & 4 deletions apps/predbat/inverter.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,7 @@ def __init__(self, base, id=0, quiet=False, rest_postCommand=None, rest_getData=
self.inv_has_ge_inverter_mode = INVERTER_DEF[self.inverter_type]["has_ge_inverter_mode"]
self.inv_has_ge_eco_toggle = INVERTER_DEF[self.inverter_type].get("has_ge_eco_toggle", False)
self.inv_num_load_entities = INVERTER_DEF[self.inverter_type]["num_load_entities"]
self.inv_clear_slot_on_disable = INVERTER_DEF[self.inverter_type].get("clear_slot_on_disable", False)
self.inv_write_and_poll_sleep = INVERTER_DEF[self.inverter_type]["write_and_poll_sleep"]
self.inv_has_idle_time = INVERTER_DEF[self.inverter_type]["has_idle_time"]
self.inv_can_span_midnight = INVERTER_DEF[self.inverter_type]["can_span_midnight"]
Expand Down Expand Up @@ -2275,6 +2276,7 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No
# Turn off scheduled discharge
if not force_export and old_discharge_enable:
self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), False)
self.clear_timed_slot_values("discharge")
self.log("Inverter {} Turning off scheduled export".format(self.id))

self.base.log("Inverter {} Adjust force export to {}, change times from {} - {} to {} - {}".format(self.id, force_export, old_start, old_end, new_start, new_end))
Expand Down Expand Up @@ -2382,16 +2384,18 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No
self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), True)
self.log("Inverter {} Turning on scheduled export".format(self.id))

# Set discharge current before the button press so the inverter registers it when the schedule is committed
if force_export and (not self.inv_has_charge_enable_time or self.inv_clear_slot_on_disable) and (self.inv_output_charge_control == "current"):
if self.inv_charge_control_immediate:
self.enable_charge_discharge_with_time_current("discharge", True)

if (new_end != old_end) or (new_start != old_start) or (force_export != old_discharge_enable):
if self.inv_time_button_press:
self.press_and_poll_button()

# Force export, turn it on after we change the window
if force_export:
self.adjust_inverter_mode(force_export, changed_start_end=changed_start_end)
if not self.inv_has_charge_enable_time and (self.inv_output_charge_control == "current"):
if self.inv_charge_control_immediate:
self.enable_charge_discharge_with_time_current("discharge", True)

# Notify
if changed_start_end:
Expand Down Expand Up @@ -2438,6 +2442,8 @@ def disable_charge_window(self, notify=True):
self.rest_enableChargeSchedule(False)
else:
self.write_and_poll_switch("scheduled_charge_enable", self.base.get_arg("scheduled_charge_enable", indirect=False, index=self.id), False)
if self.inv_has_charge_enable_time:
self.clear_timed_slot_values("charge")
# If there's no charge enable switch then we can enable using start and end time
if not self.inv_has_charge_enable_time and (self.inv_output_charge_control == "current"):
if self.inv_charge_control_immediate:
Expand Down Expand Up @@ -2544,6 +2550,59 @@ def enable_charge_discharge_with_time_current(self, direction, enable):
else:
self.set_current_from_power(direction, 0)

def clear_timed_slot_values(self, direction):
"""
Clear timed slot values for a direction after disabling a schedule.

This avoids stale overlapping times on integrations that validate overlap
even when a slot is disabled. Does nothing if inv_clear_slot_on_disable is False.
"""
if not self.inv_clear_slot_on_disable:
self.log(f"Inverter {self.id} clear_timed_slot_values: skipped (clear_slot_on_disable not set for this inverter type)")
return

if direction not in ["charge", "discharge"]:
self.log(f"Warn: Inverter {self.id} clear_timed_slot_values: invalid direction '{direction}' (must be 'charge' or 'discharge')")
return

self.log(f"Inverter {self.id} Clearing {direction} slot values to prevent overlap validation errors")

failed_writes = []

if self.inv_charge_time_format == "H M":
self.log(f"Inverter {self.id} Detected time format: H M (separate hour/minute entities)")
for edge in ["start", "end"]:
for field in ["hour", "minute"]:
name = f"{direction}_{edge}_{field}"
entity_id = self.base.get_arg(name, indirect=False, index=self.id)
if not entity_id:
continue

if isinstance(entity_id, str) and entity_id.startswith("time."):
if not self.write_and_poll_option(name, entity_id, "00:00:00"):
failed_writes.append(f"{name} ({entity_id})")
else:
if not self.write_and_poll_value(name, entity_id, 0):
failed_writes.append(f"{name} ({entity_id})")
elif self.inv_charge_time_format == "H:M-H:M":
self.log(f"Inverter {self.id} Detected time format: H:M-H:M (range entity)")
name = f"{direction}_time"
entity_id = self.base.get_arg(name, indirect=False, index=self.id)
if entity_id:
if not self.write_and_poll_value(name, entity_id, "00:00-00:00"):
failed_writes.append(f"{name} ({entity_id})")
else:
self.log(f"Inverter {self.id} Detected time format: {self.inv_charge_time_format} (no slot clearing for this format)")

current_name = f"timed_{direction}_current"
current_entity_id = self.base.get_arg(current_name, indirect=False, index=self.id)
if current_entity_id:
if not self.write_and_poll_value(current_name, current_entity_id, 0, fuzzy=1):
failed_writes.append(f"{current_name} ({current_entity_id})")

if failed_writes:
self.base.record_status(f"Warn: Inverter {self.id} failed to clear {len(failed_writes)} slot values: {', '.join(failed_writes)}", had_errors=True)

def set_current_from_power(self, direction, power):
"""
Set the timed charge/discharge current setting by converting power to current
Expand Down Expand Up @@ -2816,7 +2875,7 @@ def adjust_charge_window(self, charge_start_time, charge_end_time, minutes_now):
self.rest_enableChargeSchedule(True)
elif "scheduled_charge_enable" in self.base.args:
self.write_and_poll_switch("scheduled_charge_enable", self.base.get_arg("scheduled_charge_enable", indirect=False, index=self.id), True)
if not self.inv_has_charge_enable_time and (self.inv_output_charge_control == "current"):
if (not self.inv_has_charge_enable_time or self.inv_clear_slot_on_disable) and (self.inv_output_charge_control == "current"):
if self.inv_charge_control_immediate:
self.enable_charge_discharge_with_time_current("charge", True)
else:
Expand Down
185 changes: 185 additions & 0 deletions apps/predbat/tests/test_inverter.py
Original file line number Diff line number Diff line change
Expand Up @@ -1560,6 +1560,179 @@ def test_time_entity_hour_write(test_name, ha, inv, dummy_rest, direction, new_s
return failed


def test_disable_charge_window_clears_slot_values_hm(test_name, ha, inv):
"""
Ensure disable_charge_window clears H M slot values and timed current when
the inverter has explicit schedule enable entities (e.g. GS_fb00 style setup).
"""
failed = False
print("Test: {}".format(test_name))

inv.rest_data = None
inv.inv_charge_time_format = "H M"
inv.inv_has_charge_enable_time = True
inv.inv_clear_slot_on_disable = True
inv.inv_time_button_press = False

ha.dummy_items["switch.scheduled_charge_enable"] = "on"
ha.dummy_items["number.charge_start_hour"] = 5
ha.dummy_items["number.charge_start_minute"] = 30
ha.dummy_items["number.charge_end_hour"] = 7
ha.dummy_items["number.charge_end_minute"] = 45
ha.dummy_items["number.timed_charge_current"] = 18

inv.base.args["scheduled_charge_enable"] = "switch.scheduled_charge_enable"
inv.base.args["charge_start_hour"] = "number.charge_start_hour"
inv.base.args["charge_start_minute"] = "number.charge_start_minute"
inv.base.args["charge_end_hour"] = "number.charge_end_hour"
inv.base.args["charge_end_minute"] = "number.charge_end_minute"
inv.base.args["timed_charge_current"] = "number.timed_charge_current"

inv.disable_charge_window()

if ha.get_state("switch.scheduled_charge_enable") != "off":
print("ERROR: scheduled_charge_enable should be off got {}".format(ha.get_state("switch.scheduled_charge_enable")))
failed = True
if ha.get_state("number.charge_start_hour") != 0:
print("ERROR: charge_start_hour should be 0 got {}".format(ha.get_state("number.charge_start_hour")))
failed = True
if ha.get_state("number.charge_start_minute") != 0:
print("ERROR: charge_start_minute should be 0 got {}".format(ha.get_state("number.charge_start_minute")))
failed = True
if ha.get_state("number.charge_end_hour") != 0:
print("ERROR: charge_end_hour should be 0 got {}".format(ha.get_state("number.charge_end_hour")))
failed = True
if ha.get_state("number.charge_end_minute") != 0:
print("ERROR: charge_end_minute should be 0 got {}".format(ha.get_state("number.charge_end_minute")))
failed = True
if ha.get_state("number.timed_charge_current") != 0:
print("ERROR: timed_charge_current should be 0 got {}".format(ha.get_state("number.timed_charge_current")))
failed = True

return failed


def test_disable_discharge_clears_slot_values_time_entities(test_name, ha, inv):
"""
Ensure discharge disable path clears slot values and timed current, including
writing 00:00:00 for time.* entities.
"""
failed = False
print("Test: {}".format(test_name))

inv.rest_data = None
inv.inv_charge_time_format = "H M"
inv.inv_has_discharge_enable_time = True
inv.inv_has_ge_inverter_mode = False
inv.inv_has_charge_enable_time = True
inv.inv_clear_slot_on_disable = True
inv.inv_time_button_press = False

old_start = "10:15:00"
old_end = "11:45:00"

ha.dummy_items["switch.scheduled_discharge_enable"] = "on"
ha.dummy_items["select.discharge_start_time"] = old_start
ha.dummy_items["select.discharge_end_time"] = old_end
ha.dummy_items["time.discharge_start_hour"] = old_start
ha.dummy_items["number.discharge_start_minute"] = 15
ha.dummy_items["time.discharge_end_hour"] = old_end
ha.dummy_items["number.discharge_end_minute"] = 45
ha.dummy_items["number.timed_discharge_current"] = 21
ha.dummy_items["select.inverter_mode"] = "Timed Export"

inv.base.args["scheduled_discharge_enable"] = "switch.scheduled_discharge_enable"
inv.base.args["discharge_start_time"] = "select.discharge_start_time"
inv.base.args["discharge_end_time"] = "select.discharge_end_time"
inv.base.args["discharge_start_hour"] = "time.discharge_start_hour"
inv.base.args["discharge_start_minute"] = "number.discharge_start_minute"
inv.base.args["discharge_end_hour"] = "time.discharge_end_hour"
inv.base.args["discharge_end_minute"] = "number.discharge_end_minute"
inv.base.args["timed_discharge_current"] = "number.timed_discharge_current"
inv.base.args["inverter_mode"] = "select.inverter_mode"

inv.adjust_force_export(False, datetime.strptime(old_start, "%H:%M:%S"), datetime.strptime(old_end, "%H:%M:%S"))

if ha.get_state("switch.scheduled_discharge_enable") != "off":
print("ERROR: scheduled_discharge_enable should be off got {}".format(ha.get_state("switch.scheduled_discharge_enable")))
failed = True
if ha.get_state("time.discharge_start_hour") != "00:00:00":
print("ERROR: discharge_start_hour time entity should be 00:00:00 got {}".format(ha.get_state("time.discharge_start_hour")))
failed = True
if ha.get_state("number.discharge_start_minute") != 0:
print("ERROR: discharge_start_minute should be 0 got {}".format(ha.get_state("number.discharge_start_minute")))
failed = True
if ha.get_state("time.discharge_end_hour") != "00:00:00":
print("ERROR: discharge_end_hour time entity should be 00:00:00 got {}".format(ha.get_state("time.discharge_end_hour")))
failed = True
if ha.get_state("number.discharge_end_minute") != 0:
print("ERROR: discharge_end_minute should be 0 got {}".format(ha.get_state("number.discharge_end_minute")))
failed = True
if ha.get_state("number.timed_discharge_current") != 0:
print("ERROR: timed_discharge_current should be 0 got {}".format(ha.get_state("number.timed_discharge_current")))
failed = True

return failed


def test_enable_discharge_restores_current_for_clear_slot_inverters(test_name, ha, inv):
"""
Ensure force export enable restores timed_discharge_current for inverters
that clear slot values on disable (e.g. GS_fb00).
"""
failed = False
print("Test: {}".format(test_name))

inv.rest_data = None
inv.inv_charge_time_format = "H M"
inv.inv_has_discharge_enable_time = True
inv.inv_has_charge_enable_time = True
inv.inv_has_ge_inverter_mode = False
inv.inv_output_charge_control = "current"
inv.inv_charge_control_immediate = True
inv.inv_clear_slot_on_disable = True
inv.inv_time_button_press = False

start = "12:00:00"
end = "12:30:00"

ha.dummy_items["switch.scheduled_discharge_enable"] = "off"
ha.dummy_items["select.discharge_start_time"] = "00:00:00"
ha.dummy_items["select.discharge_end_time"] = "00:00:00"
ha.dummy_items["number.discharge_start_hour"] = 0
ha.dummy_items["number.discharge_start_minute"] = 0
ha.dummy_items["number.discharge_end_hour"] = 0
ha.dummy_items["number.discharge_end_minute"] = 0
ha.dummy_items["number.timed_discharge_current"] = 0
ha.dummy_items["number.discharge_rate"] = 2600
ha.dummy_items["select.inverter_mode"] = "Eco"

inv.base.args["scheduled_discharge_enable"] = "switch.scheduled_discharge_enable"
inv.base.args["discharge_start_time"] = "select.discharge_start_time"
inv.base.args["discharge_end_time"] = "select.discharge_end_time"
inv.base.args["discharge_start_hour"] = "number.discharge_start_hour"
inv.base.args["discharge_start_minute"] = "number.discharge_start_minute"
inv.base.args["discharge_end_hour"] = "number.discharge_end_hour"
inv.base.args["discharge_end_minute"] = "number.discharge_end_minute"
inv.base.args["timed_discharge_current"] = "number.timed_discharge_current"
inv.base.args["discharge_rate"] = "number.discharge_rate"
inv.base.args["inverter_mode"] = "select.inverter_mode"

inv.adjust_force_export(True, datetime.strptime(start, "%H:%M:%S"), datetime.strptime(end, "%H:%M:%S"))

expected_current = round(float(ha.get_state("number.discharge_rate")) / inv.battery_voltage, inv.inv_current_dp)
actual_current = ha.get_state("number.timed_discharge_current")

if ha.get_state("switch.scheduled_discharge_enable") != "on":
print("ERROR: scheduled_discharge_enable should be on got {}".format(ha.get_state("switch.scheduled_discharge_enable")))
failed = True
if actual_current != expected_current:
print("ERROR: timed_discharge_current should be {} got {}".format(expected_current, actual_current))
failed = True

return failed


def run_inverter_tests(my_predbat_dummy):
"""
Test the inverter functions
Expand Down Expand Up @@ -2273,5 +2446,17 @@ def run_inverter_tests(my_predbat_dummy):
if failed:
return failed

failed |= test_disable_charge_window_clears_slot_values_hm("disable_charge_clear_values_hm", ha, inv)
if failed:
return failed

failed |= test_disable_discharge_clears_slot_values_time_entities("disable_discharge_clear_values_time_entities", ha, inv)
if failed:
return failed

failed |= test_enable_discharge_restores_current_for_clear_slot_inverters("enable_discharge_restores_current", ha, inv)
if failed:
return failed

failed |= test_inverter_self_test("self_test1", my_predbat)
return failed
8 changes: 8 additions & 0 deletions docs/inverter-setup.md
Original file line number Diff line number Diff line change
Expand Up @@ -3110,4 +3110,12 @@ Defines the units of the SoC setting (currently not used), it defaults to "%".

Sets the number of seconds between polls of inverter settings.

### clear_slot_on_disable

When True, Predbat will clear the charge/discharge slot time and current values back to zero whenever a timed charge or discharge window is disabled on the inverter.
This prevents inverters that validate for overlapping time slots from rejecting writes when Predbat switches between charge and discharge modes.

This setting has currently only been tested with **GS_fb00** (Ginlong Solis FB00 firmware) inverters using the Solax Modbus integration in Home Assistant.
It defaults to False for all other inverter types.

<!-- markdownlint-enable MD046 -->
Loading