Add energy site dispatch_battery_power and resume_battery_operation

docs/fleet_api_energy_sites.md

@@ -416,6 +416,16 @@ and the `PrivateKeyError` raised for an unusable existing key file.
 | `get_backup_events()` | TEG | Backup event history (may timeout on some firmware) |
 | `schedule_backup_event()` | TEG | Schedule a manual backup event |
 | `cancel_backup_event()` | TEG | Cancel a scheduled backup event |
+| `dispatch_battery_power(power_watts, duration_seconds)` | TEG | Timed battery power setpoint: watts, positive discharges, negative charges; 1 to 86400 seconds |
+| `resume_battery_operation()` | TEG | Release a `dispatch_battery_power` setpoint |
+
+`dispatch_battery_power` holds the setpoint for `duration_seconds`, then the
+gateway returns to its configured operating mode by itself. Keep durations
+short and re-send to renew. The reply is an empty acknowledgement, not
+readback: read `live_status()` to see what the battery does. The gateway still
+applies its own backup reserve and grid limits. The method has the same name,
+arguments, units, sign and limits as aiopowerwall's local
+`dispatch_battery_power`, so `EnergySiteRouter` can send it on either path.
 
 ### Example
 

docs/teslemetry.md

@@ -635,6 +635,23 @@ async def main():
 asyncio.run(main())
 ```
 
+## Energy Site Battery Dispatch
+
+Teslemetry energy sites send `dispatch_battery_power(power_watts,
+duration_seconds)` and `resume_battery_operation()` through the Teslemetry
+custom routes `POST /api/1/energy_sites/{id}/command/dispatch_battery_power`
+and `POST /api/1/energy_sites/{id}/command/resume_battery_operation`.
+`power_watts` is battery power in watts: positive discharges, negative
+charges. `duration_seconds` is an integer from 1 to 86400. The library raises
+`ValueError` before any request on other values. See
+[Fleet API for Energy Sites](fleet_api_energy_sites.md#available-commands).
+
+```python
+energy_site = teslemetry.energySites.create(12345)
+await energy_site.dispatch_battery_power(power_watts=5000, duration_seconds=60)
+await energy_site.resume_battery_operation()
+```
+
 ## Migrate to OAuth
 
 The `migrate_to_oauth` method migrates from an access token to OAuth.

tesla_fleet_api/tesla/energysite.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import base64
+import math
 from typing import TYPE_CHECKING, Any
 
 from tesla_fleet_api.const import (
@@ -18,6 +19,28 @@
 if TYPE_CHECKING:
     from tesla_fleet_api.tesla.fleet import TeslaFleetApi
 
+# Upper bound on a ``dispatch_battery_power`` setpoint duration (one day),
+# matching aiopowerwall's local ``dispatch_battery_power``.
+MAX_DISPATCH_DURATION_SECONDS = 86400
+
+
+def validate_dispatch_battery_power(power_watts: float, duration_seconds: int) -> None:
+    """Raise ``ValueError`` on arguments aiopowerwall's local
+    ``dispatch_battery_power`` also rejects, so every backend fails the same way."""
+    if (
+        isinstance(power_watts, bool)
+        or not isinstance(power_watts, (int, float))  # pyright: ignore[reportUnnecessaryIsInstance]
+        or not math.isfinite(power_watts)
+    ):
+        raise ValueError("power_watts must be a finite number")
+    if type(duration_seconds) is not int or not (
+        1 <= duration_seconds <= MAX_DISPATCH_DURATION_SECONDS
+    ):
+        raise ValueError(
+            "duration_seconds must be an integer from 1 to "
+            f"{MAX_DISPATCH_DURATION_SECONDS}"
+        )
+
 
 class EnergySite:
     """Class describing the Tesla Fleet API partner endpoints"""
@@ -248,6 +271,53 @@ async def get_teg_config(self) -> dict[str, Any]:
         """Get Tesla Energy Gateway configuration from the TEG service."""
         return await self._command("teg", "get_config_request")
 
+    async def dispatch_battery_power(
+        self, power_watts: float, duration_seconds: int
+    ) -> dict[str, Any]:
+        """Request a timed battery power setpoint on the energy gateway.
+
+        Sends ``teg.dispatch_battery_power_request.real_power``.
+        ``power_watts`` is battery power in watts: **positive discharges,
+        negative charges**. The gateway holds the setpoint for
+        ``duration_seconds`` (1 to 86400) and then returns to its configured
+        operating mode by itself. A new call replaces an active setpoint.
+
+        Keep durations short and re-send to renew, so a caller that stops
+        cannot leave the battery on a stale setpoint. Call
+        :meth:`resume_battery_operation` to release early. The reply is an
+        empty acknowledgement, not readback; the gateway still applies its
+        own reserve and grid limits.
+
+        Same name, arguments, units, sign and limits as aiopowerwall's local
+        ``dispatch_battery_power``, so ``EnergySiteRouter`` can send it on
+        either path. Raises ``ValueError`` before any request on a
+        non-finite ``power_watts`` or an out-of-range ``duration_seconds``.
+        """
+        validate_dispatch_battery_power(power_watts, duration_seconds)
+        return await self._command(
+            "teg",
+            "dispatch_battery_power_request",
+            {
+                "real_power": {
+                    "power_watts": float(power_watts),
+                    "duration_seconds": duration_seconds,
+                }
+            },
+        )
+
+    async def resume_battery_operation(self) -> dict[str, Any]:
+        """Release a :meth:`dispatch_battery_power` setpoint.
+
+        Sends ``teg.dispatch_battery_power_request.resume_device_mode``; the
+        gateway goes back to its configured operating mode. The gateway also
+        acknowledges it when no setpoint is active.
+        """
+        return await self._command(
+            "teg",
+            "dispatch_battery_power_request",
+            {"resume_device_mode": {}},
+        )
+
     async def set_island_mode(
         self,
         mode: EnergyIslandMode | int,

tesla_fleet_api/teslemetry/energysite.py

@@ -24,7 +24,11 @@
     InvalidResponse,
     TeslaFleetError,
 )
-from tesla_fleet_api.tesla.energysite import EnergySite, EnergySites
+from tesla_fleet_api.tesla.energysite import (
+    EnergySite,
+    EnergySites,
+    validate_dispatch_battery_power,
+)
 
 DEFAULT_PAIRING_TIMEOUT = 600.0
 DEFAULT_PAIRING_POLL_INTERVAL = 5.0
@@ -449,6 +453,34 @@ async def remove_authorized_client(self, public_key: bytes | str) -> dict[str, A
             json={"public_key": public_key_b64},
         )
 
+    async def dispatch_battery_power(
+        self, power_watts: float, duration_seconds: int
+    ) -> dict[str, Any]:
+        """Request a timed battery power setpoint via the Teslemetry custom
+        endpoint.
+
+        ``power_watts`` is battery power in watts: **positive discharges,
+        negative charges**. The gateway holds it for ``duration_seconds``
+        (1 to 86400). See :meth:`EnergySite.dispatch_battery_power`.
+        """
+        validate_dispatch_battery_power(power_watts, duration_seconds)
+        return await self._request(
+            Method.POST,
+            f"api/1/energy_sites/{self.energy_site_id}/command/dispatch_battery_power",
+            json={
+                "power_watts": float(power_watts),
+                "duration_seconds": duration_seconds,
+            },
+        )
+
+    async def resume_battery_operation(self) -> dict[str, Any]:
+        """Release a :meth:`dispatch_battery_power` setpoint via the Teslemetry
+        custom endpoint."""
+        return await self._request(
+            Method.POST,
+            f"api/1/energy_sites/{self.energy_site_id}/command/resume_battery_operation",
+        )
+
     async def wait_until_paired(
         self,
         public_key: bytes | str,

tests/test_energysite_battery_dispatch.py

@@ -0,0 +1,217 @@
+"""Tests for dispatch_battery_power/resume_battery_operation.
+
+The Fleet API ``EnergySite`` sends the unsigned ``grpc_command`` envelope; the
+``TeslemetryEnergySite`` calls the two Teslemetry custom routes. Both share
+aiopowerwall's local method names, arguments, units, sign convention and
+duration limit, so ``EnergySiteRouter`` can send either command on the local or
+the cloud path.
+"""
+
+from __future__ import annotations
+
+import inspect
+import math
+from typing import Any
+from unittest import IsolatedAsyncioTestCase
+from unittest.mock import AsyncMock
+
+from tesla_fleet_api.const import Method
+from tesla_fleet_api.router import EnergySiteRouter
+from tesla_fleet_api.tesla.energysite import (
+    MAX_DISPATCH_DURATION_SECONDS,
+    EnergySite,
+)
+from tesla_fleet_api.teslemetry.energysite import TeslemetryEnergySite
+
+SITE_ID = 12345
+
+INVALID_ARGS: list[tuple[Any, Any]] = [
+    (math.nan, 60),
+    (math.inf, 60),
+    (-math.inf, 60),
+    (True, 60),
+    ("5000", 60),
+    (None, 60),
+    (5000, 0),
+    (5000, -1),
+    (5000, MAX_DISPATCH_DURATION_SECONDS + 1),
+    (5000, 60.0),
+    (5000, True),
+    (5000, "60"),
+]
+
+
+def _make_site(cls: type[EnergySite]) -> tuple[EnergySite, AsyncMock]:
+    request_mock: AsyncMock = AsyncMock(return_value={"response": {}})
+    parent = AsyncMock()
+    parent._request = request_mock
+    return cls(parent, SITE_ID), request_mock
+
+
+def _grpc_body(leaf: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "command_type": "grpc_command",
+        "command_properties": {
+            "message": {"teg": {"dispatch_battery_power_request": leaf}},
+            "identifier_type": 1,
+        },
+    }
+
+
+class FleetBatteryDispatchTests(IsolatedAsyncioTestCase):
+    async def test_dispatch_sends_real_power_leaf(self) -> None:
+        site, request_mock = _make_site(EnergySite)
+        await site.dispatch_battery_power(5000, 60)
+        request_mock.assert_awaited_once_with(
+            Method.POST,
+            f"api/1/energy_sites/{SITE_ID}/command",
+            json=_grpc_body(
+                {"real_power": {"power_watts": 5000.0, "duration_seconds": 60}}
+            ),
+        )
+
+    async def test_dispatch_negative_power_charges(self) -> None:
+        site, request_mock = _make_site(EnergySite)
+        await site.dispatch_battery_power(-2500.5, MAX_DISPATCH_DURATION_SECONDS)
+        leaf = request_mock.await_args.kwargs["json"]["command_properties"]["message"][
+            "teg"
+        ]["dispatch_battery_power_request"]
+        self.assertEqual(
+            leaf,
+            {
+                "real_power": {
+                    "power_watts": -2500.5,
+                    "duration_seconds": MAX_DISPATCH_DURATION_SECONDS,
+                }
+            },
+        )
+
+    async def test_resume_sends_resume_device_mode_leaf(self) -> None:
+        site, request_mock = _make_site(EnergySite)
+        await site.resume_battery_operation()
+        request_mock.assert_awaited_once_with(
+            Method.POST,
+            f"api/1/energy_sites/{SITE_ID}/command",
+            json=_grpc_body({"resume_device_mode": {}}),
+        )
+
+    async def test_dispatch_invalid_args_raise_without_request(self) -> None:
+        site, request_mock = _make_site(EnergySite)
+        for power_watts, duration_seconds in INVALID_ARGS:
+            with self.subTest(power=power_watts, duration=duration_seconds):
+                with self.assertRaises(ValueError):
+                    await site.dispatch_battery_power(power_watts, duration_seconds)
+        request_mock.assert_not_called()
+
+
+class TeslemetryBatteryDispatchTests(IsolatedAsyncioTestCase):
+    async def test_dispatch_calls_custom_route(self) -> None:
+        site, request_mock = _make_site(TeslemetryEnergySite)
+        await site.dispatch_battery_power(-3000, 30)
+        request_mock.assert_awaited_once_with(
+            Method.POST,
+            f"api/1/energy_sites/{SITE_ID}/command/dispatch_battery_power",
+            json={"power_watts": -3000.0, "duration_seconds": 30},
+        )
+
+    async def test_resume_calls_custom_route(self) -> None:
+        site, request_mock = _make_site(TeslemetryEnergySite)
+        await site.resume_battery_operation()
+        request_mock.assert_awaited_once_with(
+            Method.POST,
+            f"api/1/energy_sites/{SITE_ID}/command/resume_battery_operation",
+        )
+
+    async def test_dispatch_invalid_args_raise_without_request(self) -> None:
+        site, request_mock = _make_site(TeslemetryEnergySite)
+        for power_watts, duration_seconds in INVALID_ARGS:
+            with self.subTest(power=power_watts, duration=duration_seconds):
+                with self.assertRaises(ValueError):
+                    await site.dispatch_battery_power(power_watts, duration_seconds)
+        request_mock.assert_not_called()
+
+
+class _LocalPowerwallSite:
+    """Stand-in with aiopowerwall's local ``dispatch_battery_power`` signature."""
+
+    def __init__(self, *, fail: bool = False) -> None:
+        self.fail = fail
+        self.calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = []
+
+    async def dispatch_battery_power(
+        self, power_watts: float, duration_seconds: int
+    ) -> None:
+        self.calls.append(
+            ("dispatch_battery_power", (power_watts, duration_seconds), {})
+        )
+        if self.fail:
+            raise ConnectionError("local unreachable")
+
+    async def resume_battery_operation(self) -> None:
+        self.calls.append(("resume_battery_operation", (), {}))
+        if self.fail:
+            raise ConnectionError("local unreachable")
+
+
+class RouterBatteryDispatchTests(IsolatedAsyncioTestCase):
+    def test_cloud_signatures_match_local(self) -> None:
+        for name in ("dispatch_battery_power", "resume_battery_operation"):
+            local_params = list(
+                inspect.signature(getattr(_LocalPowerwallSite, name)).parameters
+            )
+            for cls in (EnergySite, TeslemetryEnergySite):
+                with self.subTest(cls=cls.__name__, name=name):
+                    self.assertEqual(
+                        list(inspect.signature(getattr(cls, name)).parameters),
+                        local_params,
+                    )
+
+    async def test_routes_to_local_primary(self) -> None:
+        local = _LocalPowerwallSite()
+        cloud, request_mock = _make_site(TeslemetryEnergySite)
+        router = EnergySiteRouter(local, cloud)
+
+        await router.dispatch_battery_power(power_watts=5000, duration_seconds=60)
+        await router.resume_battery_operation()
+
+        self.assertEqual(
+            local.calls,
+            [
+                ("dispatch_battery_power", (5000, 60), {}),
+                ("resume_battery_operation", (), {}),
+            ],
+        )
+        request_mock.assert_not_called()
+
+    async def test_falls_back_to_cloud_with_same_args(self) -> None:
+        local = _LocalPowerwallSite(fail=True)
+        cloud, request_mock = _make_site(TeslemetryEnergySite)
+        router = EnergySiteRouter(local, cloud)
+
+        await router.dispatch_battery_power(power_watts=-1500, duration_seconds=45)
+        request_mock.assert_awaited_once_with(
+            Method.POST,
+            f"api/1/energy_sites/{SITE_ID}/command/dispatch_battery_power",
+            json={"power_watts": -1500.0, "duration_seconds": 45},
+        )
+
+        request_mock.reset_mock()
+        await router.resume_battery_operation()
+        request_mock.assert_awaited_once_with(
+            Method.POST,
+            f"api/1/energy_sites/{SITE_ID}/command/resume_battery_operation",
+        )
+
+    async def test_falls_back_to_fleet_grpc(self) -> None:
+        local = _LocalPowerwallSite(fail=True)
+        fleet, request_mock = _make_site(EnergySite)
+        router = EnergySiteRouter(local, fleet)
+
+        await router.dispatch_battery_power(1000, 10)
+        request_mock.assert_awaited_once_with(
+            Method.POST,
+            f"api/1/energy_sites/{SITE_ID}/command",
+            json=_grpc_body(
+                {"real_power": {"power_watts": 1000.0, "duration_seconds": 10}}
+            ),
+        )