feat(vehicle): add VehicleRouter for primary/fallback command dispatch

AGENTS.md

@@ -59,6 +59,8 @@ Commands (vehicle/commands.py) - protobuf-based signed command implementation (A
 
 `VehicleSigned` uses multiple inheritance: `Commands` for signed command logic, `VehicleFleet` for data endpoints and fallback.
 
+`VehicleRouter` (vehicle/router.py) is a composition wrapper (not part of the inheritance chain) that pairs a primary and a fallback vehicle instance and dispatches each method call to the primary when healthy, with automatic per-command failover to the fallback on error — e.g. a `VehicleBluetooth` primary with a cloud (`TeslemetryVehicle`) fallback. The health check may be a `bool`, a sync callable, or an async callable returning `bool`; when omitted it attempts the primary and fails over on exception with no up-front probe. It is exported from `tesla/vehicle/__init__.py` but has no factory on the `Vehicles` collections.
+
 ### Vehicle Collections
 
 `Vehicles` (vehicle/vehicles.py) is a `dict[str, Vehicle]` with factory methods:

README.md

@@ -8,6 +8,7 @@ Tesla Fleet API is a Python library that provides an interface to interact with
 - Fleet API for energy sites
 - Fleet API with signed vehicle commands
 - Bluetooth for vehicles
+- Routing and failover between vehicle transports (e.g. Bluetooth primary, cloud fallback)
 - Teslemetry integration
 - Tessie integration
 
@@ -165,6 +166,44 @@ asyncio.run(main())
 
 For more detailed examples, see [Bluetooth for Vehicles](docs/bluetooth_vehicles.md).
 
+### Routing and Failover
+
+The `VehicleRouter` class composes a primary and a fallback vehicle instance and dispatches each method call to the primary when it is healthy, automatically failing over to the fallback on error. 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.vehicle.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())
+```
+
+By default the router attempts the primary and fails over to the fallback 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 straight to the fallback.
+
+> **Warning:** Because a failed primary call is replayed on the fallback, a non-idempotent command (e.g. `honk_horn`, `actuate_trunk`, `door_unlock`, `charge_start`) that fails _mid-flight_ — after the primary may have already partially applied it — can be **double-executed** when it is retried on the fallback. 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 `primary`/`fallback` instances directly.
+>
+> Dispatch is implemented via `__getattr__`, which does not proxy dunder methods, so `async with VehicleRouter(...)` does **not** manage the primary's BLE connection lifecycle (`__aenter__`/`__aexit__`). Commands still auto-connect on send; for explicit connect/disconnect reach through `vehicle.primary`.
+
 ### Teslemetry
 
 The `Teslemetry` class provides methods to interact with the Teslemetry service. Here's a basic example:

tesla_fleet_api.egg-info/PKG-INFO

@@ -11,13 +11,13 @@ Classifier: Operating System :: OS Independent
 Requires-Python: >=3.10
 Description-Content-Type: text/markdown
 License-File: LICENSE
-Requires-Dist: aiohttp
-Requires-Dist: aiofiles
-Requires-Dist: aiolimiter
-Requires-Dist: cryptography
-Requires-Dist: protobuf
-Requires-Dist: bleak
-Requires-Dist: bleak-retry-connector
+Requires-Dist: aiohttp>=3
+Requires-Dist: aiofiles>=24
+Requires-Dist: aiolimiter>=1
+Requires-Dist: cryptography>=43
+Requires-Dist: protobuf>=6.31.1
+Requires-Dist: bleak>=0.22
+Requires-Dist: bleak-retry-connector>=3.9
 Dynamic: license-file
 
 # Tesla Fleet API
@@ -30,6 +30,7 @@ Tesla Fleet API is a Python library that provides an interface to interact with
 - Fleet API for energy sites
 - Fleet API with signed vehicle commands
 - Bluetooth for vehicles
+- Routing and failover between vehicle transports (e.g. Bluetooth primary, cloud fallback)
 - Teslemetry integration
 - Tessie integration
 
@@ -187,6 +188,44 @@ asyncio.run(main())
 
 For more detailed examples, see [Bluetooth for Vehicles](docs/bluetooth_vehicles.md).
 
+### Routing and Failover
+
+The `VehicleRouter` class composes a primary and a fallback vehicle instance and dispatches each method call to the primary when it is healthy, automatically failing over to the fallback on error. 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.vehicle.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())
+```
+
+By default the router attempts the primary and fails over to the fallback 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 straight to the fallback.
+
+> **Warning:** Because a failed primary call is replayed on the fallback, a non-idempotent command (e.g. `honk_horn`, `actuate_trunk`, `door_unlock`, `charge_start`) that fails _mid-flight_ — after the primary may have already partially applied it — can be **double-executed** when it is retried on the fallback. 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 `primary`/`fallback` instances directly.
+>
+> Dispatch is implemented via `__getattr__`, which does not proxy dunder methods, so `async with VehicleRouter(...)` does **not** manage the primary's BLE connection lifecycle (`__aenter__`/`__aexit__`). Commands still auto-connect on send; for explicit connect/disconnect reach through `vehicle.primary`.
+
 ### Teslemetry
 
 The `Teslemetry` class provides methods to interact with the Teslemetry service. Here's a basic example:

