|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
3 | 3 | import base64 |
4 | | -from typing import Any |
| 4 | +from dataclasses import dataclass |
| 5 | +from typing import Any, cast |
5 | 6 |
|
6 | 7 | from tesla_fleet_api.const import ( |
7 | 8 | AuthorizedClientKeyType, |
| 9 | + AuthorizedClientState, |
8 | 10 | AuthorizedClientType, |
9 | 11 | Method, |
10 | 12 | ) |
11 | 13 | from tesla_fleet_api.tesla.energysite import EnergySite, EnergySites |
12 | 14 |
|
13 | 15 |
|
| 16 | +def _field(payload: dict[str, Any], *keys: str) -> Any: |
| 17 | + """Return the first present key's value. |
| 18 | +
|
| 19 | + Checks key presence rather than truthiness, so a legal falsy value |
| 20 | + (``0``, ``False``, ``""``) is never mistaken for a missing field. |
| 21 | + """ |
| 22 | + for key in keys: |
| 23 | + if key in payload: |
| 24 | + return payload[key] |
| 25 | + return None |
| 26 | + |
| 27 | + |
| 28 | +def _normalize_state(value: Any) -> AuthorizedClientState | int | str | None: |
| 29 | + """Type a raw ``state`` value against ``AuthorizedClientState``. |
| 30 | +
|
| 31 | + Tesla has not published an OpenAPI schema for this pairing endpoint, so |
| 32 | + ``const.py``'s enum is the schema of record. A recognized int or member |
| 33 | + name (case-insensitive) becomes the enum member; a present-but- |
| 34 | + unrecognized value is returned unchanged rather than dropped to |
| 35 | + ``None``, since only a genuinely absent field means ``None``. ``bool`` |
| 36 | + is excluded from the int branch since it subclasses ``int`` but is |
| 37 | + never a legal state code. |
| 38 | + """ |
| 39 | + if ( |
| 40 | + value is None |
| 41 | + or isinstance(value, AuthorizedClientState) |
| 42 | + or isinstance(value, bool) |
| 43 | + ): |
| 44 | + return value |
| 45 | + if isinstance(value, int): |
| 46 | + try: |
| 47 | + return AuthorizedClientState(value) |
| 48 | + except ValueError: |
| 49 | + return value |
| 50 | + if isinstance(value, str): |
| 51 | + try: |
| 52 | + return AuthorizedClientState[value.strip().upper()] |
| 53 | + except KeyError: |
| 54 | + return value |
| 55 | + return value |
| 56 | + |
| 57 | + |
| 58 | +def _parse_client(payload: dict[str, Any]) -> AuthorizedClient: |
| 59 | + return AuthorizedClient( |
| 60 | + public_key=_field(payload, "public_key", "publicKey"), |
| 61 | + state=_normalize_state(_field(payload, "state", "authorized_client_state")), |
| 62 | + raw=payload, |
| 63 | + ) |
| 64 | + |
| 65 | + |
| 66 | +@dataclass(frozen=True, slots=True) |
| 67 | +class AuthorizedClient: |
| 68 | + """One entry from a Teslemetry ``list_authorized_clients`` response. |
| 69 | +
|
| 70 | + Only ``public_key`` and ``state`` are modeled - the two fields a |
| 71 | + pairing flow needs to confirm a registered key. Tesla has not |
| 72 | + published this response's schema, so anything else on an entry is |
| 73 | + available via ``raw`` rather than guessed at. Each field accepts the |
| 74 | + two key-name variants observed for it (``public_key``/``publicKey``, |
| 75 | + ``state``/``authorized_client_state``). |
| 76 | + """ |
| 77 | + |
| 78 | + public_key: str | None |
| 79 | + state: AuthorizedClientState | int | str | None |
| 80 | + raw: dict[str, Any] |
| 81 | + |
| 82 | + |
| 83 | +@dataclass(frozen=True, slots=True) |
| 84 | +class AuthorizedClients: |
| 85 | + """Parsed result of :meth:`TeslemetryEnergySite.find_authorized_clients`. |
| 86 | +
|
| 87 | + ``clients`` is always a list: a ``None`` response body, an unrecognized |
| 88 | + response shape, and an explicitly empty ``authorized_clients`` list all |
| 89 | + mean "no authorized clients to report" for this endpoint and collapse |
| 90 | + to ``[]`` rather than being distinguished. |
| 91 | +
|
| 92 | + The envelope this unwraps (``{"response": {"authorized_clients": [...]}}``) |
| 93 | + is pinned from the pairing flow's own defensive handling of this |
| 94 | + undocumented endpoint. No site has been observed with a populated |
| 95 | + client list yet, so that per-entry shape is unconfirmed by a live |
| 96 | + sample - see :class:`AuthorizedClient`. |
| 97 | + """ |
| 98 | + |
| 99 | + clients: list[AuthorizedClient] |
| 100 | + raw: Any |
| 101 | + |
| 102 | + |
| 103 | +def _authorized_clients_list(payload: Any) -> list[Any]: |
| 104 | + """Return the raw authorized-clients list from a command response, or []. |
| 105 | +
|
| 106 | + A precise, single-path unwrap of the one confirmed envelope - not a |
| 107 | + search across candidate wrapper keys for this undocumented endpoint. |
| 108 | + """ |
| 109 | + if payload is None: |
| 110 | + return [] |
| 111 | + if isinstance(payload, list): |
| 112 | + return cast("list[Any]", payload) |
| 113 | + if not isinstance(payload, dict): |
| 114 | + return [] |
| 115 | + body = cast("dict[str, Any]", payload) |
| 116 | + response = body.get("response") |
| 117 | + if isinstance(response, dict): |
| 118 | + body = cast("dict[str, Any]", response) |
| 119 | + value = body.get("authorized_clients") |
| 120 | + return cast("list[Any]", value) if isinstance(value, list) else [] |
| 121 | + |
| 122 | + |
| 123 | +def _parse_authorized_clients(payload: Any) -> AuthorizedClients: |
| 124 | + """Parse a raw ``list_authorized_clients()`` response into typed clients.""" |
| 125 | + clients = [ |
| 126 | + _parse_client(cast("dict[str, Any]", entry)) |
| 127 | + for entry in _authorized_clients_list(payload) |
| 128 | + if isinstance(entry, dict) |
| 129 | + ] |
| 130 | + return AuthorizedClients(clients=clients, raw=payload) |
| 131 | + |
| 132 | + |
14 | 133 | class TeslemetryEnergySite(EnergySite): |
15 | 134 | """Teslemetry specific energy site.""" |
16 | 135 |
|
@@ -77,6 +196,17 @@ async def list_authorized_clients(self) -> dict[str, Any]: |
77 | 196 | f"api/1/energy_sites/{self.energy_site_id}/command/authorized_clients", |
78 | 197 | ) |
79 | 198 |
|
| 199 | + async def find_authorized_clients(self) -> AuthorizedClients: |
| 200 | + """List authorized clients on the energy gateway, parsed into a typed result. |
| 201 | +
|
| 202 | + Prefer this over :meth:`list_authorized_clients` for consumers that |
| 203 | + need to inspect the client list - it centralizes the response |
| 204 | + parsing (envelope unwrap, null-body handling, ``state`` typing) |
| 205 | + here instead of in the caller. See :class:`AuthorizedClients` for |
| 206 | + the exact semantics. |
| 207 | + """ |
| 208 | + return _parse_authorized_clients(await self.list_authorized_clients()) |
| 209 | + |
80 | 210 | async def remove_authorized_client( |
81 | 211 | self, params: dict[str, Any] | None = None |
82 | 212 | ) -> dict[str, Any]: |
|
0 commit comments