Put the answers in the diagnostics download

Every question asked on #267 needed something the download did not carry, so it
took three rounds of signed API requests written by hand to establish what the
integration already had in memory.

The download now includes, per inverter: which store governs and why, the
capability flags behind each unavailable control, the write-only enable pair
that nothing else can report, the periodic and legacy snapshots as they were
read, any open draft with its dirty fields and in-flight state, the duration
cooldowns, and the consecutive poll error count. Alongside it, how the client is
pacing calls and whether the fast lane is still on, which is what to look at when
something else shares the API account.

That one file would have answered the whole report: an empty dischargeTimeList,
a 0/0 enable pair with nothing recorded, and the capability flags explaining the
locked UI. Serials stay anonymised and credentials stay redacted.

Author: Poshy163

README.md

@@ -340,6 +340,29 @@ until a full periodic read succeeds. The integration never builds a
 replacement schedule from anything but a fresh read, because doing so could
 erase weekly settings, extra periods, or power values that it cannot see.
 
+### Reporting a schedule problem
+
+**Settings → Devices & Services → AlphaESS → ⋮ → Download diagnostics** answers
+most of what a report needs, without anyone having to sign API requests by hand.
+Alongside the usual entry and entity data, the `schedule` section carries, per
+inverter:
+
+- `governing_store` and `periodic_read` — which surface the system is on;
+- `capabilities` — the exact flags behind every unavailable control, so a locked
+  UI explains itself;
+- `enable_intent` / `enable_last_sent` — the write-only pair, which nothing else
+  can report;
+- `periodic_snapshot` and `legacy_snapshot` — what each store actually held,
+  including an empty period list or a stale window the app no longer shows;
+- `draft` — what is staged, which fields are dirty, and whether an Apply is in
+  flight;
+- cooldowns and consecutive poll errors.
+
+The `coordinator.api` section reports the current call spacing and whether the
+fast lane is still active, which is what to check when something else is sharing
+the same API account. Serials are replaced with `inverter_1`, `inverter_2` and so
+on, and credentials, keys and addresses are redacted.
+
 ### AlphaESS API limitations
 
 - The periodic API requires at least one complete charge period **and** one

custom_components/alphaess/coordinator.py

@@ -2593,6 +2593,71 @@ def _update_diagnostics(self) -> None:
                 self.get_periodic_read_state(serial)
             )
 
