Metadata-Version: 2.4
Name: tesla_fleet_api
Version: 1.5.2
Summary: Tesla Fleet API library for Python
Author-email: Brett Adams <hello@teslemetry.com>
License-Expression: Apache-2.0
Project-URL: Homepage, https://github.com/Teslemetry/python-tesla-fleet-api
Classifier: Development Status :: 5 - Production/Stable
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Operating System :: OS Independent
Requires-Python: >=3.13
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiohttp>=3
Requires-Dist: aiofiles>=24
Requires-Dist: aiolimiter>=1
Requires-Dist: cryptography>=43
Requires-Dist: protobuf>=6.32.0
Requires-Dist: bleak>=0.22
Requires-Dist: bleak-retry-connector>=3.9
Dynamic: license-file

# Tesla Fleet API

Tesla Fleet API is a Python library that provides an interface to interact with Tesla's Fleet API, including signed commands and encrypted local Bluetooth (BLE) communication. It also supports interactions with Teslemetry and Tessie services.

## Features

- Fleet API for vehicles
- Fleet API for energy sites
- Fleet API with signed vehicle commands
- Bluetooth for vehicles
- Routing and failover across backends for vehicles and energy sites (e.g. Bluetooth/local primary, cloud fallback)
- Teslemetry integration
- Tessie integration

## Installation

You can install the library using pip:

```bash
pip install tesla-fleet-api
```

## Usage

### Authentication

The `TeslaFleetOAuth` class provides methods that help with authenticating to the Tesla Fleet API. Here's a basic example:

```python
import asyncio
import aiohttp
from tesla_fleet_api import TeslaFleetOAuth

async def main():
    async with aiohttp.ClientSession() as session:
        oauth = TeslaFleetOAuth(
            session=session,
            client_id="<client_id>",
            client_secret="<client_secret>",
            redirect_uri="<redirect_uri>",
        )

        # Get the login URL and navigate the user to it
        login_url = oauth.get_login_url(scopes=["openid", "email", "offline_access"])
        print(f"Please go to {login_url} and authorize access.")

        # After the user authorizes access, they will be redirected to the redirect_uri with a code
        code = input("Enter the code you received: ")

        # Exchange the code for a refresh token
        await oauth.get_refresh_token(code)
        print(f"Access token: {oauth.access_token}")
        print(f"Refresh token: {oauth.refresh_token}")
        # Dont forget to store the refresh token so you can use it again later

asyncio.run(main())
```

### Fleet API for Vehicles

The `TeslaFleetApi` class provides methods to interact with the Fleet API for vehicles. Here's a basic example:

```python
import asyncio
import aiohttp
from tesla_fleet_api import TeslaFleetApi
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = TeslaFleetApi(
            access_token="<access_token>",
            session=session,
            region="na",
        )

        try:
            data = await api.vehicles.list()
            print(data)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())
```

For more detailed examples, see [Fleet API for Vehicles](docs/fleet_api_vehicles.md).

### Fleet API for Energy Sites

The `EnergySites` class provides methods to interact with the Fleet API for energy sites. Here's a basic example:

```python
import asyncio
import aiohttp
from tesla_fleet_api import TeslaFleetApi
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = TeslaFleetApi(
            access_token="<access_token>",
            session=session,
            region="na",
        )

        try:
            energy_sites = await api.energySites.list()
            print(energy_sites)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())
```

For more detailed examples, see [Fleet API for Energy Sites](docs/fleet_api_energy_sites.md).

### Fleet API with Signed Vehicle Commands

The `VehicleSigned` class provides methods to interact with the Fleet API using signed vehicle commands. Here's a basic example:

```python
import asyncio
import aiohttp
from tesla_fleet_api import TeslaFleetApi
from tesla_fleet_api.tesla.vehicle.signed import VehicleSigned
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = TeslaFleetApi(
            access_token="<access_token>",
            session=session,
            region="na",
        )

        try:
            vehicle = VehicleSigned(api, "<vin>")
            data = await vehicle.wake_up()
            print(data)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())
```

For more detailed examples, see [Fleet API with Signed Vehicle Commands](docs/fleet_api_signed_commands.md).

### Bluetooth for Vehicles

The `TeslaBluetooth` class provides methods to interact with Tesla vehicles using Bluetooth. Here's a basic example:

