From 49f2d6fa1c28c790df5aa4542d4f7cf93ae62315 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Wed, 17 Jun 2026 14:11:44 +0100 Subject: [PATCH] feat(sigenergy): publish onboard_status sensor for the SaaS UI Tracks a per-system onboarding status string and publishes it as sensor._sigenergy__onboard_status so the SaaS connect UI can show a live approval stepper: - onboard_systems records pending_approval / in_other_vpp / firmware_no_vpp / no_permission from the API result codes - run() derives active / offboarded / pending_approval for visible systems and defaults not-yet-visible expected systems to not_onboarded - publishes the sensor for every expected system_id (covers pending systems) - test: onboard_status transitions for pending-review and in-other-VPP codes Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/predbat/sigenergy.py | 36 ++++++++++++++++++++++++++++ apps/predbat/tests/test_sigenergy.py | 25 +++++++++++++++++++ 2 files changed, 61 insertions(+) diff --git a/apps/predbat/sigenergy.py b/apps/predbat/sigenergy.py index 2a91ef3be..b429e45b7 100644 --- a/apps/predbat/sigenergy.py +++ b/apps/predbat/sigenergy.py @@ -282,6 +282,7 @@ def initialize(self, app_key, app_secret, base_url=None, mqtt_host=None, ca_cert self.system_status = {} # systemId → latest systemStatus dict self.daily_summary = {} # systemId → latest summary dict self.current_mode = {} # systemId → energyStorageOperationMode int + self.onboard_status = {} # systemId → onboarding status string (published for the SaaS UI) # Control state keyed by systemId self.controls = {} # systemId → {charge: {…}, export: {…}, reserve: …} @@ -824,14 +825,25 @@ async def onboard_systems(self, system_ids): code = item_codes[0] if code == SIGENERGY_CODE_SYSTEM_PENDING_REVIEW: + for sid in system_ids: + self.onboard_status[str(sid)] = "pending_approval" self.log("Warn: SigenergyAPI: Onboard for {} pending user approval in the Sigenergy app — approve to enable controls".format(system_ids)) return None if code in (SIGENERGY_CODE_IN_OTHER_VPP, SIGENERGY_CODE_IN_OTHER_VPP_EVERGEN): + for sid in system_ids: + self.onboard_status[str(sid)] = "in_other_vpp" self.log("Warn: SigenergyAPI: Onboard failed — system {} is registered to another VPP (code={})".format(system_ids, code)) return False if code == SIGENERGY_CODE_SOFTWARE_NO_VPP: + for sid in system_ids: + self.onboard_status[str(sid)] = "firmware_no_vpp" self.log("Warn: SigenergyAPI: Onboard failed — system {} firmware does not support VPP (code=1105)".format(system_ids)) return False + if code in (SIGENERGY_CODE_NO_PERMISSION_STATION, SIGENERGY_CODE_STATION_NOT_PERMITTED): + for sid in system_ids: + self.onboard_status[str(sid)] = "no_permission" + self.log("Warn: SigenergyAPI: Onboard failed — no permission for system {} (code={})".format(system_ids, code)) + return False if result is None: self.log("Warn: SigenergyAPI: Onboard request failed for systems {} (code={})".format(system_ids, code)) return False @@ -2089,6 +2101,7 @@ async def run(self, seconds, first): # For each expected system ID not yet visible, attempt onboarding missing_ids = self.system_id_filter - set(self.systems.keys()) if self.system_id_filter else set() for sid in missing_ids: + self.onboard_status.setdefault(str(sid), "not_onboarded") slug = self._system_slug(sid) is_offboard_at_start = self.get_state_wrapper("switch.{}_sigenergy_{}_offboard".format(self.prefix, slug), default="off") == "on" if is_offboard_at_start: @@ -2136,6 +2149,29 @@ async def run(self, seconds, first): slug = self._system_slug(sid) is_offboard = self.get_state_wrapper("switch.{}_sigenergy_{}_offboard".format(self.prefix, slug), default="off") == "on" await self._manage_vpp_registration(sid, is_readonly_vpp, is_offboard) + # Derive the user-facing onboarding status for the visible system. + if is_offboard: + self.onboard_status[str(sid)] = "offboarded" + elif self.current_mode.get(sid) == SIGENERGY_MODE_VPP: + self.onboard_status[str(sid)] = "active" + else: + self.onboard_status[str(sid)] = "pending_approval" + + # Publish onboarding status for the SaaS UI. Iterates the expected system IDs + # (not just visible ones) so a system still pending approval still gets a sensor. + if first or seconds % SIGENERGY_POLL_INTERVAL == 0: + for sid in (self.system_id_filter or set(self.systems.keys())): + slug = self._system_slug(sid) + self.dashboard_item( + "sensor.{}_sigenergy_{}_onboard_status".format(self.prefix, slug), + state=self.onboard_status.get(str(sid), "not_onboarded"), + attributes={ + "friendly_name": "Sigenergy {} Onboarding Status".format(sid), + "system_id": sid, + "in_vpp": self.current_mode.get(sid) == SIGENERGY_MODE_VPP, + }, + app="sigenergy", + ) # Fetch controls from HA on first run only if first: diff --git a/apps/predbat/tests/test_sigenergy.py b/apps/predbat/tests/test_sigenergy.py index 32745d185..e083f594a 100644 --- a/apps/predbat/tests/test_sigenergy.py +++ b/apps/predbat/tests/test_sigenergy.py @@ -1661,6 +1661,30 @@ async def mock_publish_controls(system_id=None): # --------------------------------------------------------------------------- +def test_sigenergy_onboard_status(my_predbat): + """onboard_systems records the user-facing onboarding status per result code.""" + failed = False + api = MockSigenergyAPI() + assert api.onboard_status == {}, "onboard_status starts empty" + + # Pending approval (1116) → pending_approval, returns None + api._request = AsyncMock(return_value=None) + api._last_api_code = SIGENERGY_CODE_SYSTEM_PENDING_REVIEW + result = asyncio.run(api.onboard_systems(["sys-1"])) + assert result is None, "pending review returns None" + assert api.onboard_status["sys-1"] == "pending_approval", "pending_approval status set" + + # Registered to another VPP (1103) → in_other_vpp, returns False + api2 = MockSigenergyAPI() + api2._request = AsyncMock(return_value=None) + api2._last_api_code = SIGENERGY_CODE_IN_OTHER_VPP + result2 = asyncio.run(api2.onboard_systems("sys-2")) + assert result2 is False, "in-other-vpp returns False" + assert api2.onboard_status["sys-2"] == "in_other_vpp", "in_other_vpp status set" + + return failed + + def run_sigenergy_tests(my_predbat): """Run all Sigenergy API unit tests. @@ -1671,6 +1695,7 @@ def run_sigenergy_tests(my_predbat): tests = [ ("helper_functions", test_sigenergy_helper_functions), ("initialize", test_sigenergy_initialize), + ("onboard_status", test_sigenergy_onboard_status), ("system_slug", test_sigenergy_system_slug), ("battery_capacity", test_sigenergy_battery_capacity), ("publish_system_entities", test_sigenergy_publish_system_entities),