feat(energy): add local-Powerwall key-ownership merge contract

Add merge_local_into_cloud plus LOCAL_LIVE_STATUS_KEYS/LOCAL_SITE_INFO_KEYS
to router/energysite.py so a consumer can overlay local Powerwall readings
onto a cloud response without clobbering fields the local gateway can't
serve. PowerwallEnergySite.live_status() returns all cloud keys with None
for unservable ones, so presence-in-response can't be the ownership test -
a fixed owned-key set is what makes the overlay safe and lets a local
outage fall back to the cloud value.

Router dispatch is unchanged; this is a plain function next to
EnergySiteRouter, not new Router/health logic.

Claude-Session: https://claude.ai/code/session_01B18NQ2RAaJAX2rsiU31Qn6

Author: Bre77

docs/energy_local_control.md

@@ -266,6 +266,39 @@ and the `health` check).
 > status read (for example `live_status()`'s grid/island fields) before
 > trusting the response for anything actuation-critical.
 
+### Merge contract
+
+`EnergySiteRouter`'s per-command dispatch is **not** a merge - calling
+`router.live_status()` returns one whole source's response (local, or cloud on
+failover), not a blend of both. `PowerwallEnergySite.live_status()` reports all
+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`:
+
+```python
+from tesla_fleet_api.router.energysite import (
+    LOCAL_LIVE_STATUS_KEYS,
+    LOCAL_SITE_INFO_KEYS,
+    merge_local_into_cloud,
+)
+
+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)
+```
+
+- `LOCAL_LIVE_STATUS_KEYS` are the `live_status()` fields the local gateway can
+  actually serve; `LOCAL_SITE_INFO_KEYS` are the `site_info()` equivalent
+  (`backup_reserve_percent`, `default_real_mode`).
+- Only keys in the given set **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.
+
 ## See also
 
 - [Fleet API for Energy Sites](fleet_api_energy_sites.md) - the cloud

tesla_fleet_api/router/__init__.py

@@ -2,11 +2,19 @@
 
 from tesla_fleet_api.router.base import HealthCheck, Router
 from tesla_fleet_api.router.vehicle import VehicleRouter
-from tesla_fleet_api.router.energysite import EnergySiteRouter
+from tesla_fleet_api.router.energysite import (
+    EnergySiteRouter,
+    LOCAL_LIVE_STATUS_KEYS,
+    LOCAL_SITE_INFO_KEYS,
+    merge_local_into_cloud,
+)
 
 __all__ = [
     "Router",
     "VehicleRouter",
     "EnergySiteRouter",
     "HealthCheck",
+    "LOCAL_LIVE_STATUS_KEYS",
+    "LOCAL_SITE_INFO_KEYS",
+    "merge_local_into_cloud",
 ]

tesla_fleet_api/router/energysite.py

@@ -1,7 +1,47 @@
 from __future__ import annotations
 
+from typing import Any
+
 from tesla_fleet_api.router.base import PrimaryT, Router, SecondaryT
 
+LOCAL_LIVE_STATUS_KEYS: frozenset[str] = frozenset(
+    {
+        "solar_power",
+        "energy_left",
+        "total_pack_energy",
+        "percentage_charged",
+        "battery_power",
+        "load_power",
+        "grid_power",
+        "generator_power",
+        "grid_status",
+        "island_status",
+    }
+)
+LOCAL_SITE_INFO_KEYS: frozenset[str] = frozenset(
+    {"backup_reserve_percent", "default_real_mode"}
+)
+
+
+def merge_local_into_cloud(
+    cloud: dict[str, Any], local: dict[str, Any] | None, owned_keys: frozenset[str]
+) -> dict[str, Any]:
+    """Overlay owned_keys present in local onto a copy of cloud; every other key keeps its cloud value.
+
+    ``PowerwallEnergySite.live_status()`` returns all cloud keys with ``None``
+    for the ones it cannot serve, so presence-in-response cannot be the
+    ownership test — a fixed owned-key set lets a caller overlay local
+    readings without clobbering cloud values, and lets a local outage fall
+    back to the cloud value instead of an unavailable one.
+    """
+    merged = dict(cloud)
+    if local is None:
+        return merged
+    for key in owned_keys:
+        if key in local:
+            merged[key] = local[key]
+    return merged
+
 
 class EnergySiteRouter(Router[PrimaryT, SecondaryT]):
     """A :class:`Router` over energy-site instances.

tesla_fleet_api/tesla/__init__.py

@@ -6,7 +6,14 @@
 from tesla_fleet_api.tesla.charging import Charging
 from tesla_fleet_api.tesla.energysite import EnergySites, EnergySite
 from tesla_fleet_api.tesla.partner import Partner
-from tesla_fleet_api.router import Router, VehicleRouter, EnergySiteRouter
+from tesla_fleet_api.router import (
+    Router,
+    VehicleRouter,
+    EnergySiteRouter,
+    LOCAL_LIVE_STATUS_KEYS,
+    LOCAL_SITE_INFO_KEYS,
+    merge_local_into_cloud,
+)
 from tesla_fleet_api.tesla.user import User
 from tesla_fleet_api.tesla.vehicle import (
     Vehicles,
@@ -35,4 +42,7 @@
     "VehicleBluetooth",
     "Router",
     "VehicleRouter",
+    "LOCAL_LIVE_STATUS_KEYS",
+    "LOCAL_SITE_INFO_KEYS",
+    "merge_local_into_cloud",
 ]

tests/test_energysite_merge_contract.py

@@ -0,0 +1,71 @@
+"""Unit tests for the local-Powerwall key-ownership merge contract.
+
+`merge_local_into_cloud` is a plain function, independent of `Router`
+dispatch - see docs/energy_local_control.md's "Merge contract" section.
+"""
+
+import unittest
+
+from tesla_fleet_api.router import (
+    LOCAL_LIVE_STATUS_KEYS,
+    LOCAL_SITE_INFO_KEYS,
+    merge_local_into_cloud,
+)
+
+
+class TestMergeLocalIntoCloud(unittest.TestCase):
+    def test_local_none_returns_cloud_values(self) -> None:
+        cloud = {"solar_power": 100, "grid_power": 50, "generator_power": 0}
+        result = merge_local_into_cloud(cloud, None, LOCAL_LIVE_STATUS_KEYS)
+        self.assertEqual(result, cloud)
+
+    def test_owned_key_present_in_local_is_overlaid(self) -> None:
+        cloud = {"solar_power": 100, "grid_power": 50}
+        local = {"solar_power": 200}
+        result = merge_local_into_cloud(cloud, local, LOCAL_LIVE_STATUS_KEYS)
+        self.assertEqual(result["solar_power"], 200)
+        self.assertEqual(result["grid_power"], 50)
+
+    def test_owned_key_absent_from_local_keeps_cloud_value(self) -> None:
+        cloud = {"solar_power": 100, "grid_power": 50}
+        local: dict = {"solar_power": 200}
+        result = merge_local_into_cloud(cloud, local, LOCAL_LIVE_STATUS_KEYS)
+        self.assertEqual(result["grid_power"], 50)
+
+    def test_falsy_but_present_local_value_is_overlaid(self) -> None:
+        cloud = {"grid_power": 999}
+        local = {"grid_power": 0}
+        result = merge_local_into_cloud(cloud, local, LOCAL_LIVE_STATUS_KEYS)
+        self.assertEqual(result["grid_power"], 0)
+
+    def test_local_keys_outside_owned_set_are_ignored(self) -> None:
+        cloud = {"solar_power": 100}
+        local = {"solar_power": 200, "some_unowned_field": "surprise"}
+        result = merge_local_into_cloud(cloud, local, LOCAL_LIVE_STATUS_KEYS)
+        self.assertNotIn("some_unowned_field", result)
+
+    def test_does_not_mutate_inputs(self) -> None:
+        cloud = {"solar_power": 100}
+        local = {"solar_power": 200}
+        cloud_copy = dict(cloud)
+        local_copy = dict(local)
+        result = merge_local_into_cloud(cloud, local, LOCAL_LIVE_STATUS_KEYS)
+        self.assertEqual(cloud, cloud_copy)
+        self.assertEqual(local, local_copy)
+        self.assertIsNot(result, cloud)
+
+    def test_returns_new_dict_when_local_is_none(self) -> None:
+        cloud = {"solar_power": 100}
+        result = merge_local_into_cloud(cloud, None, LOCAL_LIVE_STATUS_KEYS)
+        self.assertIsNot(result, cloud)
+
+    def test_site_info_keys(self) -> None:
+        cloud = {"backup_reserve_percent": 20, "default_real_mode": "self_consumption"}
+        local = {"backup_reserve_percent": 30}
+        result = merge_local_into_cloud(cloud, local, LOCAL_SITE_INFO_KEYS)
+        self.assertEqual(result["backup_reserve_percent"], 30)
+        self.assertEqual(result["default_real_mode"], "self_consumption")
+
+
+if __name__ == "__main__":
+    unittest.main()