From c114136d02cdaaeb929438b557a4763ef7d350ca Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Sat, 5 Sep 2026 11:43:22 +1000 Subject: [PATCH] feat(energy): add merge_live_status/merge_site_info wrappers Thin key-set-bound wrappers around merge_local_into_cloud so consumers don't need to import LOCAL_LIVE_STATUS_KEYS/LOCAL_SITE_INFO_KEYS themselves. Claude-Session: https://claude.ai/code/session_0147mRrswSGt8WKikQHd8B6s --- docs/energy_local_control.md | 22 ++++++++--------- tesla_fleet_api/router/__init__.py | 4 +++ tesla_fleet_api/router/energysite.py | 10 ++++++++ tesla_fleet_api/tesla/__init__.py | 4 +++ tests/test_energysite_merge_contract.py | 33 +++++++++++++++++++++++++ 5 files changed, 62 insertions(+), 11 deletions(-) diff --git a/docs/energy_local_control.md b/docs/energy_local_control.md index 02ce018..b172187 100644 --- a/docs/energy_local_control.md +++ b/docs/energy_local_control.md @@ -275,30 +275,30 @@ of the cloud response's keys but fills in `None` for the ones it cannot serve locally, so a consumer that wants "local readings where available, cloud otherwise" needs to merge the two responses itself, field by field. -`tesla_fleet_api.router.energysite` provides that merge as a plain function, -independent of `Router`: +`tesla_fleet_api.router.energysite` provides that merge via two entry points, +one per response shape: ```python -from tesla_fleet_api.router.energysite import ( - LOCAL_LIVE_STATUS_KEYS, - LOCAL_SITE_INFO_KEYS, - merge_local_into_cloud, -) +from tesla_fleet_api.router.energysite import merge_live_status, merge_site_info cloud_status = await teslemetry_energysite.live_status() local_status = await local_energysite.live_status() -merged = merge_local_into_cloud(cloud_status, local_status, LOCAL_LIVE_STATUS_KEYS) +merged = merge_live_status(cloud_status, local_status) ``` -- `LOCAL_LIVE_STATUS_KEYS` are the `live_status()` fields the local gateway can - actually serve; `LOCAL_SITE_INFO_KEYS` are the `site_info()` equivalent +- `merge_live_status` overlays the `live_status()` fields the local gateway can + actually serve; `merge_site_info` does the same for `site_info()` (`backup_reserve_percent`, `default_real_mode`). -- Only keys in the given set **and** present in `local` are overlaid onto a +- Only keys the wrapper owns **and** present in `local` are overlaid onto a copy of `cloud`; every other key keeps its cloud value. - If the local read failed entirely (`local is None`, e.g. the LAN call raised), the result is just `cloud` - the same fallback-to-cloud behavior `Router` gives non-merged commands. +Both wrappers call the underlying primitive, `merge_local_into_cloud(cloud, +local, owned_keys)`, which is still public for a caller with its own key set +(`LOCAL_LIVE_STATUS_KEYS`/`LOCAL_SITE_INFO_KEYS` are also exported). + ## See also - [Fleet API for Energy Sites](fleet_api_energy_sites.md) - the cloud diff --git a/tesla_fleet_api/router/__init__.py b/tesla_fleet_api/router/__init__.py index 473efe7..cdf2bfd 100644 --- a/tesla_fleet_api/router/__init__.py +++ b/tesla_fleet_api/router/__init__.py @@ -7,6 +7,8 @@ LOCAL_LIVE_STATUS_KEYS, LOCAL_SITE_INFO_KEYS, merge_local_into_cloud, + merge_live_status, + merge_site_info, ) __all__ = [ @@ -17,4 +19,6 @@ "LOCAL_LIVE_STATUS_KEYS", "LOCAL_SITE_INFO_KEYS", "merge_local_into_cloud", + "merge_live_status", + "merge_site_info", ] diff --git a/tesla_fleet_api/router/energysite.py b/tesla_fleet_api/router/energysite.py index 700bd6b..57b5317 100644 --- a/tesla_fleet_api/router/energysite.py +++ b/tesla_fleet_api/router/energysite.py @@ -43,6 +43,16 @@ def merge_local_into_cloud( return merged +def merge_live_status(cloud: dict[str, Any], local: dict[str, Any] | None) -> dict[str, Any]: + """Overlay a local Powerwall live_status onto the cloud document; see merge_local_into_cloud.""" + return merge_local_into_cloud(cloud, local, LOCAL_LIVE_STATUS_KEYS) + + +def merge_site_info(cloud: dict[str, Any], local: dict[str, Any] | None) -> dict[str, Any]: + """Overlay a local Powerwall site_info onto the cloud document; see merge_local_into_cloud.""" + return merge_local_into_cloud(cloud, local, LOCAL_SITE_INFO_KEYS) + + class EnergySiteRouter(Router[PrimaryT, SecondaryT]): """A :class:`Router` over energy-site instances. diff --git a/tesla_fleet_api/tesla/__init__.py b/tesla_fleet_api/tesla/__init__.py index 7f035cb..140983c 100644 --- a/tesla_fleet_api/tesla/__init__.py +++ b/tesla_fleet_api/tesla/__init__.py @@ -13,6 +13,8 @@ LOCAL_LIVE_STATUS_KEYS, LOCAL_SITE_INFO_KEYS, merge_local_into_cloud, + merge_live_status, + merge_site_info, ) from tesla_fleet_api.tesla.user import User from tesla_fleet_api.tesla.vehicle import ( @@ -45,4 +47,6 @@ "LOCAL_LIVE_STATUS_KEYS", "LOCAL_SITE_INFO_KEYS", "merge_local_into_cloud", + "merge_live_status", + "merge_site_info", ] diff --git a/tests/test_energysite_merge_contract.py b/tests/test_energysite_merge_contract.py index 144ade8..f90d4f3 100644 --- a/tests/test_energysite_merge_contract.py +++ b/tests/test_energysite_merge_contract.py @@ -9,7 +9,9 @@ from tesla_fleet_api.router import ( LOCAL_LIVE_STATUS_KEYS, LOCAL_SITE_INFO_KEYS, + merge_live_status, merge_local_into_cloud, + merge_site_info, ) @@ -67,5 +69,36 @@ def test_site_info_keys(self) -> None: self.assertEqual(result["default_real_mode"], "self_consumption") +class TestMergeLiveStatus(unittest.TestCase): + def test_matches_generic_call_with_live_status_keys(self) -> None: + cloud = {"solar_power": 100, "grid_power": 50} + local = {"solar_power": 200, "some_unowned_field": "surprise"} + expected = merge_local_into_cloud(cloud, local, LOCAL_LIVE_STATUS_KEYS) + result = merge_live_status(cloud, local) + self.assertEqual(result, expected) + self.assertEqual(result["solar_power"], 200) + self.assertEqual(result["grid_power"], 50) + self.assertNotIn("some_unowned_field", result) + + def test_local_none_passthrough(self) -> None: + cloud = {"solar_power": 100} + self.assertEqual(merge_live_status(cloud, None), cloud) + + +class TestMergeSiteInfo(unittest.TestCase): + def test_matches_generic_call_with_site_info_keys(self) -> None: + cloud = {"backup_reserve_percent": 20, "default_real_mode": "self_consumption"} + local = {"backup_reserve_percent": 30} + expected = merge_local_into_cloud(cloud, local, LOCAL_SITE_INFO_KEYS) + result = merge_site_info(cloud, local) + self.assertEqual(result, expected) + self.assertEqual(result["backup_reserve_percent"], 30) + self.assertEqual(result["default_real_mode"], "self_consumption") + + def test_local_none_passthrough(self) -> None: + cloud = {"backup_reserve_percent": 20} + self.assertEqual(merge_site_info(cloud, None), cloud) + + if __name__ == "__main__": unittest.main()