```python
import asyncio
from bleak import BleakScanner
from tesla_fleet_api import TeslaBluetooth

async def main():
    scanner = BleakScanner()
    devices = await scanner.discover()
    for device in devices:
        if TeslaBluetooth().valid_name(device.name):
            print(f"Found Tesla vehicle: {device.name}")

asyncio.run(main())
```

For more detailed examples, see [Bluetooth for Vehicles](docs/bluetooth_vehicles.md).

### Routing and Failover

The `Router` class composes an ordered list of two-or-more backends that share a common method surface and dispatches each method call down the chain, automatically failing over on error. `VehicleRouter` and `EnergySiteRouter` are thin entity-specific subclasses. A common setup is a local `VehicleBluetooth` primary with a cloud fallback (e.g. a `TeslemetryVehicle`), so commands go over Bluetooth when the vehicle is reachable and route to the cloud otherwise:

```python
import asyncio
import aiohttp
from tesla_fleet_api import TeslaBluetooth, Teslemetry
from tesla_fleet_api.tesla.router import VehicleRouter
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        # Primary: local Bluetooth
        tesla_bluetooth = TeslaBluetooth()
        await tesla_bluetooth.get_private_key("path/to/private_key.pem")
        primary = tesla_bluetooth.vehicles.create("<vin>")

        # Fallback: Teslemetry cloud
        teslemetry = Teslemetry(access_token="<access_token>", session=session)
        fallback = teslemetry.vehicles.create("<vin>")

        vehicle = VehicleRouter(primary, fallback)

        try:
            await vehicle.wake_up()
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())
```

The constructor is `Router(primary, fallback, *more_backends, health=None)`; the two-argument form shown above is fully backward compatible, and any number of extra backends may follow to extend the chain. Each call is tried on the first backend that has the method and, on any exception, retried on the next backend that has it, returning the first success (raising the last error only if every applicable backend fails). Non-callable attributes (e.g. `vin`) resolve to the first backend that has them.

By default the router attempts the primary and fails over on any error, with no up-front probe. You can also pass an explicit `health` check — a `bool`, a sync callable, or an async callable returning `bool` — to decide up front whether to route to the primary or skip straight to the rest of the chain. The health check gates **only the primary** (the first backend); later backends are reached purely through per-command failover.

`EnergySiteRouter` follows the same pattern for energy sites, pairing a duck-typed local `EnergySite`-shaped object (e.g. aiopowerwall's `PowerwallEnergySite`, no dependency added) with a cloud `TeslemetryEnergySite` fallback:

```python
from tesla_fleet_api.tesla.router import EnergySiteRouter

router = EnergySiteRouter(local_energysite, teslemetry_energysite)
await router.set_operation(...)  # local first, cloud on failure
```

`Router`, `VehicleRouter`, and `EnergySiteRouter` are all importable from `tesla_fleet_api.tesla.router` (and from `tesla_fleet_api.tesla`).

> **Warning:** Because a failed call is replayed on the next backend, a non-idempotent command (e.g. `honk_horn`, `actuate_trunk`, `door_unlock`, `charge_start`) that fails _mid-flight_ — after a backend may have already partially applied it — can be **double-executed** (or executed more than once across a longer chain) when it is retried on the next backend. This is a deliberate tradeoff of per-command failover. Callers needing exactly-once semantics for such commands should gate dispatch with an explicit `health` check or call the underlying backends directly.
>
> Dispatch is implemented via `__getattr__`, which does not proxy dunder methods, so `async with Router(...)` does **not** manage a backend's BLE connection lifecycle (`__aenter__`/`__aexit__`). Commands still auto-connect on send; for explicit connect/disconnect reach through `router.primary` (or `router.backends`).

### Teslemetry

The `Teslemetry` class provides methods to interact with the Teslemetry service. Here's a basic example:

```python
import asyncio
import aiohttp
from tesla_fleet_api import Teslemetry
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = Teslemetry(
            access_token="<access_token>",
            session=session,
        )

        try:
            data = await api.vehicles.list()
            print(data)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())
```

For more detailed examples, see [Teslemetry](docs/teslemetry.md).

### Tessie

The `Tessie` class provides methods to interact with the Tessie service. Here's a basic example:

```python
import asyncio
import aiohttp
from tesla_fleet_api import Tessie
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = Tessie(
            access_token="<access_token>",
            session=session,
        )

        try:
            data = await api.vehicles.list()
            print(data)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())
```

For more detailed examples, see [Tessie](docs/tessie.md).