tesla_fleet_api.egg-info/SOURCES.txt

@@ -23,6 +23,7 @@ tesla_fleet_api/tesla/vehicle/__init__.py
 tesla_fleet_api/tesla/vehicle/bluetooth.py
 tesla_fleet_api/tesla/vehicle/commands.py
 tesla_fleet_api/tesla/vehicle/fleet.py
+tesla_fleet_api/tesla/vehicle/router.py
 tesla_fleet_api/tesla/vehicle/signed.py
 tesla_fleet_api/tesla/vehicle/vehicle.py
 tesla_fleet_api/tesla/vehicle/vehicles.py
@@ -54,4 +55,5 @@ tesla_fleet_api/tessie/__init__.py
 tesla_fleet_api/tessie/tessie.py
 tesla_fleet_api/tessie/vehicles.py
 tests/test_fleet_auth_refresh.py
-tests/test_tessie_vehicle_params.py
+tests/test_tessie_vehicle_params.py
+tests/test_vehicle_router.py

tesla_fleet_api.egg-info/requires.txt

@@ -1,7 +1,7 @@
-aiohttp
-aiofiles
-aiolimiter
-cryptography
-protobuf
-bleak
-bleak-retry-connector
+aiohttp>=3
+aiofiles>=24
+aiolimiter>=1
+cryptography>=43
+protobuf>=6.31.1
+bleak>=0.22
+bleak-retry-connector>=3.9

tesla_fleet_api/tesla/vehicle/__init__.py

@@ -4,6 +4,7 @@
 from tesla_fleet_api.tesla.vehicle.fleet import VehicleFleet
 from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth
 from tesla_fleet_api.tesla.vehicle.signed import VehicleSigned
+from tesla_fleet_api.tesla.vehicle.router import VehicleRouter
 from tesla_fleet_api.tesla.vehicle.vehicle import Vehicle
 
 __all__ = [
@@ -13,4 +14,5 @@
     "VehicleFleet",
     "VehicleBluetooth",
     "VehicleSigned",
+    "VehicleRouter",
 ]

tesla_fleet_api/tesla/vehicle/router.py

