diff --git a/tesla_fleet_api/const.py b/tesla_fleet_api/const.py index eb4078e..c10e816 100644 --- a/tesla_fleet_api/const.py +++ b/tesla_fleet_api/const.py @@ -136,6 +136,22 @@ class Scope(StrEnum): ENERGY_CMDS = "energy_cmds" +class EnergyOperationMode(StrEnum): + """Energy Operation Mode options""" + + AUTONOMOUS = "autonomous" + SELF_CONSUMPTION = "self_consumption" + BACKUP = "backup" + + +class EnergyExportMode(StrEnum): + """Energy Export Mode options""" + + BATTERY_OK = "battery_ok" + PV_ONLY = "pv_only" + NEVER = "never" + + class TelemetryField(StrEnum): """Fields available in telemetry streams""" diff --git a/tesla_fleet_api/energy.py b/tesla_fleet_api/energy.py index 7d4c4fe..98289b6 100644 --- a/tesla_fleet_api/energy.py +++ b/tesla_fleet_api/energy.py @@ -1,5 +1,5 @@ from typing import Any -from .const import Method +from .const import Method, EnergyOperationMode, EnergyExportMode from .energyspecific import EnergySpecific @@ -91,24 +91,16 @@ async def grid_import_export( self, energy_site_id: int, disallow_charge_from_grid_with_solar_installed: bool | None = None, - customer_preferred_export_rule: str | None = None, + customer_preferred_export_rule: EnergyExportMode|str | None = None, ) -> dict[str, Any]: """Allow/disallow charging from the grid and exporting energy to the grid.""" - data = {} - if disallow_charge_from_grid_with_solar_installed is not None: - data[ - "disallow_charge_from_grid_with_solar_installed" - ] = disallow_charge_from_grid_with_solar_installed - if customer_preferred_export_rule is not None: - data["customer_preferred_export_rule"] = customer_preferred_export_rule - if not data: - raise ValueError( - "At least one of disallow_charge_from_grid_with_solar_installed or customer_preferred_export_rule must be set." - ) return await self._request( Method.POST, f"api/1/energy_sites/{energy_site_id}/grid_import_export", - json=data, + json={ + "disallow_charge_from_grid_with_solar_installed": disallow_charge_from_grid_with_solar_installed, + "customer_preferred_export_rule": customer_preferred_export_rule, + }, ) async def live_status(self, energy_site_id: int) -> dict[str, Any]: @@ -131,7 +123,7 @@ async def off_grid_vehicle_charging_reserve( ) async def operation( - self, energy_site_id: int, default_real_mode: str + self, energy_site_id: int, default_real_mode: EnergyOperationMode | str ) -> dict[str, Any]: """Set the site's mode.""" return await self._request( diff --git a/tesla_fleet_api/energyspecific.py b/tesla_fleet_api/energyspecific.py index c76510b..e78a5f7 100644 --- a/tesla_fleet_api/energyspecific.py +++ b/tesla_fleet_api/energyspecific.py @@ -1,4 +1,5 @@ from typing import Any +from .const import EnergyExportMode, EnergyOperationMode class EnergySpecific: @@ -69,7 +70,7 @@ async def energy_history( async def grid_import_export( self, disallow_charge_from_grid_with_solar_installed: bool | None = None, - customer_preferred_export_rule: str | None = None, + customer_preferred_export_rule: EnergyExportMode | str | None = None, ) -> dict[str, Any]: """Allow/disallow charging from the grid and exporting energy to the grid.""" return await self._parent.grid_import_export( @@ -90,7 +91,9 @@ async def off_grid_vehicle_charging_reserve( self.energy_site_id, off_grid_vehicle_charging_reserve_percent ) - async def operation(self, default_real_mode: str) -> dict[str, Any]: + async def operation( + self, default_real_mode: EnergyOperationMode | str + ) -> dict[str, Any]: """Set the site's mode.""" return await self._parent.operation( self.energy_site_id, diff --git a/tesla_fleet_api/exceptions.py b/tesla_fleet_api/exceptions.py index 49bfe50..d78a0ce 100644 --- a/tesla_fleet_api/exceptions.py +++ b/tesla_fleet_api/exceptions.py @@ -6,7 +6,7 @@ class TeslaFleetError(BaseException): """Base class for all Tesla exceptions.""" message: str - status: int + status: int = 0 error: str | None error_description: str | None @@ -245,7 +245,7 @@ async def raise_for_status(resp: aiohttp.ClientResponse) -> None: elif error == Error.UNAUTHORIZED_CLIENT: raise UnauthorizedClient(data) from e else: - raise TeslaFleetError(data) from e + raise InvalidRequest({error: e.message}) from e elif resp.status == 401: error = data.get("error") if error == Error.TOKEN_EXPIRED: @@ -253,7 +253,7 @@ async def raise_for_status(resp: aiohttp.ClientResponse) -> None: elif error == Error.MOBILE_ACCESS_DISABLED: raise MobileAccessDisabled(data) from e else: - raise InvalidToken(data) from e + raise InvalidToken({error: e.message}) from e elif resp.status == 402: raise PaymentRequired(data) from e elif resp.status == 403: diff --git a/tesla_fleet_api/teslafleetapi.py b/tesla_fleet_api/teslafleetapi.py index b2a0e77..71b0bb9 100644 --- a/tesla_fleet_api/teslafleetapi.py +++ b/tesla_fleet_api/teslafleetapi.py @@ -1,5 +1,6 @@ import logging import aiohttp +from json import dumps from .exceptions import raise_for_status, InvalidRegion, LibraryError from typing import Any from .const import SERVERS, Method, Error @@ -93,7 +94,7 @@ async def _request( LOGGER.debug("Parameters: %s", params) if json: json = {k: v for k, v in json.items() if v is not None} - LOGGER.debug("Body: %s", json) + LOGGER.debug("Body: %s", dumps(json)) async with self.session.request( method,