diff --git a/docs/fleet_api_energy_sites.md b/docs/fleet_api_energy_sites.md index 01f2437..50d4435 100644 --- a/docs/fleet_api_energy_sites.md +++ b/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 diff --git a/docs/teslemetry.md b/docs/teslemetry.md index 2fd1452..afac13a 100644 --- a/docs/teslemetry.md +++ b/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. diff --git a/tesla_fleet_api/tesla/energysite.py b/tesla_fleet_api/tesla/energysite.py index 50dab6c..1ea78ed 100644 --- a/tesla_fleet_api/tesla/energysite.py +++ b/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, diff --git a/tesla_fleet_api/teslemetry/energysite.py b/tesla_fleet_api/teslemetry/energysite.py index c1ec2a1..ad476ec 100644 --- a/tesla_fleet_api/teslemetry/energysite.py +++ b/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, diff --git a/tests/test_energysite_battery_dispatch.py b/tests/test_energysite_battery_dispatch.py new file mode 100644 index 0000000..49d2226 --- /dev/null +++ b/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}} + ), + )