Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
- **Per-command debug logging chokepoints and the `command=` name it derives**: `LOGGER.debug` lines of the form `command=<name> transport=<t> result=...` are emitted from exactly four places, not per-method - `Commands._sendVehicleSecurity`/`_getVehicleSecurity`/`_sendInfotainment`/`_getInfotainment` (`commands.py`, covers both BLE and Fleet-signed) and `TeslaFleetApi._request` (`fleet.py`, covers Fleet/Teslemetry/Tessie REST). `transport` comes from a `_transport_name` `ClassVar` set per concrete class (`"bluetooth"`/`"fleet"`/`"teslemetry"`/`"tessie"`), mirroring the existing `_auth_method` pattern - add that ClassVar to any new `Commands`/`TeslaFleetApi` subclass. For BLE/Fleet-signed, `command` is **not** the Python method name; it's derived from the populated protobuf oneof field (`vcsec_command_name`/`infotainment_command_name` in `commands.py`), e.g. `door_lock()` logs as `RKE_ACTION_LOCK` and `set_charge_limit()` as `chargingSetLimitAction` - deliberately robust to call-site changes since it reads the message being sent, not the call stack. `VehicleBluetooth`'s `verify_commands` resolution logs a second, separate line (`verify_commands=resolved`/`unresolved`) rather than duplicating the base class's raw-attempt line. `Router._dispatch` (`router/base.py`) logs `command=... backend=<ClassName> result=...` per backend tried, independent of the above. See `docs/bluetooth_vehicles.md`'s "Troubleshooting: Enable Debug Logging" section for the user-facing format; `tests/test_command_logging.py` locks in the exact line shapes.
- **`_log_request_result` (`fleet.py`) must tolerate any JSON-legal REST body, not just dicts**: it runs after the HTTP request already succeeded, so it's a logging convenience only - a non-dict body (`null`, a list, a bare scalar; live case: Teslemetry's `list_authorized_clients` returning `null`) must never raise there. It guards with `isinstance(data, dict)` before calling `.get()`, logging `result=success` and returning for anything else. Regression tests in `tests/test_command_logging.py` (`test_null_json_body_returns_none_without_raising` etc.) cover null/list/scalar bodies.
- **Typed accessor pattern for undocumented raw-dict responses**: `TeslemetryEnergySite.find_authorized_clients()` (`teslemetry/energysite.py`) is the library's first frozen-dataclass typed wrapper over a raw `dict[str, Any]`/`None`/`list` REST response, added so API-parsing logic (envelope unwrap, field lookup, shape validation, enum typing) lives in the library instead of each consumer (e.g. Home Assistant's config flow) reimplementing it. `_authorized_clients_list()` does one precise, non-recursive envelope unwrap (`{"response": {"authorized_clients": [...]}}` or `{"response": {"clients": [...]}}`, or a bare list with no envelope) rather than a multi-key/depth-first search - the `clients` key variant was added after a live capture (Tesla Release 953) showed the endpoint's real key differs from the originally-documented `authorized_clients`, confirmed against a populated 5-entry sample. `AuthorizedClient` models only the two fields a pairing flow actually reads (`public_key`, `state`), each accepting only the specific key-name variant pairs empirically evidenced, not speculative extras. Two rules any future typed accessor over an undocumented response shape must keep, demonstrated here even though `AuthorizedClientState` (`const.py`) has no falsy member: (1) field lookup must check key presence (`key in payload`), never `payload.get(key) or default` - a legal falsy value is not "missing", and a present-but-unrecognized enum value (`_normalize_state()`) is returned raw rather than coerced to `None`; (2) a `None` body and an unrecognized response shape (not a dict/list, or an envelope that unwraps to neither accepted list key) are malformed data, not "zero clients" - Tesla's endpoint intermittently returns HTTP 200 with a null body, so `_authorized_clients_list()` raises `InvalidResponse` (`exceptions.py`) for both rather than collapsing them to `[]`; only a genuinely empty list under either accepted key parses to `AuthorizedClients.clients == []` without raising. Tesla has not published an OpenAPI schema for pairing endpoints, so `const.py`'s enums are the schema of record; widen `AuthorizedClient`'s modeled fields only against a further live sample, not speculatively. The raw `list_authorized_clients()` method is kept alongside as the untyped escape hatch and still returns a null body unchanged rather than raising. Tests: `tests/test_teslemetry_authorized_clients.py`.
- **`networking_status`'s `ipv4_config` fields are raw big-endian uint32 ints, not strings**: the wire format applies to all of `ipv4_config.address`/`subnet_mask`/`gateway`, but `TeslemetryEnergySite.find_gateway_address()` (`teslemetry/energysite.py`, second typed accessor after `find_authorized_clients()`, same rules) decodes only `address` - confirmed against a live Powerwall 3 capture where `3232235914` decodes network-byte-order (`struct.pack(">I", ...)`) to `192.168.1.138`, not little-endian. It considers only `eth`/`wifi` (never `gsm` - cellular isn't a LAN path), preferring whichever has `active_route` set and a decodable address, else falling back to the first of the two (in that order) with any decodable address; `0`/`0xFFFFFFFF` are treated as undecodable so an unconfigured interface never shadows a real one with `0.0.0.0`. A `{"response": null}` envelope raises `InvalidResponse` (the endpoint's known intermittent malformed mode), while a well-formed response with no usable interface returns `None` (not a raise). Tests: `tests/test_teslemetry_gateway_address.py`.

## Maintaining this file

Expand Down
8 changes: 8 additions & 0 deletions docs/energy_local_control.md
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,14 @@ async def make_local_energysite(session: aiohttp.ClientSession) -> PowerwallEner
return PowerwallEnergySite(powerwall_client)
```

The example uses `192.168.91.1`, the gateway's own WiFi access point. When
your client connects over your own LAN instead, you can discover the
gateway's LAN IP over the cloud with
`TeslemetryEnergySite.find_gateway_address()`, which reads the gateway's
`networking_status` and returns the decoded IPv4 of its active `eth`/`wifi`
interface (or `None` when no usable interface is reported) - see
[Teslemetry](teslemetry.md#energy-site-gateway-address).

## 4. Verify the key is paired by using it

The gateway takes registration (step 2) and confirmation (auto-verify, or a
Expand Down
38 changes: 38 additions & 0 deletions docs/teslemetry.md
Original file line number Diff line number Diff line change
Expand Up @@ -585,6 +585,44 @@ async def main():
asyncio.run(main())
```

## Energy Site Gateway Address

Teslemetry energy sites expose the raw `get_networking_status` command, plus
a typed `find_gateway_address` helper that discovers the gateway's LAN IPv4
address - for example to pre-fill the host for the signed local control path
shown in [Energy: Local Control](energy_local_control.md). The `ipv4_config`
fields in a `networking_status` response are raw big-endian uint32 integers,
not dotted-quad strings; the helper decodes them and selects an interface for
you. Only the `eth` and `wifi` interfaces are considered (never `gsm` -
cellular is not a LAN path): the helper prefers whichever has `active_route`
set and a decodable address, then falls back to the first of the two (in
`eth`, `wifi` order) with any decodable address. A null response body or an
unrecognized response shape raises
`tesla_fleet_api.exceptions.InvalidResponse`, so malformed data is never
mistaken for "no address"; a well-formed response where no interface yields a
usable address returns `None` instead.

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

energy_site = teslemetry.energySites.create(12345)

address = await energy_site.find_gateway_address()
print(address) # e.g. "192.168.1.138", or None

# The untyped response is still available when callers need the exact
# Teslemetry payload.
raw = await energy_site.get_networking_status()
print(raw)

asyncio.run(main())
```

## Migrate to OAuth

The `migrate_to_oauth` method migrates from an access token to OAuth.
Expand Down
2 changes: 1 addition & 1 deletion tesla_fleet_api.egg-info/PKG-INFO
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
Metadata-Version: 2.4
Name: tesla_fleet_api
Version: 1.7.3
Version: 1.7.4
Summary: Tesla Fleet API library for Python
Author-email: Brett Adams <hello@teslemetry.com>
License-Expression: Apache-2.0
Expand Down
1 change: 1 addition & 0 deletions tesla_fleet_api.egg-info/SOURCES.txt
Original file line number Diff line number Diff line change
Expand Up @@ -92,4 +92,5 @@ tests/test_power_mode_commands.py
tests/test_router.py
tests/test_tesla_private_key.py
tests/test_teslemetry_authorized_clients.py
tests/test_teslemetry_gateway_address.py
tests/test_tessie_vehicle_params.py
102 changes: 102 additions & 0 deletions tesla_fleet_api/teslemetry/energysite.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
from __future__ import annotations

import base64
import socket
import struct
from dataclasses import dataclass
from typing import Any, cast

Expand Down Expand Up @@ -147,6 +149,90 @@ def _parse_authorized_clients(payload: Any) -> AuthorizedClients:
return AuthorizedClients(clients=clients, raw=payload)


_GATEWAY_INTERFACES = ("eth", "wifi")


def _decode_ipv4(value: Any) -> str | None:
"""Decode a raw big-endian uint32 into dotted-quad form.

``ipv4_config.address``/``subnet_mask``/``gateway`` in a
``networking_status`` response are network-byte-order uint32 integers,
not strings - confirmed against a live Powerwall 3 capture where
``3232235914`` decodes to ``192.168.1.138``. ``bool`` is excluded since
it subclasses ``int``; an out-of-range or non-int value returns
``None`` rather than raising, since a single bad address shouldn't
fail the whole lookup. ``0`` and ``0xFFFFFFFF`` are also rejected -
``0.0.0.0``/``255.255.255.255`` are never a usable host address, and an
unconfigured interface reporting ``address: 0`` must not shadow a real
address on another interface in the fallback selection.
"""
if not isinstance(value, int) or isinstance(value, bool):
return None
if not 0 < value < 0xFFFFFFFF:
return None
return socket.inet_ntoa(struct.pack(">I", value))


def _interface_address(interface: Any) -> str | None:
if not isinstance(interface, dict):
return None
ipv4 = cast("dict[str, Any]", interface).get("ipv4_config")
if not isinstance(ipv4, dict) or "address" not in ipv4:
return None
return _decode_ipv4(cast("dict[str, Any]", ipv4)["address"])


def _networking_status_body(payload: Any) -> dict[str, Any]:
"""Unwrap a ``networking_status`` response into its interface-block dict.

Raises :class:`~tesla_fleet_api.exceptions.InvalidResponse` for a null
body, a shape that isn't a dict, or a ``response`` envelope whose value
isn't a dict (``{"response": null}`` is the endpoint's known
intermittent malformed mode, not "no address") - anything else is a
well-formed body, even if it carries no usable interface.
"""
if payload is None:
raise InvalidResponse("networking_status response body was null")
if not isinstance(payload, dict):
raise InvalidResponse(str(payload))
body = cast("dict[str, Any]", payload)
if "response" in body:
response = body["response"]
if not isinstance(response, dict):
raise InvalidResponse(cast("dict[str, Any]", payload))
body = cast("dict[str, Any]", response)
return body


def _parse_gateway_address(payload: Any) -> str | None:
"""Pick the gateway's LAN IPv4 from a ``networking_status`` response.

Considers only ``eth``/``wifi`` (never ``gsm`` - cellular isn't a LAN
path): prefers whichever of those has ``active_route`` set and a
decodable address, then falls back to the first of the two (in
``eth``, ``wifi`` order) that has any decodable address. Returns
``None`` when neither yields one - a well-formed response can simply
lack a usable interface.
"""
body = _networking_status_body(payload)
interfaces = [body.get(name) for name in _GATEWAY_INTERFACES]

for interface in interfaces:
if (
isinstance(interface, dict)
and cast("dict[str, Any]", interface).get("active_route")
and (address := _interface_address(interface)) is not None
):
return address

for interface in interfaces:
address = _interface_address(interface)
if address is not None:
return address

return None


class TeslemetryEnergySite(EnergySite):
"""Teslemetry specific energy site."""

Expand Down Expand Up @@ -201,6 +287,22 @@ async def get_networking_status(self) -> dict[str, Any]:
f"api/1/energy_sites/{self.energy_site_id}/command/networking_status",
)

async def find_gateway_address(self) -> str | None:
"""Discover the gateway's LAN IPv4 address, for pre-filling a local
control host.

Prefer this over :meth:`get_networking_status` for consumers that
just need a host to connect to - it centralizes the ``eth``/``wifi``
interface selection and uint32-to-dotted-quad decoding here instead
of in the caller. Raises
:class:`~tesla_fleet_api.exceptions.InvalidResponse` on a null
response body or an unrecognized response shape; returns ``None``
when the response is well-formed but no interface yields an
address. See :func:`_parse_gateway_address` for the exact selection
rule.
"""
return _parse_gateway_address(await self.get_networking_status())

async def list_authorized_clients(self) -> dict[str, Any]:
"""List authorized clients on the energy gateway via the Teslemetry
custom endpoint.
Expand Down
Loading
Loading