diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 37e0728..761d218 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -3,8 +3,27 @@ name: Test tesla_api on: pull_request jobs: - build: + flake8: + name: Lint with flake8 + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + - uses: actions/setup-python@v2 + - run: | + pip install -r requirements-flake8.txt + flake8 + + mypy: + name: Check annotations with Mypy + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + - uses: actions/setup-python@v2 + - run: pip install aiohttp mypy + - run: mypy + test: + name: Tests runs-on: ubuntu-latest strategy: matrix: @@ -18,12 +37,8 @@ jobs: python-version: ${{ matrix.python-version }} - name: Install dependencies run: | - python -m pip install --upgrade pip - pip install -e .[dev] - pip install -r requirements-dev.txt - - name: Run flake8 - run: | - flake8 + python -m pip install -U pip + pip install aiohttp pytest-cov - name: Test with pytest run: | pytest -vv --cov=tesla_api --cov-report=xml diff --git a/.gitignore b/.gitignore index 1f7ec5e..28563ff 100644 --- a/.gitignore +++ b/.gitignore @@ -1,7 +1,8 @@ test.py +.mypy_cache/ .vscode/ **/__pycache__/ build/ tesla_api.egg-info/ -dist/ \ No newline at end of file +dist/ diff --git a/.mypy.ini b/.mypy.ini new file mode 100644 index 0000000..1be381e --- /dev/null +++ b/.mypy.ini @@ -0,0 +1,25 @@ +[mypy] +files = tesla_api/ +check_untyped_defs = True +follow_imports_for_stubs = True +disallow_any_decorated = True +disallow_any_expr = True +disallow_any_explicit = True +disallow_any_generics = True +disallow_any_unimported = True +disallow_incomplete_defs = True +disallow_subclassing_any = True +disallow_untyped_calls = True +disallow_untyped_decorators = True +disallow_untyped_defs = True +implicit_reexport = False +no_implicit_optional = True +show_error_codes = True +strict_equality = True +warn_incomplete_stub = True +warn_no_return = True +warn_redundant_casts = True +warn_unreachable = True +warn_unused_configs = True +warn_unused_ignores = True +warn_return_any = True diff --git a/requirements-dev.txt b/requirements-flake8.txt similarity index 84% rename from requirements-dev.txt rename to requirements-flake8.txt index ce1f291..a621973 100644 --- a/requirements-dev.txt +++ b/requirements-flake8.txt @@ -1,6 +1,5 @@ +flake8 flake8-bugbear #flake8-docstrings # TODO: Resolve violations. flake8-import-order flake8-quotes -pytest -pytest-cov diff --git a/setup.py b/setup.py index 8780a9a..7428a78 100644 --- a/setup.py +++ b/setup.py @@ -12,6 +12,7 @@ long_description=long_description, long_description_content_type="text/markdown", url="https://github.com/mlowijs/tesla_api", + package_data={"tesla_api": ["py.typed"]}, packages=find_packages(), classifiers=[ "Programming Language :: Python :: 3.7", diff --git a/tesla_api/__init__.py b/tesla_api/__init__.py index 6d5ab02..3eb0acc 100644 --- a/tesla_api/__init__.py +++ b/tesla_api/__init__.py @@ -1,13 +1,22 @@ import asyncio import json from datetime import datetime, timedelta +from types import TracebackType +from typing import (Awaitable, Callable, List, Literal, Mapping, Optional, Type, TypeVar, + TypedDict, Union, cast) import aiohttp +from .datatypes import (BaseResponse, EnergySite, ErrorResponse, ProductsResponse, + TokenParams, TokenResponse, VehiclesResponse) from .energy import Energy -from .exceptions import ApiError, AuthenticationError, VehicleUnavailableError +from .exceptions import (ApiError, AuthenticationError, VehicleInServiceError, + VehicleUnavailableError) from .vehicle import Vehicle +__all__ = ("Energy", "Vehicle", "TeslaApiClient", "ApiError", "AuthenticationError", + "VehicleInServiceError", "VehicleUnavailableError") + TESLA_API_BASE_URL = "https://owner-api.teslamotors.com/" TOKEN_URL = TESLA_API_BASE_URL + "oauth/token" API_URL = TESLA_API_BASE_URL + "api/1" @@ -16,12 +25,35 @@ OAUTH_CLIENT_SECRET = "c7257eb71a564034f9419ee651c7d0e5f7aa6bfbd18bafb5c5c033b093bb2fa3" +class AuthHeaders(TypedDict): + Authorization: str + + +class _AuthParamsPassword(TypedDict): + grant_type: Literal["password"] + email: str + password: str + + +class _AuthParamsRefresh(TypedDict): + grant_type: Literal["refresh_token"] + refresh_token: str + + +AuthParams = Union[_AuthParamsPassword, _AuthParamsRefresh] +T = TypeVar("T", bound="TeslaApiClient") + + class TeslaApiClient: - callback_update = None # Called when vehicle's state has been updated. - callback_wake_up = None # Called when attempting to wake a vehicle. + # Called when vehicle's state has been updated. + callback_update: Optional[Callable[[Vehicle], Awaitable[None]]] = None + # Called when attempting to wake a vehicle. + callback_wake_up: Optional[Callable[[Vehicle], Awaitable[None]]] = None timeout = 30 # Default timeout for operations such as Vehicle.wake_up(). - def __init__(self, email=None, password=None, token=None, on_new_token=None): + def __init__(self, email: Optional[str] = None, password: Optional[str] = None, + token: Optional[str] = None, + on_new_token: Optional[Callable[[str], Awaitable[None]]] = None): """Creates client from provided credentials. If token is not provided, or is no longer valid, then a new token will @@ -35,28 +67,22 @@ def __init__(self, email=None, password=None, token=None, on_new_token=None): assert token is not None or (email is not None and password is not None) self._email = email self._password = password - self._token = json.loads(token) if token else None + self._token = cast(TokenResponse, json.loads(token)) if token else None self._new_token_callback = on_new_token self._session = aiohttp.ClientSession() - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - await self.close() - - async def close(self): + async def close(self) -> None: await self._session.close() - async def _get_token(self, data): - request_data = { + async def _get_token(self, data: AuthParams) -> TokenResponse: + request_data = cast(TokenParams, { "client_id": OAUTH_CLIENT_ID, "client_secret": OAUTH_CLIENT_SECRET, - } - request_data.update(data) + **data + }) async with self._session.post(TOKEN_URL, data=request_data) as resp: - response_json = await resp.json() + response_json = await cast(Awaitable[TokenResponse], resp.json()) if resp.status == 401: raise AuthenticationError(response_json) @@ -66,15 +92,17 @@ async def _get_token(self, data): return response_json - async def _get_new_token(self): - data = {"grant_type": "password", "email": self._email, "password": self._password} + async def _get_new_token(self) -> TokenResponse: + assert self._email is not None and self._password is not None + data: _AuthParamsPassword = {"grant_type": "password", "email": self._email, + "password": self._password} return await self._get_token(data) - async def _refresh_token(self, refresh_token): - data = {"grant_type": "refresh_token", "refresh_token": refresh_token} + async def _refresh_token(self, refresh_token: str) -> TokenResponse: + data: _AuthParamsRefresh = {"grant_type": "refresh_token", "refresh_token": refresh_token} return await self._get_token(data) - async def authenticate(self): + async def authenticate(self) -> None: if not self._token: self._token = await self._get_new_token() @@ -84,40 +112,53 @@ async def authenticate(self): if datetime.utcnow() >= expiration_date: self._token = await self._refresh_token(self._token["refresh_token"]) - def _get_headers(self): - return {"Authorization": "Bearer {}".format(self._token["access_token"])} - - async def get(self, endpoint, params=None): - await self.authenticate() - url = "{}/{}".format(API_URL, endpoint) + def _get_headers(self) -> AuthHeaders: + assert self._token is not None + return { + "Authorization": "Bearer {}".format(self._token["access_token"]) + } - async with self._session.get(url, headers=self._get_headers(), params=params) as resp: - response_json = await resp.json() + async def get(self, endpoint: str, params: Optional[Mapping[str, str]] = None) -> object: + return await self._send_request("get", endpoint, params=params) - if "error" in response_json: - if "vehicle unavailable" in response_json["error"]: - raise VehicleUnavailableError() - raise ApiError(response_json["error"]) + async def post(self, endpoint: str, data: Optional[Mapping[str, object]] = None) -> object: + return await self._send_request("post", endpoint, data=data) - return response_json["response"] - - async def post(self, endpoint, data=None): + async def _send_request(self, method: Literal["get", "post"], endpoint: str, *, + data: Optional[Mapping[str, object]] = None, + params: Optional[Mapping[str, str]] = None) -> object: await self.authenticate() url = "{}/{}".format(API_URL, endpoint) - async with self._session.post(url, headers=self._get_headers(), json=data) as resp: - response_json = await resp.json() + async with self._session.request(method, url, headers=self._get_headers(), + json=data, params=params) as resp: + # TODO(Mypy): https://github.com/python/mypy/issues/8884 + response_json = await cast(Awaitable[Union[BaseResponse, ErrorResponse]], resp.json()) if "error" in response_json: - if "vehicle unavailable" in response_json["error"]: + error_response = cast(ErrorResponse, response_json) + error = error_response["error"] + if "vehicle unavailable" in error: raise VehicleUnavailableError() - raise ApiError(response_json["error"]) + elif "in service" in error: + raise VehicleInServiceError() + raise ApiError(error) + response_json = cast(BaseResponse, response_json) return response_json["response"] - async def list_vehicles(self): - return [Vehicle(self, vehicle) for vehicle in await self.get("vehicles")] + async def list_vehicles(self) -> List[Vehicle]: + vehicles = cast(VehiclesResponse, await self.get("vehicles")) + return [Vehicle(self, v) for v in vehicles] + + async def list_energy_sites(self) -> List[Energy]: + products = cast(ProductsResponse, await self.get("products")) + return [Energy(self, cast(EnergySite, p)["energy_site_id"]) + for p in products if "energy_site_id" in p] - async def list_energy_sites(self): - return [Energy(self, product["energy_site_id"]) - for product in await self.get("products") if "energy_site_id" in product] + async def __aenter__(self: T) -> T: + return self + + async def __aexit__(self, exc_type: Optional[Type[BaseException]], + exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]) -> None: + await self.close() diff --git a/tesla_api/charge.py b/tesla_api/charge.py index fbb0b08..e48c523 100644 --- a/tesla_api/charge.py +++ b/tesla_api/charge.py @@ -1,22 +1,31 @@ +from typing import TYPE_CHECKING, cast + +from .datatypes import ChargeStateResponse + +if TYPE_CHECKING: + from .vehicle import Vehicle + + class Charge: - def __init__(self, vehicle): + def __init__(self, vehicle: "Vehicle"): self._vehicle = vehicle self._api_client = vehicle._api_client - async def get_state(self): - return await self._api_client.get( - "vehicles/{}/data_request/charge_state".format(self._vehicle.id)) + async def get_state(self) -> ChargeStateResponse: + endpoint = "vehicles/{}/data_request/charge_state".format(self._vehicle.id) + return cast(ChargeStateResponse, await self._api_client.get(endpoint)) - async def start_charging(self): - return await self._vehicle._command("charge_start") + async def start_charging(self) -> bool: + return cast(bool, await self._vehicle._command("charge_start")) - async def stop_charging(self): - return await self._vehicle._command("charge_stop") + async def stop_charging(self) -> bool: + return cast(bool, await self._vehicle._command("charge_stop")) - async def set_charge_limit(self, percentage): + async def set_charge_limit(self, percentage: int) -> bool: # TODO: int or float? percentage = round(percentage) if not (50 <= percentage <= 100): raise ValueError("Percentage should be between 50 and 100") - return await self._vehicle._command("set_charge_limit", {"percent": percentage}) + args = {"percent": percentage} + return cast(bool, await self._vehicle._command("set_charge_limit", args)) diff --git a/tesla_api/climate.py b/tesla_api/climate.py index 28eba37..5270e22 100644 --- a/tesla_api/climate.py +++ b/tesla_api/climate.py @@ -1,40 +1,59 @@ +from enum import Enum from functools import partialmethod +from typing import Optional, TYPE_CHECKING, cast + +from .datatypes import ClimateStateResponse + +if TYPE_CHECKING: + from .vehicle import Vehicle + + +class SeatPosition(Enum): + DRIVER = 0 + PASSENGER = 1 + REAR_LEFT = 2 + REAR_CENTER = 4 + REAR_RIGHT = 5 class Climate: - def __init__(self, vehicle): + def __init__(self, vehicle: "Vehicle"): self._vehicle = vehicle self._api_client = vehicle._api_client - async def get_state(self): - return await self._api_client.get( - "vehicles/{}/data_request/climate_state".format(self._vehicle.id)) + async def get_state(self) -> ClimateStateResponse: + endpoint = "vehicles/{}/data_request/climate_state".format(self._vehicle.id) + return cast(ClimateStateResponse, await self._api_client.get(endpoint)) - async def start_climate(self): - return await self._vehicle._command("auto_conditioning_start") + async def start_climate(self) -> bool: + return cast(bool, await self._vehicle._command("auto_conditioning_start")) - async def stop_climate(self): - return await self._vehicle._command("auto_conditioning_stop") + async def stop_climate(self) -> bool: + return cast(bool, await self._vehicle._command("auto_conditioning_stop")) - async def set_temperature(self, driver_temperature, passenger_temperature=None): + async def set_temperature(self, driver_temperature: float, # TODO: Does int work? + passenger_temperature: Optional[float] = None) -> bool: data = {"driver_temp": driver_temperature, "passenger_temp": passenger_temperature or driver_temperature} - return await self._vehicle._command("set_temps", data) - - async def set_seat_heater(self, temp=0, seat=0): - # temp = The desired level for the heater. (0-3) - # The desired seat to heat. (0-5) - # 0 - Driver - # 1 - Passenger - # 2 - Rear left - # 4 - Rear center - # 5 - Rear right - return await self._vehicle._command("remote_seat_heater_request", - {"heater": seat, "level": temp}) - - async def steering_wheel_heater(self, on: bool): - return await self._vehicle._command("remote_steering_wheel_heater_request", - {"on": on}) - + return cast(bool, await self._vehicle._command("set_temps", data)) + + async def set_seat_heater(self, temp: int = 0, + seat: SeatPosition = SeatPosition.DRIVER) -> bool: + """Set a seat heater. + + Args: + temp: The desired level for the heater. (0-3) + seat: The desired seat to heat. + """ + if temp < 0 or temp > 3: + raise ValueError("temp must be in the range 0-3") + + args = {"heater": seat, "level": temp} + return cast(bool, await self._vehicle._command("remote_seat_heater_request", args)) + + async def steering_wheel_heater(self, on: bool) -> bool: + endpoint = "remote_steering_wheel_heater_request" + args = {"on": on} + return cast(bool, await self._vehicle._command(endpoint, args)) start_steering_wheel_heater = partialmethod(steering_wheel_heater, True) stop_steering_wheel_heater = partialmethod(steering_wheel_heater, False) diff --git a/tesla_api/controls.py b/tesla_api/controls.py index f59a360..1a639de 100644 --- a/tesla_api/controls.py +++ b/tesla_api/controls.py @@ -1,31 +1,38 @@ +from enum import Enum from functools import partialmethod +from typing import TYPE_CHECKING, cast -STATE_VENT = "vent" -STATE_CLOSE = "close" +if TYPE_CHECKING: + from .vehicle import Vehicle + + +class SunroofState(Enum): + STATE_VENT = "vent" + STATE_CLOSE = "close" class Controls: - def __init__(self, vehicle): + def __init__(self, vehicle: "Vehicle"): self._vehicle = vehicle self._api_client = vehicle._api_client - async def _set_sunroof_state(self, state): - return await self._vehicle._command("sun_roof_control", {"state": state}) - - vent_sunroof = partialmethod(_set_sunroof_state, STATE_VENT) - close_sunroof = partialmethod(_set_sunroof_state, STATE_CLOSE) + async def _set_sunroof_state(self, state: SunroofState) -> bool: + args = {"state": state} + return cast(bool, await self._vehicle._command("sun_roof_control", args)) + vent_sunroof = partialmethod(_set_sunroof_state, SunroofState.STATE_VENT) + close_sunroof = partialmethod(_set_sunroof_state, SunroofState.STATE_CLOSE) - async def flash_lights(self): - return await self._vehicle._command("flash_lights") + async def flash_lights(self) -> bool: + return cast(bool, await self._vehicle._command("flash_lights")) - async def honk_horn(self): - return await self._vehicle._command("honk_horn") + async def honk_horn(self) -> bool: + return cast(bool, await self._vehicle._command("honk_horn")) - async def open_charge_port(self): - return await self._vehicle._command("charge_port_door_open") + async def open_charge_port(self) -> bool: + return cast(bool, await self._vehicle._command("charge_port_door_open")) - async def door_lock(self): - return await self._vehicle._command("door_lock") + async def door_lock(self) -> bool: + return cast(bool, await self._vehicle._command("door_lock")) - async def door_unlock(self): - return await self._vehicle._command("door_unlock") + async def door_unlock(self) -> bool: + return cast(bool, await self._vehicle._command("door_unlock")) diff --git a/tesla_api/datatypes.py b/tesla_api/datatypes.py new file mode 100644 index 0000000..8e4b808 --- /dev/null +++ b/tesla_api/datatypes.py @@ -0,0 +1,346 @@ +""" +A collection of objects used for typing across the library. + +Mostly this is TypedDict instances used to represent API responses. +""" + +from typing import List, Literal, Optional, TypedDict, Union + +# asleep seems to only happen when in service. +VehicleState = Literal["asleep", "offline", "online", "shutdown"] + + +class _BaseResponseBase(TypedDict): + response: object + + +class BaseResponse(_BaseResponseBase, total=False): + count: int + + +class CommandResponse(TypedDict): + reason: str + result: bool + + +class ErrorResponse(TypedDict): + response: None + error: str + error_description: Literal[""] + + +class ChargeStateResponse(TypedDict): # TODO + battery_heater_on: bool + battery_level: int # Percent + battery_range: float # TODO: Range in miles? + charge_current_request: int + charge_current_request_max: int + charge_enable_request: bool + charge_energy_added: float + charge_limit_soc: int + charge_limit_soc_max: int + charge_limit_soc_min: int + charge_limit_soc_std: int + charge_miles_added_ideal: float + charge_miles_added_rated: float + charge_port_cold_weather_mode: Optional[bool] + charge_port_door_open: bool + charge_port_latch: Literal["Blocking"] + charge_rate: float + charge_to_max_range: bool + charger_actual_current: int + charger_phases: None + charger_pilot_current: int + charger_power: int + charger_voltage: int + charging_state: Literal["Disconnected"] + conn_charge_cable: Literal[""] + est_battery_range: float + fast_charger_brand: Literal[""] + fast_charger_present: bool + fast_charger_type: Literal[""] + ideal_battery_range: float + managed_charging_active: bool + managed_charging_start_time: None + managed_charging_user_canceled: bool + max_range_charge_counter: int + minutes_to_full_charge: int # TODO + not_enough_power_to_heat: bool + scheduled_charging_pending: bool + scheduled_charging_start_time: None + time_to_full_charge: float # TODO + timestamp: int + trip_charging: bool + usable_battery_level: int + user_charge_enable_request: None + + +class ClimateStateResponse(TypedDict): # TODO + # The Optional attributes here are all None if the car is not awake. + battery_heater: bool + battery_heater_no_power: bool + climate_keeper_mode: Literal["camp", "dog", "off"] + defrost_mode: int + driver_temp_setting: float + fan_status: int + inside_temp: Optional[float] + is_auto_conditioning_on: Optional[bool] + is_climate_on: bool + is_front_defroster_on: bool + is_preconditioning: bool + is_rear_defroster_on: bool + left_temp_direction: Optional[int] + max_avail_temp: float + min_avail_temp: float + outside_temp: Optional[float] + passenger_temp_setting: float + remote_heater_control_enabled: bool + right_temp_direction: Optional[int] + seat_heater_left: int + seat_heater_rear_center: int + seat_heater_rear_left: int + seat_heater_rear_right: int + seat_heater_right: int + side_mirror_heaters: bool + steering_wheel_heater: bool + timestamp: int + wiper_blade_heater: bool + + +class DriveStateResponse(TypedDict): + # This differs from timestamp when the GNSS system has not got a signal yet. + gps_as_of: int # Seconds + heading: int # TODO + latitude: float + longitude: float + native_latitude: float + native_location_supported: int + native_longitude: float + native_type: Literal["wgs"] + power: int # TODO: KW? + shift_state: Literal[None, "D", "P", "R"] # TODO + speed: Optional[int] # TODO: mph? + timestamp: int # Milliseconds + + +class GUISettingsResponse(TypedDict): + gui_24_hour_time: bool + gui_charge_rate_units: Literal["kW"] # TODO + gui_distance_units: Literal["mi/hr", "km/hr"] + gui_range_display: Literal["Rated"] # TODO: 'Percent'? + gui_temperature_units: Literal["C", "F"] # TODO + show_range_units: bool # TODO + timestamp: int # Milliseconds + + +class TokenResponse(TypedDict): + access_token: str + token_type: Literal["bearer"] + expires_in: int # Seconds + refresh_token: str + created_at: int # Timestamp (seconds) + + +class _TokenParamsPassword(TypedDict): + grant_type: Literal["password"] + client_id: str + client_secret: str + email: str + password: str + + +class _TokenParamsRefresh(TypedDict): + grant_type: Literal["refresh_token"] + client_id: str + client_secret: str + refresh_token: str + + +TokenParams = Union[_TokenParamsPassword, _TokenParamsRefresh] + + +class VehicleConfigResponse(TypedDict): + can_accept_navigation_requests: bool + can_actuate_trunks: bool + car_special_type: Literal["base"] + car_type: Literal["models2"] + charge_port_type: Literal["EU", "US"] + ece_restrictions: bool # UNECE regulations (e.g. limit turning of steering wheel). + eu_vehicle: bool + exterior_color: Literal["Black"] + has_air_suspension: bool + has_ludicrous_mode: bool + motorized_charge_port: bool + plg: bool # ?? + rear_seat_heaters: int # TODO + rear_seat_type: int # 0 + rhd: bool # Right-hand drive + roof_color: Literal["Glass"] + seat_type: int # 2 = 2020 seats + spoiler_type: Literal["None"] + sun_roof_installed: Literal[0] # 0 = Not installed + third_row_seats: Literal["None"] + timestamp: int # Milliseconds + use_range_badging: bool # ?? + wheel_type: Literal["Slipstream19Carbon"] + + +class _VehicleStateMedia(TypedDict): + remote_control_enabled: bool + + +class _VehicleStateSoftwareUpdate(TypedDict): + download_perc: int + expected_duration_sec: int + install_perc: int + # '' = No firmware updates available (if duration is not 0, + # then other updates are available, such as a games update). + # available = Downloaded, ready to install. + # downloading_wifi_wait = Waiting for WiFi connection to resume download. + status: Literal["", "available", "downloading", "downloading_wifi_wait", "scheduled"] + # Empty string if no version available or not a firmware update. + version: str + + +class _VehicleStateSpeedLimit(TypedDict): + active: bool + current_limit_mph: float + max_limit_mph: int + min_limit_mph: int + pin_code_set: bool + + +class _VehicleStateResponseBase(TypedDict): + api_version: int + autopark_state_v2: Literal["standby", "ready", "unavailable"] + calendar_supported: bool + car_version: str # Software version + # 0 = off, 2,3,4 = ?? + center_display_state: int + df: int # ?? + dr: int # ?? + ft: int # ?? + is_user_present: bool + locked: bool + media_state: _VehicleStateMedia + notifications_supported: bool + odometer: float # TODO: Miles + parsed_calendar_supported: bool + pf: int # ?? + pr: int # ?? + remote_start: bool # Currently activated + remote_start_enabled: bool # Available to activate via the API + remote_start_supported: bool + rt: int # ?? + sentry_mode: bool + sentry_mode_available: bool # This will be False when driving etc. + software_update: _VehicleStateSoftwareUpdate + speed_limit_mode: _VehicleStateSpeedLimit + timestamp: int # milliseconds + valet_mode: bool + valet_pin_needed: bool + vehicle_name: Optional[str] + + +class VehicleStateResponse(_VehicleStateResponseBase, total=False): + autopark_style: Literal["dead_man"] + homelink_device_count: int + homelink_nearby: bool + last_autopark_error: Literal["no_error"] + smart_summon_available: bool + summon_standby_mode_enabled: bool + + +class _VehiclesIdResponseBase(TypedDict): + id: int + vehicle_id: int + vin: str + display_name: str # ?? + # No longer correct. https://tesla-api.timdorr.com/vehicle/optioncodes + option_codes: str + color: None # ?? + # Seems to hold 2 tokens, with a new one pushing out the older one every few minutes. + tokens: List[str] # ?? + state: VehicleState + in_service: bool + id_s: str # String version of id. + calendar_enabled: bool + access_type: Literal["OWNER"] + api_version: int + backseat_token: None # ?? + backseat_token_updated_at: None # ?? + + +class VehiclesIdResponse(_VehiclesIdResponseBase, total=False): + user_id: int + + +class VehicleDataResponse(VehiclesIdResponse): + drive_state: DriveStateResponse + climate_state: ClimateStateResponse + charge_state: ChargeStateResponse + gui_settings: GUISettingsResponse + vehicle_state: VehicleStateResponse + vehicle_config: VehicleConfigResponse + + +VehiclesResponse = List[VehiclesIdResponse] + + +# Energy sites + +class EnergySite(TypedDict): + energy_site_id: int + resource_type: Literal["solar"] + site_name: str + id: int + solar_power: int + sync_grid_alert_enabled: bool + breaker_alert_enabled: bool + + +class _EnergySiteInfoUserSettings(TypedDict): + storm_mode_enabled: None + sync_grid_alert_enabled: bool + breaker_alert_enabled: bool + + +class _EnergySiteInfoComponents(TypedDict): + solar: bool + battery: bool + grid: bool + backup: bool + gateway: str + load_meter: bool + tou_capable: bool + storm_mode_capable: bool + flex_energy_request_capable: bool + car_charging_data_supported: bool + configurable: bool + grid_services_enabled: bool + + +class EnergySiteInfoResponse(TypedDict): + id: str + site_name: str + installation_date: str + user_settings: _EnergySiteInfoUserSettings + components: _EnergySiteInfoComponents + backup_reserve_percent: int # TODO + default_real_mode: str + version: str + battery_count: int + time_zone_offset: int + + +class EnergySiteLiveStatusResponse(TypedDict): + solar_power: int + grid_status: str + grid_services_active: bool + percentage_charged: int + energy_left: float + total_pack_energy: int + timestamp: str + + +ProductsResponse = List[Union[VehiclesIdResponse, EnergySite]] diff --git a/tesla_api/energy.py b/tesla_api/energy.py index a5a7eac..0db481b 100644 --- a/tesla_api/energy.py +++ b/tesla_api/energy.py @@ -24,56 +24,63 @@ # SOFTWARE. from datetime import date, datetime, time -from typing import Optional, Union +from typing import Any, Dict, Literal, Optional, TYPE_CHECKING, Union, cast + +from .datatypes import EnergySiteInfoResponse, EnergySiteLiveStatusResponse + +if TYPE_CHECKING: + from . import TeslaApiClient class Energy: - def __init__(self, api_client, energy_site_id): + def __init__(self, api_client: "TeslaApiClient", energy_site_id: int): self._api_client = api_client self._energy_site_id = energy_site_id @property - def site_id(self): + def site_id(self) -> int: return self._energy_site_id - async def get_energy_site_info(self): - return await self._api_client.get( - "energy_sites/{}/site_info".format(self._energy_site_id)) + async def get_energy_site_info(self) -> EnergySiteInfoResponse: + endpoint = "energy_sites/{}/site_info".format(self._energy_site_id) + return cast(EnergySiteInfoResponse, await self._api_client.get(endpoint)) # Helper functions for get_energy_site_info - async def get_backup_reserve_percent(self): + async def get_backup_reserve_percent(self) -> int: info = await self.get_energy_site_info() - return int(info["backup_reserve_percent"]) + return int(info["backup_reserve_percent"]) # TODO: remove int? - async def get_operating_mode(self): + async def get_operating_mode(self) -> str: info = await self.get_energy_site_info() return info["default_real_mode"] - async def get_version(self): + async def get_version(self) -> str: info = await self.get_energy_site_info() return info["version"] - async def get_battery_count(self): + async def get_battery_count(self) -> int: info = await self.get_energy_site_info() - return int(info["battery_count"]) + return int(info["battery_count"]) # TODO - async def get_energy_site_calendar_history_data( - self, kind="energy", period="day", - end_date: Optional[Union[str, date]] = None) -> dict: + # TODO: Find out what return type is for this endpoint and add to datatypes. + async def get_energy_site_calendar_history_data( # type: ignore[misc] + self, kind: Literal["power", "energy", "self_consumption"] = "energy", + period: Literal["day", "week", "month", "year", "lifetime"] = "day", + end_date: Optional[Union[str, date]] = None) -> Dict[str, Any]: """Return historical energy data. Args: kind: [power, energy, self_consumption] period: Amount of time to include in report. One of day, week, month, year, - and lifetime. When kind is 'power', this parameter is ignored, and the - period is always 'day'. + and lifetime. When kind is "power", this parameter is ignored, and the + period is always "day". end_date: A date/datetime object, or a str in ISO 8601 format (e.g. 2019-12-23T17:39:18.546Z). The response report interval ends at this datetime and starts at the beginning of the given period. For example, with datetime(year=2020, month=5, day=1), this gets all data for May 1st. Defaults to the current time. """ - params = {"kind": kind, "period": period} + params: Dict[str, str] = {"kind": kind, "period": period} if isinstance(end_date, date): if not isinstance(end_date, datetime): @@ -89,57 +96,58 @@ async def get_energy_site_calendar_history_data( if end_date is not None: params["end_date"] = end_date - return await self._api_client.get( - "energy_sites/{}/calendar_history".format(self._energy_site_id), params=params) + endpoint = "energy_sites/{}/calendar_history".format(self._energy_site_id) + return cast(Dict[str, Any], # type: ignore[misc] + await self._api_client.get(endpoint, params=params)) - async def get_energy_site_live_status(self): - return await self._api_client.get( - "energy_sites/{}/live_status".format(self._energy_site_id)) + async def get_energy_site_live_status(self) -> EnergySiteLiveStatusResponse: + endpoint = "energy_sites/{}/live_status".format(self._energy_site_id) + return cast(EnergySiteLiveStatusResponse, await self._api_client.get(endpoint)) # Helper functions for get_energy_site_live_status - async def get_energy_site_live_status_percentage_charged(self): + async def get_energy_site_live_status_percentage_charged(self) -> int: status = await self.get_energy_site_live_status() return int(status["percentage_charged"]) - async def get_energy_site_live_status_energy_left(self): + async def get_energy_site_live_status_energy_left(self) -> float: status = await self.get_energy_site_live_status() return float(status["energy_left"]) - async def get_energy_site_live_status_total_pack_energy(self): + async def get_energy_site_live_status_total_pack_energy(self) -> int: status = await self.get_energy_site_live_status() return int(status["total_pack_energy"]) - async def get_solar_power(self): + async def get_solar_power(self) -> int: status = await self.get_energy_site_live_status() - return int(status["solar_power"]) + return status["solar_power"] # Setting of the backup_reserve_percent used in self_consumption # (i.e. self-powered mode). # On my Powerwall 2, setting backup_reserve_percent > energy_left # causes the battery to charge at 1.7kW - async def set_backup_reserve_percent(self, backup_reserve_percent): + async def set_backup_reserve_percent(self, backup_reserve_percent: int) -> bool: assert 0 <= backup_reserve_percent <= 100 - return await self._api_client.post( - endpoint="energy_sites/{}/backup".format(self._energy_site_id), - data={"backup_reserve_percent": backup_reserve_percent}, - ) + endpoint = "energy_sites/{}/backup".format(self._energy_site_id) + args = {"backup_reserve_percent": backup_reserve_percent} + return cast(bool, await self._api_client.post(endpoint, args)) # Correspondence between mode names and the Tesla app: # mode = 'self_consumption' = "self-powered" on app # mode = 'backup' = "backup-only" on app # mode = 'autonomous' = "Advanced - Time-based control" on app # Note: setting 'backup' mode causes my Powerwall 2 to charge at 3.4kW - async def set_operating_mode(self, mode): - return await self._api_client.post( - endpoint="energy_sites/{}/operation".format(self._energy_site_id), - data={"default_real_mode": mode}) + async def set_operating_mode( + self, mode: Literal["self_consumption", "backup", "autonomous"]) -> bool: + endpoint = "energy_sites/{}/operation".format(self._energy_site_id) + args = {"default_real_mode": mode} + return cast(bool, await self._api_client.post(endpoint, args)) # helper functions for set_operating_mode - async def set_operating_mode_self_consumption(self): + async def set_operating_mode_self_consumption(self) -> bool: return await self.set_operating_mode("self_consumption") - async def set_operating_mode_backup(self): + async def set_operating_mode_backup(self) -> bool: return await self.set_operating_mode("backup") - async def set_operating_mode_autonomous(self): + async def set_operating_mode_autonomous(self) -> bool: return await self.set_operating_mode("autonomous") diff --git a/tesla_api/exceptions.py b/tesla_api/exceptions.py index 31d4956..06b405d 100644 --- a/tesla_api/exceptions.py +++ b/tesla_api/exceptions.py @@ -1,14 +1,22 @@ +from aiohttp.client_exceptions import ClientConnectionError as ClientError # noqa: F401 + + class AuthenticationError(Exception): - def __init__(self, error): + def __init__(self, error: object): super().__init__("Authentication to the Tesla API failed: {}".format(error)) class ApiError(Exception): - def __init__(self, error): + def __init__(self, error: object): super().__init__("Tesla API call failed: {}".format(error)) self.reason = error class VehicleUnavailableError(Exception): - def __init__(self): + def __init__(self) -> None: super().__init__("Vehicle failed to wake up.") + + +class VehicleInServiceError(VehicleUnavailableError): + def __init__(self) -> None: + Exception.__init__(self, "Vehicle is currently in service.") diff --git a/tesla_api/py.typed b/tesla_api/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/tesla_api/vehicle.py b/tesla_api/vehicle.py index fc34532..2fafc29 100644 --- a/tesla_api/vehicle.py +++ b/tesla_api/vehicle.py @@ -1,13 +1,36 @@ import asyncio +from typing import Any, Iterable, List, Mapping, Optional, TYPE_CHECKING, cast from .charge import Charge from .climate import Climate from .controls import Controls +from .datatypes import ( + CommandResponse, DriveStateResponse, GUISettingsResponse, VehicleDataResponse, + VehicleState, VehicleStateResponse, VehiclesIdResponse) from .exceptions import ApiError, VehicleUnavailableError +if TYPE_CHECKING: + from . import TeslaApiClient + class Vehicle: - def __init__(self, api_client, vehicle): + id: int + user_id: Optional[int] # Only available when car is awake. + vehicle_id: int + vin: str + display_name: str + option_codes: str + color: None + tokens: List[str] + state: VehicleState + in_service: bool + id_s: str + calendar_enabled: bool + api_version: int + backseat_token: None + backseat_token_updated_at: None + + def __init__(self, api_client: "TeslaApiClient", vehicle: VehiclesIdResponse): self._api_client = api_client self._vehicle = vehicle @@ -15,7 +38,9 @@ def __init__(self, api_client, vehicle): self.climate = Climate(self) self.controls = Controls(self) - async def _command(self, command_endpoint, data=None, _retry=True): # noqa: C901 + async def _command(self, command_endpoint: str, # noqa: C901 + data: Optional[Mapping[str, object]] = None, + _retry: bool = True) -> None: """Handles vehicle commands with the common reason/result response. Args: @@ -31,43 +56,51 @@ async def _command(self, command_endpoint, data=None, _retry=True): # noqa: C90 endpoint = "vehicles/{}/command/{}".format(self.id, command_endpoint) try: - res = await self._api_client.post(endpoint, data) + res = cast(CommandResponse, await self._api_client.post(endpoint, data)) except VehicleUnavailableError: # If first attempt, retry with a wake up. if _retry: self._vehicle["state"] = "offline" - return await self._command(command_endpoint, data, _retry=False) + await self._command(command_endpoint, data, _retry=False) raise if res.get("result") is not True: raise ApiError(res.get("reason", "")) - def _update_vehicle(self, state): + def _update_vehicle(self, state: VehiclesIdResponse) -> None: self._vehicle = state + # Ensure user_id is set if car is not awake. + self._vehicle.setdefault("user_id", None) # type: ignore[arg-type,misc] + if self._api_client.callback_update is not None: asyncio.create_task(self._api_client.callback_update(self)) - async def is_mobile_access_enabled(self): - return await self._api_client.get("vehicles/{}/mobile_enabled".format(self.id)) + async def is_mobile_access_enabled(self) -> bool: + return cast(bool, await self._api_client.get("vehicles/{}/mobile_enabled".format(self.id))) + + async def get_data(self) -> VehicleDataResponse: + endpoint = "vehicles/{}/vehicle_data".format(self.id) + data = cast(VehicleDataResponse, await self._api_client.get(endpoint)) + + vehicle = cast(VehiclesIdResponse, {k: v for k, v in data.items() + if not isinstance(v, dict)}) + self._update_vehicle(vehicle) - async def get_data(self): - data = await self._api_client.get("vehicles/{}/vehicle_data".format(self.id)) - self._update_vehicle({k: v for k, v in data.items() if not isinstance(v, dict)}) return data - async def get_state(self): - return await self._api_client.get( - "vehicles/{}/data_request/vehicle_state".format(self.id)) + async def get_state(self) -> VehicleStateResponse: + endpoint = "vehicles/{}/data_request/vehicle_state".format(self.id) + return cast(VehicleStateResponse, await self._api_client.get(endpoint)) - async def get_drive_state(self): - return await self._api_client.get( - "vehicles/{}/data_request/drive_state".format(self.id)) + async def get_drive_state(self) -> DriveStateResponse: + endpoint = "vehicles/{}/data_request/drive_state".format(self.id) + return cast(DriveStateResponse, await self._api_client.get(endpoint)) - async def get_gui_settings(self): - return await self._api_client.get( - "vehicles/{}/data_request/gui_settings".format(self.id)) + async def get_gui_settings(self) -> GUISettingsResponse: + endpoint = "vehicles/{}/data_request/gui_settings".format(self.id) + return cast(GUISettingsResponse, await self._api_client.get(endpoint)) - async def wake_up(self, timeout=-1): # noqa: C901 + async def wake_up(self, timeout: Optional[float] = -1) -> None: # noqa: C901 """Attempt to wake up the car. Vehicle will be online when this function returns successfully. @@ -79,6 +112,7 @@ async def wake_up(self, timeout=-1): # noqa: C901 Raises: VehicleUnavailableError: Timeout exceeded without success. """ + delay: float if timeout is None: delay = 2 else: @@ -86,12 +120,13 @@ async def wake_up(self, timeout=-1): # noqa: C901 timeout = self._api_client.timeout delay = timeout / 100 - async def _wake(): - state = await self._api_client.post("vehicles/{}/wake_up".format(self.id)) + async def _wake() -> None: + endpoint = "vehicles/{}/wake_up".format(self.id) + state = cast(VehiclesIdResponse, await self._api_client.post(endpoint)) self._update_vehicle(state) while self._vehicle["state"] != "online": await asyncio.sleep(delay) - state = await self._api_client.post("vehicles/{}/wake_up".format(self.id)) + state = cast(VehiclesIdResponse, await self._api_client.post(endpoint)) self._update_vehicle(state) if self._api_client.callback_wake_up is not None: @@ -102,24 +137,26 @@ async def _wake(): except asyncio.TimeoutError: raise VehicleUnavailableError() - async def remote_start(self, password): + async def remote_start(self, password: str) -> bool: """Enable keyless driving (must start car within a 2 minute window). password - The account password to reauthenticate. """ - return await self._command("remote_start_drive", data={"password": password}) + return cast(bool, await self._command("remote_start_drive", {"password": password})) - async def update(self): - self._update_vehicle(await self._api_client.get("vehicles/{}".format(self.id))) + async def update(self) -> None: + endpoint = "vehicles/{}".format(self.id) + vehicle = cast(VehiclesIdResponse, await self._api_client.get(endpoint)) + self._update_vehicle(vehicle) - def __dir__(self): + def __dir__(self) -> Iterable[str]: """Include _vehicle keys in dir(), which are accessible with __getattr__().""" return super().__dir__() | self._vehicle.keys() - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: # type: ignore[misc] """Allow attribute access to _vehicle details.""" try: - return self._vehicle[name] + return self._vehicle[name] # type: ignore[misc] except KeyError: raise AttributeError( "'{}' object has no attribute '{}'".format(self.__class__.__name__, name))