fix(deye): apply the SOC floor only at the API boundary

Copilot review on #4363 spotted that publish_schedule_settings_ha()
clamped the published reserve while apply_reserve_live() stored the raw
value, so the two could disagree.

Resolved the other way round from the suggestion. Clamping at entry as
well would make the internal state self-consistent but would not fix the
problem described: Predbat writes this entity and then reads it back to
confirm (write_and_poll_value), so if it writes 4 and we publish 14 the
read-back cannot match whichever end is clamped, and it retries until it
gives up — the exact "didn't complete got 0.0" failure this component
was just fixed for.

The entity is Predbat's control surface, so it now echoes what Predbat
wrote, everywhere: the entity read, the control event and the published
state all store and emit the raw value. The floor is applied in one
place, build_dynamic_payload, immediately before the payload leaves for
the API — which is what actually protects the battery, and cannot be
bypassed by any caller.

Narrow in practice: adjust_reserve already clamps to reserve_percent,
which comes from the battery_min_soc sensor this component publishes, so
the automatic path can never write below the floor. A manually configured
install that does not map battery_min_soc defaults it to 4.0 and could.

The clamp now logs once per serial when it actually lifts a value.
Routine before Predbat has written the reserve; a persistent complaint
means Predbat is planning below the inverter's floor, which is worth
seeing rather than silently correcting.

Adds a test that the reserve entity echoes a below-floor write verbatim
while the payload built from it is still lifted to the floor. Verified to
fail against the clamped-publish version: "expected the entity to echo 4,
got 14".

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: springfall2008

apps/predbat/deye.py