+    def schedule_diagnostics(self, serial: str) -> dict[str, Any]:
+        """Return everything needed to explain a schedule state from a report.
+
+        Every field here answers a question that has cost a round trip with a
+        tester: which store governs, why the controls are locked, what the two
+        write-only enable flags were last set to, and what the stores actually
+        held at the time. It must never raise - a diagnostics download is often
+        the only thing a user can still produce.
+        """
+        draft = self._schedule_drafts.get(serial)
+        dirty = self._schedule_draft_dirty.get(serial) or {}
+        now = time_mod.monotonic()
+
+        def _cooldown(store: dict[str, float]) -> float | None:
+            started = store.get(serial)
+            return None if started is None else round(now - started, 1)
+
+        if self.is_periodic_schedule_readable(serial):
+            governing_store = "periodic"
+        elif self.is_legacy_backup_active(serial):
+            governing_store = "legacy-backup"
+        else:
+            governing_store = "none"
+
+        return {
+            "governing_store": governing_store,
+            "periodic_read": self.get_periodic_read_state(serial),
+            "periodic_write_denied": serial in self._periodic_write_denied,
+            "capabilities": {
+                "time_based_control_active": self.is_time_based_control_active(serial),
+                "can_stage_schedule": self.can_stage_schedule(serial),
+                "can_modify_time_controls": self.can_modify_time_controls(serial),
+                "can_unlock_time_controls": self.can_unlock_time_controls(serial),
+                "can_reset_schedule": self.can_reset_schedule(serial),
+            },
+            # Write-only on the periodic API: the read reports 0 for both
+            # whatever the inverter is set to, so this record is the only
+            # answer there is, and an absent one blocks every write.
+            "enable_intent": self._periodic_enable_intent.get(serial),
+            "enable_last_sent": self._periodic_enable_sent.get(serial),
+            "periodic_snapshot": self._periodic_schedules.get(serial),
+            "legacy_snapshot": self._legacy_schedules.get(serial),
+            "draft": None if draft is None else {
+                "dirty": {side: sorted(fields) for side, fields in dirty.items()},
+                "revision": self._schedule_draft_revisions.get(serial, 0),
+                "apply_in_progress": serial in self._schedule_apply_in_progress,
+                "staged": draft,
+            },
+            "number_settings": self.number_settings.get(serial),
+            "seconds_since_last_charge_write": _cooldown(self.last_charge_update),
+            "seconds_since_last_discharge_write": _cooldown(self.last_discharge_update),
+            "consecutive_poll_errors": self._inverter_error_count.get(serial, 0),
+        }
+
+    def api_diagnostics(self) -> dict[str, Any]:
+        """Return how the client is currently pacing and answering."""
+        return {
+            "fast_api_lane": self._fast_api_lane,
+            "call_interval_seconds": self._call_interval(),
+            "throttle_multiplier": self.throttle_multiplier,
+            "last_poll_type": self._last_poll_type,
+            "last_full_poll": self._last_full_poll_utc or "never",
+            "poll_tick_count": self._poll_tick_count,
+        }
+
     def get_periodic_read_state(self, serial: str) -> str:
         """Return whether the periodic schedule can be read on this account.
 

custom_components/alphaess/diagnostics.py

@@ -45,6 +45,15 @@ async def async_get_config_entry_diagnostics(
         for idx, serial in enumerate(coordinator.data or {})
     }
 
+    # The state behind the schedule surface: which store governs, why the
+    # controls are in the state they are, and what the stores actually hold.
+    # None of it identifies anyone, and it is what a report needs to be
+    # answerable without a round trip.
+    schedule = {
+        f"inverter_{idx + 1}": coordinator.schedule_diagnostics(serial)
+        for idx, serial in enumerate(coordinator.data or {})
+    }
+
     return {
         "entry": {
             "data": async_redact_data(dict(entry.data), TO_REDACT_CONFIG),
@@ -62,6 +71,8 @@ async def async_get_config_entry_diagnostics(
             "model_list": coordinator.model_list,
             "has_throttle": coordinator.has_throttle,
             "periodic_schedule_read": periodic_schedule_read,
+            "api": coordinator.api_diagnostics(),
         },
+        "schedule": schedule,
         "data": inverters,
     }

tests/test_device_diagnostics.py

@@ -1,4 +1,6 @@
 """Tests for device info builders and diagnostics."""
+import time as time_mod
+
 from custom_components.alphaess.device import (
     build_ev_charger_device_info,
     build_inverter_device_info,
@@ -97,3 +99,112 @@ async def test_diagnostics_with_empty_data(self, mock_hass, make_coordinator):
 
         result = await async_get_config_entry_diagnostics(mock_hass, entry)
         assert result["data"] == {}
+
+
+class TestScheduleDiagnostics:
+    """Every field here exists because answering a report needed it and the
+    download did not have it."""
+
+    def _periodic(self):
+        return {
+            "executeCycleType": 1,
+            "chargeTimeList": [
+                {
+                    "beginTime": "05:00",
+                    "endTime": "06:00",
+                    "weeks": [1, 2, 3, 4, 5, 6, 7],
+                    "chargePower": 5000,
+                    "chargeLimit": 55,
+                }
+            ],
+            "dischargeTimeList": [],
+        }
+
+    async def test_a_periodic_system_reports_why_it_is_in_that_state(
+        self, mock_hass, make_coordinator
+    ):
+        coordinator = make_coordinator()
+        coordinator.data = {"AL_SERIAL": {"Model": "SMILE-G3-S5-INV"}}
+        coordinator._periodic_readable["AL_SERIAL"] = True
+        coordinator._periodic_schedules["AL_SERIAL"] = self._periodic()
+        coordinator.set_periodic_enable_intent(
+            "AL_SERIAL", grid_charge=1, ctr_dis=0,
+        )
+        coordinator.last_charge_update["AL_SERIAL"] = time_mod.monotonic()
+        entry = FakeEntry()
+        entry.runtime_data = coordinator
+
+        result = await async_get_config_entry_diagnostics(mock_hass, entry)
+        schedule = result["schedule"]["inverter_1"]
+
+        assert schedule["governing_store"] == "periodic"
+        assert schedule["periodic_read"] == "readable"
+        assert schedule["periodic_write_denied"] is False
+        # The write-only pair: the only record there is.
+        assert schedule["enable_intent"] == {
+            "gridChargeCycle": 1, "ctrDisCycle": 0,
+        }
+        # The resource itself, so an empty list is visible without asking.
+        assert schedule["periodic_snapshot"]["dischargeTimeList"] == []
+        assert schedule["capabilities"]["can_modify_time_controls"] is True
+        assert schedule["draft"] is None
+        assert schedule["seconds_since_last_charge_write"] is not None
+        assert schedule["seconds_since_last_discharge_write"] is None
+        assert result["coordinator"]["api"]["fast_api_lane"] is True
+
+    async def test_a_backup_system_reports_the_legacy_snapshot(
+        self, mock_hass, make_coordinator
+    ):
+        """The store a migrated system no longer uses is exactly what needs to
+        be visible when it is the one being read."""
+        coordinator = make_coordinator()
+        coordinator.data = {"AL_SERIAL": {"Model": "SMILE5-INV"}}
+        coordinator._periodic_readable["AL_SERIAL"] = False
+        coordinator._legacy_schedules["AL_SERIAL"] = {
+            "charge": {"gridCharge": 0, "timeChaf1": "17:00", "timeChae1": "17:15"},
+            "discharge": {"ctrDis": 0, "timeDisf1": "00:00", "timeDise1": "00:00"},
+        }
+        entry = FakeEntry()
+        entry.runtime_data = coordinator
+
+        result = await async_get_config_entry_diagnostics(mock_hass, entry)
+        schedule = result["schedule"]["inverter_1"]
+
+        assert schedule["governing_store"] == "legacy-backup"
+        assert schedule["legacy_snapshot"]["charge"]["timeChaf1"] == "17:00"
+        assert schedule["capabilities"]["can_reset_schedule"] is True
+
+    async def test_an_open_draft_is_described(self, mock_hass, make_coordinator):
+        coordinator = make_coordinator()
+        coordinator.data = {"AL_SERIAL": {"Model": "SMILE-G3-S5-INV"}}
+        coordinator._periodic_readable["AL_SERIAL"] = True
+        coordinator._periodic_schedules["AL_SERIAL"] = self._periodic()
+        coordinator.set_periodic_enable_intent(
+            "AL_SERIAL", grid_charge=1, ctr_dis=1,
+        )
+        coordinator.stage_schedule_change("AL_SERIAL", charge={"timeChaf1": "02:00"})
+        entry = FakeEntry()
+        entry.runtime_data = coordinator
+
+        result = await async_get_config_entry_diagnostics(mock_hass, entry)
+        draft = result["schedule"]["inverter_1"]["draft"]
+
+        assert draft["dirty"]["charge"] == ["timeChaf1"]
+        assert draft["apply_in_progress"] is False
+        assert draft["staged"]["charge"]["timeChaf1"] == "02:00"
+
+    async def test_a_system_with_no_usable_store_says_so(
+        self, mock_hass, make_coordinator
+    ):
+        coordinator = make_coordinator()
+        coordinator.data = {"AL_SERIAL": {"Model": "SMILE5-INV"}}
+        entry = FakeEntry()
+        entry.runtime_data = coordinator
+
+        result = await async_get_config_entry_diagnostics(mock_hass, entry)
+        schedule = result["schedule"]["inverter_1"]
+
+        assert schedule["governing_store"] == "none"
+        assert schedule["periodic_read"] == "unknown"
+        assert schedule["enable_intent"] is None
+        assert schedule["capabilities"]["can_stage_schedule"] is False