@@ -0,0 +1,180 @@
+from __future__ import annotations
+
+import inspect
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Generic,
+    TypeVar,
+    Union,
+)
+
+from tesla_fleet_api.const import LOGGER
+from tesla_fleet_api.exceptions import TeslaFleetError
+
+PrimaryT = TypeVar("PrimaryT")
+FallbackT = TypeVar("FallbackT")
+
+# A health check may be a static bool, a sync callable returning bool, or an
+# async callable returning bool. When omitted the router attempts the primary
+# directly and fails over to the fallback on exception (no up-front probe).
+HealthCheck = Union[bool, Callable[[], bool], Callable[[], Awaitable[bool]]]
+
+
+async def _maybe_await(value: Any) -> Any:
+    """Await ``value`` if it is awaitable, otherwise return it unchanged."""
+    if inspect.isawaitable(value):
+        return await value
+    return value
+
+
+class VehicleRouter(Generic[PrimaryT, FallbackT]):
+    """Routes method calls to a primary instance when healthy, else a fallback.
+
+    Composes an ordered pair of vehicle instances that share a common method
+    surface (e.g. a :class:`VehicleBluetooth` primary and a cloud
+    :class:`VehicleFleet`/``TeslemetryVehicle`` fallback). Method calls are
+    dispatched dynamically:
+
+    - Present and callable on *both* -> the primary is attempted (subject to an
+      explicit health check) with automatic failover to the fallback.
+    - Present only on the fallback -> the fallback is called (no health gate).
+    - Present only on the primary -> the primary is called.
+    - Present on neither -> :class:`AttributeError`.
+
+    Non-callable attributes (e.g. ``vin``) resolve to the primary's value when
+    present, otherwise the fallback's.
+
+    Dispatch to a both-present callable performs *per-command* failover: the
+    primary is attempted first and, if it raises any exception (a connection
+    failure or a mid-command transport error such as a write/notify failure or a
+    disconnect), the same call is automatically retried on the fallback with the
+    same arguments. The error only propagates when the fallback also fails.
+
+    .. warning::
+
+        Because a failed primary call is replayed on the fallback, a
+        *non-idempotent* command (e.g. ``honk_horn``, ``actuate_trunk``,
+        ``door_unlock``, ``charge_start``) that fails *mid-flight* — after the
+        primary may have already partially applied it — can be **double-executed**
+        when it is retried on the fallback. This is an accepted, deliberate
+        tradeoff of per-command failover. Callers that need exactly-once
+        semantics for non-idempotent commands should gate dispatch with an
+        explicit health check (so a failing primary skips the primary entirely
+        rather than replaying on the fallback) or call the underlying
+        :attr:`primary`/:attr:`fallback` instances directly.
+
+    The health check may be provided as a ``bool``, a sync callable, or an async
+    callable returning ``bool``. When an explicit check evaluates ``False`` the
+    primary is skipped entirely and the call routes straight to the fallback;
+    when it evaluates ``True`` the primary is attempted with the same
+    fall-back-on-exception behaviour. When omitted (the default) the primary is
+    attempted directly with fall-back-on-exception and no up-front probe.
+
+    Dispatch is implemented via :meth:`__getattr__`, which does **not** proxy
+    special/dunder methods (Python looks those up on the type, not the instance).
+    In particular ``async with VehicleRouter(...)`` does *not* enter the primary's
+    async context manager, so a :class:`VehicleBluetooth` primary's BLE connection
+    lifecycle (``__aenter__``/``__aexit__``) is not managed by the router — its
+    commands still auto-connect on send, but explicit connect/disconnect must be
+    done by reaching through :attr:`primary`.
+
+    The class is deliberately unbound over its two type parameters so the same
+    pattern can later wrap a pair of energy site classes.
+    """
+
+    _primary: PrimaryT
+    _fallback: FallbackT
+    _health: HealthCheck | None
+
+    def __init__(
+        self,
+        primary: PrimaryT,
+        fallback: FallbackT,
+        health: HealthCheck | None = None,
+    ):
+        self._primary = primary
+        self._fallback = fallback
+        self._health = health
+
+    async def is_healthy(self) -> bool:
+        """Resolve an explicit health check to a bool.
+
+        When no explicit check is configured the router does not probe up front —
+        it attempts the primary and fails over on exception — so this reports
+        ``True`` in that case.
+        """
+        health = self._health
+        if health is None:
+            return True
+        if isinstance(health, bool):
+            return health
+        return bool(await _maybe_await(health()))
+
+    def _dispatch(
+        self,
+        name: str,
+        primary_attr: Callable[..., Any],
+        fallback_attr: Callable[..., Any],
+    ) -> Callable[..., Awaitable[Any]]:
+        """Build an async wrapper with per-command failover for a shared method."""
+
+        async def _routed(*args: Any, **kwargs: Any) -> Any:
+            if self._health is not None and not await self.is_healthy():
+                return await _maybe_await(fallback_attr(*args, **kwargs))
+            try:
+                return await _maybe_await(primary_attr(*args, **kwargs))
+            except (Exception, TeslaFleetError) as e:  # noqa: BLE001 - any primary failure -> fallback
+                LOGGER.debug("Primary call %r failed, routing to fallback: %s", name, e)
+                return await _maybe_await(fallback_attr(*args, **kwargs))
+
+        _routed.__name__ = name
+        _routed.__qualname__ = f"{type(self).__name__}.{name}"
+        return _routed
+
+    def __getattr__(self, name: str) -> Any:
+        # __getattr__ is only reached when normal lookup fails. Guard the private
+        # attributes so an access before __init__ completes raises rather than
+        # recursing infinitely through this method.
+        if name in ("_primary", "_fallback", "_health"):
+            raise AttributeError(name)
+
+        primary = self._primary
+        fallback = self._fallback
+        has_primary = hasattr(primary, name)
+        has_fallback = hasattr(fallback, name)
+
+        if not has_primary and not has_fallback:
+            raise AttributeError(
+                f"{type(self).__name__!r} routes to neither primary "
+                f"{type(primary).__name__!r} nor fallback "
+                f"{type(fallback).__name__!r} for {name!r}"
+            )
+
+        primary_attr = getattr(primary, name) if has_primary else None
+        fallback_attr = getattr(fallback, name) if has_fallback else None
+
+        if (
+            has_primary
+            and has_fallback
+            and callable(primary_attr)
+            and callable(fallback_attr)
+        ):
+            return self._dispatch(name, primary_attr, fallback_attr)
+
+        # Only one side has it (pure fall-through), or the shared attribute is not
+        # callable. Prefer the primary's value when present.
+        if has_primary:
+            return primary_attr
+        return fallback_attr
+
+    @property
+    def primary(self) -> PrimaryT:
+        """The primary instance calls are routed to when healthy."""
+        return self._primary
+
+    @property
+    def fallback(self) -> FallbackT:
+        """The fallback instance calls are routed to when the primary is unhealthy."""
+        return self._fallback