@@ -129,6 +129,7 @@ def initialize(
         self._tier_refreshed = {}
         self._cache_restored = False
         self._saved_ratings = None  # signature of the ratings last written, to skip no-op saves
+        self._soc_floor_warned = set()
         self.battery_nominal_voltage = self._as_float(battery_nominal_voltage, 0.0)
         # auth_method defaults to app_credentials, but an injected access token with no
         # developer app credentials can only be oauth — and in app_credentials mode
@@ -639,9 +640,17 @@ def build_dynamic_payload(self, sn, schedule, current_soc, now_minutes=None):
         # 14%. Applied here, at the last step, so no other caller can bypass it.
         floor = self.battery_reserve_min(sn)
         if floor > 0:
+            lifted = False
             for slot in slots:
                 if slot.get(TOU_FIELD["soc"], 0) < floor:
                     slot[TOU_FIELD["soc"]] = floor
+                    lifted = True
+            # Say so once per serial. Routine before Predbat has written the reserve, but a
+            # persistent complaint means Predbat is planning below the inverter's floor —
+            # usually battery_min_soc not being mapped, which is worth surfacing.
+            if lifted and sn not in self._soc_floor_warned:
+                self._soc_floor_warned.add(sn)
+                self.log(f"Info: DEYE {sn} raising requested slot SOC to the inverter's {floor}% floor (config/battery battLowCapacity)")
         return {
             "deviceSn": sn,
             "workMode": active["work_mode"],
@@ -802,15 +811,16 @@ async def publish_data(self):
     async def publish_schedule_settings_ha(self, sn):
         """Publish the charge/export schedule control entities for one inverter."""
         local = self.local_schedule.get(sn, {})
-        # The entity starts life at 0 and only reaches a real value once Predbat writes it,
-        # so surface the inverter's own floor as both the published value and the entity
-        # minimum rather than advertising a reserve the hardware would never honour.
-        floor = self.battery_reserve_min(sn)
-        reserve = max(int(local.get("reserve", 0)), floor)
+        # Deliberately NOT clamped to the inverter floor. This entity is Predbat's control
+        # surface: it writes a value then reads it back to confirm (write_and_poll_value),
+        # so publishing anything other than what was written guarantees a mismatch and a
+        # retry storm. The floor is enforced at the API boundary instead, in
+        # build_dynamic_payload, which is what actually protects the battery.
+        reserve = int(local.get("reserve", 0))
         self.dashboard_item(
             self._control_name("number", sn, "battery_schedule_reserve"),
             state=reserve,
-            attributes={"min": floor, "max": 100, "step": 1, "unit_of_measurement": "%", "friendly_name": f"DEYE {sn} Battery Schedule Reserve", "icon": "mdi:gauge"},
+            attributes={"min": 0, "max": 100, "step": 1, "unit_of_measurement": "%", "friendly_name": f"DEYE {sn} Battery Schedule Reserve", "icon": "mdi:gauge"},
             app="deye",
         )
         for direction in ("charge", "export"):
@@ -842,10 +852,7 @@ async def get_schedule_settings_ha(self, sn):
         before Predbat republishes - falls back to 0 rather than raising and crashing the
         reconciliation loop (mirrors the Fox defensiveness).
         """
-        # Clamped to the inverter's own floor, exactly as fox.py does with fdsoc_min: the
-        # entity reads 0 until Predbat has written it, and 0 is below what the hardware
-        # will honour.
-        schedule = {"reserve": max(int(self._as_float(self.get_state_wrapper(self._control_name("number", sn, "battery_schedule_reserve"), default=0), 0)), self.battery_reserve_min(sn))}
+        schedule = {"reserve": int(self._as_float(self.get_state_wrapper(self._control_name("number", sn, "battery_schedule_reserve"), default=0), 0))}
         for direction in ("charge", "export"):
             schedule[direction] = {
                 "enable": self.get_state_wrapper(self._control_name("switch", sn, f"battery_schedule_{direction}_enable"), default="off") == "on",
@@ -912,7 +919,9 @@ def update_local_schedule(self, sn, entity_id, value):
         """
         schedule = self.local_schedule.setdefault(sn, {})
         if entity_id.endswith("battery_schedule_reserve"):
-            schedule["reserve"] = max(int(self._as_float(value, 0)), self.battery_reserve_min(sn))
+            # Stored exactly as written; the floor is applied at the API boundary so the
+            # read-back Predbat performs always matches what it wrote.
+            schedule["reserve"] = int(self._as_float(value, 0))
             return True
         direction = "charge" if "_charge_" in entity_id else ("export" if "_export_" in entity_id else None)
         if not direction:

apps/predbat/tests/test_deye_api.py

@@ -50,6 +50,7 @@ def __init__(self, auth_method="app_credentials", data_center="eu", inverter_sn=
         self.cached_values = {}
         self._tier_refreshed = {}
         self._cache_restored = False
+        self._soc_floor_warned = set()
         self.log_messages = []
         self.local_tz = pytz.timezone("Europe/London")
         self.base = MagicMock()

apps/predbat/tests/test_deye_publish.py

@@ -498,6 +498,42 @@ async def fake_apply(sn, force=False):
     assert not failed, "test_write_button_applies_and_is_not_stored_as_schedule"
 
 
+def test_reserve_entity_echoes_the_written_value_even_below_the_floor():
+    """The reserve entity must echo exactly what Predbat wrote, floor or no floor.
+
+    Predbat writes this entity then reads it back to confirm (write_and_poll_value), so
+    republishing a clamped value can never match what was written and would retry until it
+    gave up — the same "didn't complete got 0.0" failure this component already had. The
+    floor is enforced at the API boundary in build_dynamic_payload instead, which is what
+    actually protects the battery.
+    """
+    failed = False
+    d = RecordingDeye()
+    d.device_list = ["INV1"]
+    d.device_values = {"INV1": {"soc": 50.0}}
+    d.device_battery_config = {"INV1": {"battLowCapacity": 14}}
+    entity = "number.predbat_deye_inv1_battery_schedule_reserve"
+    import tests.test_infra as ti
+
+    async def fake_apply(sn, schedule, current_soc, force=False):
+        """Stand in for the live control write."""
+        return True
+
+    # A below-floor write is echoed verbatim so the read-back matches
+    with patch.object(d, "apply_dynamic_control", side_effect=fake_apply):
+        ti.run_async(d.number_event(entity, 4))
+    if d.published.get(entity) != 4:
+        print(f"ERROR: expected the entity to echo 4, got {d.published.get(entity)!r}")
+        failed = True
+
+    # ...but the payload that reaches the inverter is still lifted to the floor
+    socs = [s["soc"] for s in d.build_dynamic_payload("INV1", d.local_schedule["INV1"], 50)["timeUseSettingItems"]]
+    if any(s < 14 for s in socs):
+        print(f"ERROR: the payload must still respect the 14% floor: {socs}")
+        failed = True
+    assert not failed, "test_reserve_entity_echoes_the_written_value_even_below_the_floor"
+
+
 def test_reserve_write_is_republished():
     """A reserve change is pushed to the inverter and republished for the read-back."""
     failed = False
@@ -559,6 +595,7 @@ def run_deye_publish_tests(my_predbat):
         ("control_writes_republished", test_control_entity_writes_are_republished),
         ("write_button_not_stored", test_write_button_applies_and_is_not_stored_as_schedule),
         ("reserve_republished", test_reserve_write_is_republished),
+        ("reserve_echoes_written", test_reserve_entity_echoes_the_written_value_even_below_the_floor),
         ("unknown_entity_ignored", test_unrelated_entity_does_not_corrupt_schedule),
     ]:
         try: