From 90df94d53e5e3ffc2e8fff6f9e191e57f320d247 Mon Sep 17 00:00:00 2001 From: James Hodgkinson Date: Sun, 28 Apr 2024 18:33:34 +1000 Subject: [PATCH 1/3] typing fixes --- tesla_fleet_api/__init__.py | 15 +++++++++++++++ tesla_fleet_api/energy.py | 2 +- tesla_fleet_api/teslafleetapi.py | 11 ++++++----- tesla_fleet_api/teslafleetoauth.py | 6 +++--- tesla_fleet_api/teslemetry.py | 4 ++-- tesla_fleet_api/tessie.py | 3 +-- tesla_fleet_api/vehicle.py | 7 +++---- tesla_fleet_api/vehiclespecific.py | 2 +- 8 files changed, 32 insertions(+), 18 deletions(-) diff --git a/tesla_fleet_api/__init__.py b/tesla_fleet_api/__init__.py index 450e497..14af6b9 100644 --- a/tesla_fleet_api/__init__.py +++ b/tesla_fleet_api/__init__.py @@ -9,3 +9,18 @@ from .user import User from .vehicle import Vehicle from .vehiclespecific import VehicleSpecific + + +__all__ = [ + "TeslaFleetApi", + "TeslaFleetOAuth", + "Teslemetry", + "Tessie", + "Charging", + "Energy", + "EnergySpecific", + "Partner", + "User", + "Vehicle", + "VehicleSpecific", +] diff --git a/tesla_fleet_api/energy.py b/tesla_fleet_api/energy.py index 98289b6..85df226 100644 --- a/tesla_fleet_api/energy.py +++ b/tesla_fleet_api/energy.py @@ -91,7 +91,7 @@ async def grid_import_export( self, energy_site_id: int, disallow_charge_from_grid_with_solar_installed: bool | None = None, - customer_preferred_export_rule: EnergyExportMode|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._request( diff --git a/tesla_fleet_api/teslafleetapi.py b/tesla_fleet_api/teslafleetapi.py index 4e42d55..5906802 100644 --- a/tesla_fleet_api/teslafleetapi.py +++ b/tesla_fleet_api/teslafleetapi.py @@ -15,7 +15,7 @@ class TeslaFleetApi: """Class describing the Tesla Fleet API.""" - server: str + server: str | None = None session: aiohttp.ClientSession headers: dict[str, str] raise_for_status: bool @@ -23,7 +23,7 @@ class TeslaFleetApi: def __init__( self, session: aiohttp.ClientSession, - access_token: str, + access_token: str | None = None, region: str | None = None, server: str | None = None, raise_for_status: bool = True, @@ -61,7 +61,8 @@ async def find_server(self) -> str: for server in SERVERS.values(): self.server = server try: - response = await (self.user.region()).get("response") + region_response = await self.user.region() + response = region_response.get("response") if response: self.server = response["fleet_api_base_url"] LOGGER.debug("Using server %s", self.server) @@ -74,8 +75,8 @@ async def _request( self, method: Method, path: str, - params: dict[str:Any] | None = None, - json: dict[str:Any] | None = None, + params: dict[str, Any] | None = None, + json: dict[str, Any] | None = None, ): """Send a request to the Tesla Fleet API.""" diff --git a/tesla_fleet_api/teslafleetoauth.py b/tesla_fleet_api/teslafleetoauth.py index f809614..7607a88 100644 --- a/tesla_fleet_api/teslafleetoauth.py +++ b/tesla_fleet_api/teslafleetoauth.py @@ -63,13 +63,13 @@ async def get_refresh_token(self, client_secret: str, code: str, redirect_uri: s region = code.split("_")[0].lower() self.server = SERVERS.get(region) - async def check_access_token(self) -> str | None: + async def check_access_token(self) -> dict[str, Any] | None: """Get the access token.""" if self.access_token and self.expires > time.time(): - return + return None return await self.refresh_access_token() - async def refresh_access_token(self) -> str: + async def refresh_access_token(self) -> dict[str, Any]: """Refresh the access token.""" if not self.refresh_token: raise ValueError("Refresh token is missing") diff --git a/tesla_fleet_api/teslemetry.py b/tesla_fleet_api/teslemetry.py index 15f1dfa..8de8cfd 100644 --- a/tesla_fleet_api/teslemetry.py +++ b/tesla_fleet_api/teslemetry.py @@ -60,8 +60,8 @@ async def _request( self, method: Method, path: str, - params: dict[str:Any] | None = None, - json: dict[str:Any] | None = None, + params: dict[str, Any] | None = None, + json: dict[str, Any] | None = None, ): """Send a request to the Teslemetry API.""" async with rate_limit: diff --git a/tesla_fleet_api/tessie.py b/tesla_fleet_api/tessie.py index 44d22cf..38694f3 100644 --- a/tesla_fleet_api/tessie.py +++ b/tesla_fleet_api/tessie.py @@ -2,7 +2,6 @@ from typing import Any from .teslafleetapi import TeslaFleetApi from .const import Method -from .vehiclespecific import VehicleSpecific class Tessie(TeslaFleetApi): @@ -23,7 +22,7 @@ def __init__( energy_scope=False, ) - async def find_server(self): + async def find_server(self) -> str: """Find the server URL for the Tesla Fleet API.""" raise NotImplementedError("Do not use this function for Tessie.") diff --git a/tesla_fleet_api/vehicle.py b/tesla_fleet_api/vehicle.py index 6764927..1e4b722 100644 --- a/tesla_fleet_api/vehicle.py +++ b/tesla_fleet_api/vehicle.py @@ -567,7 +567,7 @@ async def trigger_homelink( lon: float | None = None, ) -> dict[str, Any]: """Turns on HomeLink (used to open and close garage doors).""" - data = {} + data: dict[str, str | float] = {} if token: data["token"] = token if lat and lon: @@ -742,12 +742,11 @@ async def vehicle_data( endpoints: List[VehicleDataEndpoint] | List[str] | None = None, ) -> dict[str, Any]: """Makes a live call to the vehicle. This may return cached data if the vehicle is offline. For vehicles running firmware versions 2023.38+, location_data is required to fetch vehicle location. This will result in a location sharing icon to show on the vehicle UI.""" - if isinstance(endpoints, list): - endpoints = ";".join(endpoints) + endpoint_payload = ";".join(endpoints) if endpoints else None return await self._request( Method.GET, f"api/1/vehicles/{vehicle_tag}/vehicle_data", - {"endpoints": endpoints}, + {"endpoints": endpoint_payload}, ) async def vehicle_subscriptions( diff --git a/tesla_fleet_api/vehiclespecific.py b/tesla_fleet_api/vehiclespecific.py index 35fd2d7..881ec52 100644 --- a/tesla_fleet_api/vehiclespecific.py +++ b/tesla_fleet_api/vehiclespecific.py @@ -13,7 +13,7 @@ class VehicleSpecific: """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle.""" - def __init__(self, parent, vin: str | None = None): + def __init__(self, parent, vin: str | int | None = None): self._parent = parent self.vin = vin From d6a9fa44c8e72f58849aedea4e708e94d36da2eb Mon Sep 17 00:00:00 2001 From: James Hodgkinson Date: Sun, 28 Apr 2024 18:38:20 +1000 Subject: [PATCH 2/3] return value annotation --- tesla_fleet_api/teslemetry.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tesla_fleet_api/teslemetry.py b/tesla_fleet_api/teslemetry.py index 8de8cfd..1c92b5a 100644 --- a/tesla_fleet_api/teslemetry.py +++ b/tesla_fleet_api/teslemetry.py @@ -52,7 +52,7 @@ async def metadata(self, update_region=True) -> dict[str, Any]: LOGGER.debug("Using server %s", self.server) return resp - async def find_server(self): + async def find_server(self) -> None: """Find the server URL for the Tesla Fleet API.""" await self.metadata(True) From 08c1534413062d7d4228c265d008ee80111be012 Mon Sep 17 00:00:00 2001 From: James Hodgkinson Date: Sun, 28 Apr 2024 18:44:07 +1000 Subject: [PATCH 3/3] more type-fix-chasing --- tesla_fleet_api/teslafleetapi.py | 11 ++++++----- tesla_fleet_api/teslafleetoauth.py | 10 ++++++---- tesla_fleet_api/teslemetry.py | 3 ++- 3 files changed, 14 insertions(+), 10 deletions(-) diff --git a/tesla_fleet_api/teslafleetapi.py b/tesla_fleet_api/teslafleetapi.py index 5906802..459a088 100644 --- a/tesla_fleet_api/teslafleetapi.py +++ b/tesla_fleet_api/teslafleetapi.py @@ -38,9 +38,10 @@ def __init__( self.session = session self.access_token = access_token - if region and not server and region not in SERVERS: - raise ValueError(f"Region must be one of {', '.join(SERVERS.keys())}") - self.server = server or SERVERS.get(region) + if region is not None: + if not server and region not in SERVERS: + raise ValueError(f"Region must be one of {', '.join(SERVERS.keys())}") + self.server = server or SERVERS.get(region) self.raise_for_status = raise_for_status LOGGER.debug("Using server %s", self.server) @@ -77,7 +78,7 @@ async def _request( path: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, - ): + ) -> dict[str, Any] | str: """Send a request to the Tesla Fleet API.""" if not self.server: @@ -126,7 +127,7 @@ async def _request( LOGGER.debug("Response Text: %s", data) return data - async def status(self): + async def status(self) -> str: """This endpoint returns the string "ok" if the API is operating normally. No HTTP headers are required.""" if not self.server: raise ValueError("Server was not set at init. Call find_server() first.") diff --git a/tesla_fleet_api/teslafleetoauth.py b/tesla_fleet_api/teslafleetoauth.py index 7607a88..631eee3 100644 --- a/tesla_fleet_api/teslafleetoauth.py +++ b/tesla_fleet_api/teslafleetoauth.py @@ -42,7 +42,9 @@ def get_login_url( """Get the login URL.""" return f"https://auth.tesla.com/oauth2/v3/authorize?response_type=code&client_id={self.client_id}&redirect_uri={redirect_uri}&scope={' '.join(scopes)}&state={state}" - async def get_refresh_token(self, client_secret: str, code: str, redirect_uri: str): + async def get_refresh_token( + self, client_secret: str, code: str, redirect_uri: str + ) -> None: """Get the refresh token.""" async with self.session.post( "https://auth.tesla.com/oauth2/v3/token", @@ -93,9 +95,9 @@ async def _request( self, method: Method, path: str, - params: dict | None = None, - data: dict | None = None, - ): + params: dict[str, Any] | None = None, + data: dict[str, Any] | None = None, + ) -> str | dict[str, Any]: """Send a request to the Tesla Fleet API.""" await self.check_access_token() return await super()._request(method, path, params, data) diff --git a/tesla_fleet_api/teslemetry.py b/tesla_fleet_api/teslemetry.py index 1c92b5a..b6630ff 100644 --- a/tesla_fleet_api/teslemetry.py +++ b/tesla_fleet_api/teslemetry.py @@ -52,6 +52,7 @@ async def metadata(self, update_region=True) -> dict[str, Any]: LOGGER.debug("Using server %s", self.server) return resp + # TODO: type this properly, it probably should return something async def find_server(self) -> None: """Find the server URL for the Tesla Fleet API.""" await self.metadata(True) @@ -62,7 +63,7 @@ async def _request( path: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, - ): + ) -> str | dict[str, Any]: """Send a request to the Teslemetry API.""" async with rate_limit: return await super()._request(method, path, params, json)