From c8ad07a05dcdbbe96fdf583369dee7c130a2d807 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Tue, 14 Jul 2026 16:59:56 +1000 Subject: [PATCH 1/3] feat(teslemetry): add find_gateway_address() typed accessor Discovers the gateway's LAN IPv4 from networking_status so Home Assistant's local-control config flow can pre-fill the host instead of requiring the user to type it. Follows the find_authorized_clients() typed-accessor precedent: eth/wifi only (never gsm), prefers the active-route interface, falls back to the first with a decodable address, and decodes ipv4_config's raw big-endian uint32 fields into dotted-quad form. --- AGENTS.md | 1 + tesla_fleet_api/teslemetry/energysite.py | 96 +++++++++ tests/test_teslemetry_gateway_address.py | 235 +++++++++++++++++++++++ uv.lock | 2 +- 4 files changed, 333 insertions(+), 1 deletion(-) create mode 100644 tests/test_teslemetry_gateway_address.py diff --git a/AGENTS.md b/AGENTS.md index 85be2ae..68dead3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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= transport= 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= 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**: `TeslemetryEnergySite.find_gateway_address()` (`teslemetry/energysite.py`, second typed accessor after `find_authorized_clients()`, same rules) decodes `ipv4_config.address`/`subnet_mask`/`gateway` - 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; returns `None` (not a raise) when a well-formed response simply has no usable interface. Tests: `tests/test_teslemetry_gateway_address.py`. ## Maintaining this file diff --git a/tesla_fleet_api/teslemetry/energysite.py b/tesla_fleet_api/teslemetry/energysite.py index 7799975..3eeb3b9 100644 --- a/tesla_fleet_api/teslemetry/energysite.py +++ b/tesla_fleet_api/teslemetry/energysite.py @@ -1,6 +1,8 @@ from __future__ import annotations import base64 +import socket +import struct from dataclasses import dataclass from typing import Any, cast @@ -147,6 +149,84 @@ 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. + """ + 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 or a shape that isn't a dict (with or without a ``response`` + envelope) - 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) + response = body.get("response") + if isinstance(response, dict): + 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.""" @@ -201,6 +281,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. diff --git a/tests/test_teslemetry_gateway_address.py b/tests/test_teslemetry_gateway_address.py new file mode 100644 index 0000000..70a3c6f --- /dev/null +++ b/tests/test_teslemetry_gateway_address.py @@ -0,0 +1,235 @@ +"""Tests for TeslemetryEnergySite.find_gateway_address(), the typed accessor +over the Teslemetry ``command/networking_status`` endpoint. + +Only ``eth``/``wifi`` are considered (never ``gsm`` - cellular isn't a LAN +path). An interface with ``active_route`` set wins; otherwise the first of +``eth``, ``wifi`` (in that order) with a decodable address is used. Address +fields on the wire are raw big-endian uint32 integers, not strings - see +``GatewayAddressRealCaptureTests`` for the real captured sample this decoding +is pinned against (see ``_parse_gateway_address`` in +``tesla_fleet_api/teslemetry/energysite.py``). + +A null body or an unrecognized response shape is malformed data and must +raise :class:`~tesla_fleet_api.exceptions.InvalidResponse`; a well-formed +response that simply has no usable interface returns ``None``. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from typing import Any +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock + +from tesla_fleet_api.exceptions import InvalidResponse +from tesla_fleet_api.teslemetry.teslemetry import Teslemetry + +_UNSET = object() + + +def _fake_response(*, json_body: object = _UNSET) -> MagicMock: + resp = MagicMock() + resp.status = 200 + resp.ok = True + resp.content_type = "application/json" + resp.url = "https://example.com/x" + resp.headers = {} + resp.json = AsyncMock(return_value={} if json_body is _UNSET else json_body) + resp.text = AsyncMock(return_value="") + return resp + + +def _make_session(response: object) -> MagicMock: + session = MagicMock() + + @asynccontextmanager + async def _ctx(*args: Any, **kwargs: Any): + yield response + + session.request = MagicMock(side_effect=lambda *a, **k: _ctx(*a, **k)) + return session + + +def _make_site(json_body: object): + api = Teslemetry( + session=_make_session(_fake_response(json_body=json_body)), + access_token="token", + ) + return api.energySites.create(12345) + + +class GatewayAddressRealCaptureTests(IsolatedAsyncioTestCase): + """Captured 2026-07-14 from a Powerwall 3's ``networking_status`` + response (identifiers anonymized, structure verbatim). ``wifi`` is the + active route; ``eth`` is populated but not active; ``gsm`` must never + be considered. + """ + + REAL_CAPTURE_RESPONSE = { + "response": { + "wifi_config": {"ssid": "ANONYMIZED_SSID"}, + "wifi": { + "mac_address": "ANONYMIZED_MAC_1", + "enabled": True, + "active_route": True, + "ipv4_config": { + "dhcp_enabled": True, + "address": 3232235914, + "subnet_mask": 4294967040, + "gateway": 3232235777, + }, + "connectivity_status": { + "connected_physical": True, + "connected_internet": True, + "connected_tesla": True, + "rssi": {"signal_strength_percent": {"value": 42}}, + }, + "device_state": 6, + "device_state_reason": 1, + }, + "eth": { + "mac_address": "ANONYMIZED_MAC_2", + "enabled": True, + "ipv4_config": { + "dhcp_enabled": True, + "address": 3232258562, + "subnet_mask": 4294967040, + }, + "connectivity_status": {"rssi": {"signal_strength_percent": {}}}, + }, + "gsm": { + "enabled": True, + "ipv4_config": { + "address": 168866907, + "subnet_mask": 4294967295, + "gateway": 168866907, + }, + "connectivity_status": { + "connected_physical": True, + "connected_internet": True, + "connected_tesla": True, + "rssi": {"signal_strength_percent": {"value": 60}}, + }, + }, + } + } + + async def test_real_captured_sample_selects_active_route_wifi(self) -> None: + site = _make_site(self.REAL_CAPTURE_RESPONSE) + + result = await site.find_gateway_address() + + self.assertEqual(result, "192.168.1.138") + + +class GatewayAddressSelectionTests(IsolatedAsyncioTestCase): + async def test_eth_preferred_when_eth_has_active_route(self) -> None: + site = _make_site( + { + "response": { + "eth": { + "active_route": True, + "ipv4_config": {"address": 3232235777}, + }, + "wifi": { + "active_route": True, + "ipv4_config": {"address": 3232235914}, + }, + } + } + ) + + result = await site.find_gateway_address() + + self.assertEqual(result, "192.168.1.1") + + async def test_no_active_route_falls_back_to_eth_first(self) -> None: + site = _make_site( + { + "response": { + "eth": {"ipv4_config": {"address": 3232258562}}, + "wifi": {"ipv4_config": {"address": 3232235914}}, + } + } + ) + + result = await site.find_gateway_address() + + self.assertEqual(result, "192.168.90.2") + + async def test_no_active_route_falls_back_to_wifi_when_eth_has_no_address( + self, + ) -> None: + site = _make_site( + { + "response": { + "eth": {"enabled": True}, + "wifi": {"ipv4_config": {"address": 3232235914}}, + } + } + ) + + result = await site.find_gateway_address() + + self.assertEqual(result, "192.168.1.138") + + async def test_gsm_only_returns_none(self) -> None: + site = _make_site( + { + "response": { + "gsm": { + "active_route": True, + "ipv4_config": {"address": 168866907}, + } + } + } + ) + + result = await site.find_gateway_address() + + self.assertIsNone(result) + + async def test_empty_response_returns_none(self) -> None: + site = _make_site({"response": {}}) + + result = await site.find_gateway_address() + + self.assertIsNone(result) + + async def test_interfaces_missing_ipv4_config_return_none(self) -> None: + site = _make_site( + {"response": {"eth": {"enabled": True}, "wifi": {"enabled": True}}} + ) + + result = await site.find_gateway_address() + + self.assertIsNone(result) + + async def test_bare_body_without_response_envelope(self) -> None: + site = _make_site( + {"wifi": {"active_route": True, "ipv4_config": {"address": 3232235914}}} + ) + + result = await site.find_gateway_address() + + self.assertEqual(result, "192.168.1.138") + + +class GatewayAddressInvalidResponseTests(IsolatedAsyncioTestCase): + async def test_null_body_raises_invalid_response(self) -> None: + site = _make_site(None) + + with self.assertRaises(InvalidResponse): + await site.find_gateway_address() + + async def test_unrecognized_non_dict_body_raises_invalid_response(self) -> None: + site = _make_site("not-a-valid-shape") + + with self.assertRaises(InvalidResponse): + await site.find_gateway_address() + + async def test_list_body_raises_invalid_response(self) -> None: + site = _make_site([]) + + with self.assertRaises(InvalidResponse): + await site.find_gateway_address() diff --git a/uv.lock b/uv.lock index 8ad292a..a48a120 100644 --- a/uv.lock +++ b/uv.lock @@ -728,7 +728,7 @@ wheels = [ [[package]] name = "tesla-fleet-api" -version = "1.7.3" +version = "1.7.4" source = { editable = "." } dependencies = [ { name = "aiofiles" }, From 58521d3e757f3f7caabf581438d71e6b78529eca Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Tue, 14 Jul 2026 17:07:25 +1000 Subject: [PATCH 2/3] no-mistakes(review): reject null envelope and 0/broadcast addresses in gateway lookup --- AGENTS.md | 2 +- tesla_fleet_api.egg-info/PKG-INFO | 2 +- tesla_fleet_api.egg-info/SOURCES.txt | 1 + tesla_fleet_api/teslemetry/energysite.py | 20 ++++++---- tests/test_teslemetry_gateway_address.py | 49 ++++++++++++++++++++++++ 5 files changed, 65 insertions(+), 9 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 68dead3..84737f4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -143,7 +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= transport= 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= 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**: `TeslemetryEnergySite.find_gateway_address()` (`teslemetry/energysite.py`, second typed accessor after `find_authorized_clients()`, same rules) decodes `ipv4_config.address`/`subnet_mask`/`gateway` - 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; returns `None` (not a raise) when a well-formed response simply has no usable interface. Tests: `tests/test_teslemetry_gateway_address.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 diff --git a/tesla_fleet_api.egg-info/PKG-INFO b/tesla_fleet_api.egg-info/PKG-INFO index 85c8143..357ce84 100644 --- a/tesla_fleet_api.egg-info/PKG-INFO +++ b/tesla_fleet_api.egg-info/PKG-INFO @@ -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 License-Expression: Apache-2.0 diff --git a/tesla_fleet_api.egg-info/SOURCES.txt b/tesla_fleet_api.egg-info/SOURCES.txt index ec90220..555455b 100644 --- a/tesla_fleet_api.egg-info/SOURCES.txt +++ b/tesla_fleet_api.egg-info/SOURCES.txt @@ -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 \ No newline at end of file diff --git a/tesla_fleet_api/teslemetry/energysite.py b/tesla_fleet_api/teslemetry/energysite.py index 3eeb3b9..5b31301 100644 --- a/tesla_fleet_api/teslemetry/energysite.py +++ b/tesla_fleet_api/teslemetry/energysite.py @@ -161,11 +161,14 @@ def _decode_ipv4(value: Any) -> str | None: ``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. + 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: + if not 0 < value < 0xFFFFFFFF: return None return socket.inet_ntoa(struct.pack(">I", value)) @@ -183,17 +186,20 @@ 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 or a shape that isn't a dict (with or without a ``response`` - envelope) - anything else is a well-formed body, even if it carries no - usable interface. + 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) - response = body.get("response") - if isinstance(response, dict): + 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 diff --git a/tests/test_teslemetry_gateway_address.py b/tests/test_teslemetry_gateway_address.py index 70a3c6f..ea54054 100644 --- a/tests/test_teslemetry_gateway_address.py +++ b/tests/test_teslemetry_gateway_address.py @@ -173,6 +173,49 @@ async def test_no_active_route_falls_back_to_wifi_when_eth_has_no_address( self.assertEqual(result, "192.168.1.138") + async def test_zero_address_is_undecodable(self) -> None: + site = _make_site( + { + "response": { + "eth": {"active_route": True, "ipv4_config": {"address": 0}}, + } + } + ) + + result = await site.find_gateway_address() + + self.assertIsNone(result) + + async def test_broadcast_address_is_undecodable(self) -> None: + site = _make_site( + { + "response": { + "eth": { + "active_route": True, + "ipv4_config": {"address": 4294967295}, + }, + } + } + ) + + result = await site.find_gateway_address() + + self.assertIsNone(result) + + async def test_fallback_skips_eth_with_zero_address(self) -> None: + site = _make_site( + { + "response": { + "eth": {"ipv4_config": {"address": 0}}, + "wifi": {"ipv4_config": {"address": 3232235914}}, + } + } + ) + + result = await site.find_gateway_address() + + self.assertEqual(result, "192.168.1.138") + async def test_gsm_only_returns_none(self) -> None: site = _make_site( { @@ -222,6 +265,12 @@ async def test_null_body_raises_invalid_response(self) -> None: with self.assertRaises(InvalidResponse): await site.find_gateway_address() + async def test_null_response_envelope_raises_invalid_response(self) -> None: + site = _make_site({"response": None}) + + with self.assertRaises(InvalidResponse): + await site.find_gateway_address() + async def test_unrecognized_non_dict_body_raises_invalid_response(self) -> None: site = _make_site("not-a-valid-shape") From b1f74f9632136d2ab255b758e23870059e148189 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Tue, 14 Jul 2026 17:14:39 +1000 Subject: [PATCH 3/3] no-mistakes(document): document find_gateway_address in teslemetry and local-control docs --- docs/energy_local_control.md | 8 ++++++++ docs/teslemetry.md | 38 ++++++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/docs/energy_local_control.md b/docs/energy_local_control.md index d78ea24..da1de32 100644 --- a/docs/energy_local_control.md +++ b/docs/energy_local_control.md @@ -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 diff --git a/docs/teslemetry.md b/docs/teslemetry.md index 8fac5d6..191ebd3 100644 --- a/docs/teslemetry.md +++ b/docs/teslemetry.md @@ -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="", + ) + + 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.