Skip to content

Commit

Permalink
Rename enums and add level
Browse files Browse the repository at this point in the history
  • Loading branch information
Bre77 committed Feb 8, 2024
1 parent ac1e90a commit 35428a0
Show file tree
Hide file tree
Showing 12 changed files with 218 additions and 194 deletions.
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

setuptools.setup(
name="tesla_fleet_api",
version="0.3.2",
version="0.4.0",
author="Brett Adams",
author_email="hello@teslemetry.com",
description="Tesla Fleet API library for Python",
Expand Down
8 changes: 3 additions & 5 deletions tesla_fleet_api/charging.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
from typing import Any
from .const import (
Methods,
)
from .const import Method


class Charging:
Expand All @@ -22,7 +20,7 @@ async def history(
) -> dict[str, Any]:
"""Returns the paginated charging history."""
return await self._request(
Methods.GET,
Method.GET,
"api/1/dx/charging/history",
{
"vin": vin,
Expand All @@ -45,7 +43,7 @@ async def sessions(
) -> dict[str, Any]:
"""Returns the charging session information including pricing and energy data. This endpoint is only available for business accounts that own a fleet of vehicles."""
return await self._request(
Methods.GET,
Method.GET,
"api/1/dx/charging/sessions",
{
"vin": vin,
Expand Down
49 changes: 34 additions & 15 deletions tesla_fleet_api/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,22 +3,22 @@


class IntEnum(int, Enum):
pass
"""Integer Enum."""


class StrEnum(str, Enum):
pass
"""String Enum."""


class Methods(StrEnum):
"""HTTP methods."""
class Method(StrEnum):
"""HTTP Methods."""

GET = "GET"
POST = "POST"
DELETE = "DELETE"


class Errors(StrEnum):
class Error(StrEnum):
"""Tesla Fleet API error codes."""

INVALID_COMMAND = "invalid_command"
Expand All @@ -39,13 +39,32 @@ class Errors(StrEnum):
}


class Trunks(StrEnum):
class Trunk(StrEnum):
"""Trunk options"""

FRONT: "front"
REAR: "rear"


class Seat(IntEnum):
"""Seat positions"""

FRONT_LEFT = 0
FRONT_RIGHT = 1
REAR_LEFT = 2
REAR_CENTER = 4
REAR_RIGHT = 5


class Level(IntEnum):
"""Level options"""

OFF = 0
LOW = 1
MEDIUM = 2
HIGH = 3


class ClimateKeeperMode(IntEnum):
"""Climate Keeper Mode options"""

Expand All @@ -55,15 +74,15 @@ class ClimateKeeperMode(IntEnum):
CAMP_MODE = 3


class CabinOverheatProtectionTemps(IntEnum):
class CabinOverheatProtectionTemp(IntEnum):
"""COP Temp options"""

LOW = 0 # 30C 90F
MEDIUM = 1 # 35C 95F
HIGH = 2 # 40C 100F


class VehicleDataEndpoints(StrEnum):
class VehicleDataEndpoint(StrEnum):
"""Endpoints options"""

CHARGE_STATE = "charge_state"
Expand All @@ -77,22 +96,22 @@ class VehicleDataEndpoints(StrEnum):
VEHICLE_DATA_COMBO = "vehicle_data_combo"


class SunRoofCommands(StrEnum):
class SunRoofCommand(StrEnum):
"""Sunroof options"""

STOP = "stop"
CLOSE = "close"
VENT = "vent"


class WindowCommands(StrEnum):
class WindowCommand(StrEnum):
"""Window Control options"""

VENT = "vent"
CLOSE = "close"


class DeviceTypes(StrEnum):
class DeviceType(StrEnum):
"""Device Type options"""

ANDROID = "android"
Expand All @@ -102,8 +121,8 @@ class DeviceTypes(StrEnum):
IOS_PRODUCTION = "ios-production"


class Scopes(StrEnum):
"""Fleet API Scopes"""
class Scope(StrEnum):
"""Fleet API Scope"""

OPENID = "openid"
OFFLINE_ACCESS = "offline_access"
Expand All @@ -115,7 +134,7 @@ class Scopes(StrEnum):
ENERGY_CMDS = "energy_cmds"


class TelemetryFields(StrEnum):
class TelemetryField(StrEnum):
"""Fields available in telemetry streams"""

AC_CHARGING_ENERGY_IN = "ACChargingEnergyIn"
Expand Down Expand Up @@ -270,7 +289,7 @@ class TelemetryFields(StrEnum):
VERSION = "Version"


class TelemetryAlerts(StrEnum):
class TelemetryAlert(StrEnum):
"""Alerts available in telemetry streams"""

CUSTOMER = "Customer"
Expand Down
22 changes: 11 additions & 11 deletions tesla_fleet_api/energy.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from typing import Any
from .const import Methods
from .const import Method
from .energyspecific import EnergySpecific


Expand All @@ -18,7 +18,7 @@ async def backup(
) -> dict[str, Any]:
"""Adjust the site's backup reserve."""
return await self._request(
Methods.POST,
Method.POST,
f"api/1/energy_sites/{energy_site_id}/backup",
json={"backup_reserve_percent": backup_reserve_percent},
)
Expand All @@ -34,7 +34,7 @@ async def backup_history(
) -> dict[str, Any]:
"""Returns the backup (off-grid) event history of the site in duration of seconds."""
return await self._request(
Methods.GET,
Method.GET,
f"api/1/energy_sites/{energy_site_id}/calendar_history",
json={
"kind": kind,
Expand All @@ -55,7 +55,7 @@ async def charge_history(
) -> dict[str, Any]:
"""Returns the charging history of a wall connector."""
return await self._request(
Methods.GET,
Method.GET,
f"api/1/energy_sites/{energy_site_id}/telemetry_history",
query={
"kind": kind,
Expand All @@ -76,7 +76,7 @@ async def energy_history(
) -> dict[str, Any]:
"""Returns the energy measurements of the site, aggregated to the requested period."""
return await self._request(
Methods.GET,
Method.GET,
f"api/1/energy_sites/{energy_site_id}/calendar_history",
query={
"kind": kind,
Expand Down Expand Up @@ -106,15 +106,15 @@ async def grid_import_export(
"At least one of disallow_charge_from_grid_with_solar_installed or customer_preferred_export_rule must be set."
)
return await self._request(
Methods.POST,
Method.POST,
f"api/1/energy_sites/{energy_site_id}/grid_import_export",
json=data,
)

async def live_status(self, energy_site_id: int) -> dict[str, Any]:
"""Returns the live status of the site (power, state of energy, grid status, storm mode)."""
return await self._request(
Methods.GET,
Method.GET,
f"api/1/energy_sites/{energy_site_id}/live_status",
)

Expand All @@ -123,7 +123,7 @@ async def off_grid_vehicle_charging_reserve(
) -> dict[str, Any]:
"""Adjust the site's off-grid vehicle charging backup reserve."""
return await self._request(
Methods.POST,
Method.POST,
f"api/1/energy_sites/{energy_site_id}/off_grid_vehicle_charging_reserve",
json={
"off_grid_vehicle_charging_reserve_percent": off_grid_vehicle_charging_reserve_percent
Expand All @@ -135,22 +135,22 @@ async def operation(
) -> dict[str, Any]:
"""Set the site's mode."""
return await self._request(
Methods.POST,
Method.POST,
f"api/1/energy_sites/{energy_site_id}/operation",
json={"default_real_mode": default_real_mode},
)

async def site_info(self, energy_site_id: int) -> dict[str, Any]:
"""Returns information about the site. Things like assets (has solar, etc), settings (backup reserve, etc), and features (storm_mode_capable, etc)."""
return await self._request(
Methods.GET,
Method.GET,
f"api/1/energy_sites/{energy_site_id}/site_info",
)

async def storm_mode(self, energy_site_id: int, enabled: bool) -> dict[str, Any]:
"""Update storm watch participation."""
return await self._request(
Methods.POST,
Method.POST,
f"api/1/energy_sites/{energy_site_id}/storm_mode",
json={"enabled": enabled},
)
22 changes: 11 additions & 11 deletions tesla_fleet_api/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import aiohttp
from .const import Errors
from .const import Error


class TeslaFleetError(BaseException):
Expand Down Expand Up @@ -87,9 +87,9 @@ class PaymentRequired(TeslaFleetError):


class Forbidden(TeslaFleetError):
"""Access to this resource is not authorized, developers should check required scopes."""
"""Access to this resource is not authorized, developers should check required Scope."""

message = "Access to this resource is not authorized, developers should check required scopes."
message = "Access to this resource is not authorized, developers should check required Scope."
status = 403


Expand Down Expand Up @@ -232,25 +232,25 @@ async def raise_for_status(resp: aiohttp.ClientResponse) -> None:
except aiohttp.ClientResponseError as e:
if resp.status == 400:
error = data.get("error")
if error == Errors.INVALID_COMMAND:
if error == Error.INVALID_COMMAND:
raise InvalidCommand(data) from e
elif error == Errors.INVALID_FIELD:
elif error == Error.INVALID_FIELD:
raise InvalidField(data) from e
elif error == Errors.INVALID_REQUEST:
elif error == Error.INVALID_REQUEST:
raise InvalidRequest(data) from e
elif error == Errors.INVALID_AUTH_CODE:
elif error == Error.INVALID_AUTH_CODE:
raise InvalidAuthCode(data) from e
elif error == Errors.INVALID_REDIRECT_URL:
elif error == Error.INVALID_REDIRECT_URL:
raise InvalidRedirectUrl(data) from e
elif error == Errors.UNAUTHORIZED_CLIENT:
elif error == Error.UNAUTHORIZED_CLIENT:
raise UnauthorizedClient(data) from e
else:
raise TeslaFleetError(data) from e
elif resp.status == 401:
error = data.get("error")
if error == Errors.TOKEN_EXPIRED:
if error == Error.TOKEN_EXPIRED:
raise OAuthExpired(data) from e
elif error == Errors.MOBILE_ACCESS_DISABLED:
elif error == Error.MOBILE_ACCESS_DISABLED:
raise MobileAccessDisabled(data) from e
else:
raise InvalidToken(data) from e
Expand Down
8 changes: 4 additions & 4 deletions tesla_fleet_api/partner.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from typing import Any
from .const import Methods
from .const import Method


class Partner:
Expand All @@ -11,17 +11,17 @@ def __init__(self, parent):
async def public_key(self, domain: str | None = None) -> dict[str, Any]:
"""Returns the public key associated with a domain. It can be used to ensure the registration was successful."""
return await self._request(
Methods.GET, "api/1/partner_accounts/public_key", params={"domain": domain}
Method.GET, "api/1/partner_accounts/public_key", params={"domain": domain}
)

async def register(self, domain: str) -> dict[str, Any]:
"""Registers an existing account before it can be used for general API access. Each application from developer.tesla.com must complete this step."""
return await self._request(
Methods.POST, "api/1/partner_accounts", json={"domain": domain}
Method.POST, "api/1/partner_accounts", json={"domain": domain}
)

async def fleet_telemetry_errors(self) -> dict[str, Any]:
"""Returns recent fleet telemetry errors reported by vehicles after receiving the config."""
return await self._request(
Methods.GET, "api/1/fleet_telemetry/fleet_telemetry_errors"
Method.GET, "api/1/fleet_telemetry/fleet_telemetry_errors"
)
10 changes: 5 additions & 5 deletions tesla_fleet_api/teslafleetapi.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import aiohttp
from .exceptions import raise_for_status, InvalidRegion, LibraryError
from typing import Any
from .const import SERVERS, Methods, Errors
from .const import SERVERS, Method, Error
from .charging import Charging
from .energy import Energy
from .partner import Partner
Expand Down Expand Up @@ -67,7 +67,7 @@ async def find_server(self) -> str:

async def _request(
self,
method: Methods,
method: Method,
path: str,
params: dict[str:Any] | None = None,
json: dict[str:Any] | None = None,
Expand All @@ -77,7 +77,7 @@ async def _request(
if not self.server:
raise ValueError("Server was not set at init. Call find_server() first.")

if method == Methods.GET and json is not None:
if method == Method.GET and json is not None:
raise ValueError("GET requests cannot have a body.")

if params:
Expand All @@ -101,7 +101,7 @@ async def _request(
# Manufacture a response since Tesla doesn't provide a body for token expiration.
return {
"response": None,
"error": Errors.INVALID_TOKEN,
"error": Error.INVALID_TOKEN,
"error_message": "The OAuth token has expired.",
}
if resp.content_type == "application/json":
Expand All @@ -118,6 +118,6 @@ async def status(self):
async def products(self) -> dict[str, Any]:
"""Returns products mapped to user."""
return await self._request(
Methods.GET,
Method.GET,
"api/1/products",
)
Loading

0 comments on commit 35428a0

Please sign in to comment.