Coverage for src/dev_shell/api_objtree.py: 90%
219 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-10-03 09:02 +0000
1"""ApiObjTree - API client mode's read-only `obj[...]`, built entirely from
2Home Assistant's standard websocket API (get_states, config/*_registry/list)
3- no dev_shell_server component needed, works against any instance an admin
4token can reach. Bound to the same name, `obj`, as the live tree (see
5apirepl.py) so a snippet that only touches `obj` runs unchanged in either
6mode. Entities only have what that API exposes: state + attributes from
7get_states, and whatever the entity registry's own partial dict exposes (no
8live component instance, so no calling methods on it, no fields an
9integration keeps off the registry/state).
11Mirrors custom_components/dev_shell_server/objtree.py's shape - __getitem__,
12.keys()/.values()/.items(), find(), show() all behave the same way modulo the
13smaller attribute set - but lives in this package rather than that one (a
14HACS-deployed component with its own packaging boundary), so the handful of
15small, stable, HA-import-free pieces it needs (the ordered-view mixin,
16parse_path, the None/empty-collection cleanup) are duplicated here rather than
17imported across that boundary. Revisit if API client mode sticks and a shared
18package becomes worth the packaging work.
20Unlike the live tree, this one is explicitly a *snapshot*: fetching is async
21(it's a websocket call) but Mapping's `__getitem__`/`__iter__`/`__len__` are
22not, so refreshing can't happen lazily on access without risking asyncio
23reentrancy (this is normally used from inside an already-running event loop).
24Instead the owning REPL loop decides when to call `await cache.refresh()`
25(e.g. between prompts, when `cache.is_stale()`) - see apirepl.py. `reset_cache()`
26just marks the cache stale; it does not fetch.
27"""
29from __future__ import annotations
31import time
32from collections.abc import (
33 ItemsView,
34 Iterable,
35 Iterator,
36 KeysView,
37 Mapping,
38 Sequence,
39 ValuesView,
40)
41from dataclasses import dataclass
42from datetime import datetime
43from typing import Any, NamedTuple, Protocol
46class _ApiClient(Protocol):
47 """What Cache actually needs from a client - just enough to let tests use
48 a lightweight fake instead of a real websocket Client."""
50 async def call(self, type_: str, **payload: Any) -> Any: ...
53def parse_path(path: str) -> tuple[str, ...]:
54 """Split "/integration[/domain[/object_id]]" into 1-3 parts. Duplicated
55 from custom_components/dev_shell_server/paths.py - see the module
56 docstring for why."""
57 parts = tuple(p for p in path.split("/") if p)
58 if not parts or len(parts) > 3: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true
59 raise ValueError(f"{path!r} - expected /integration/domain/object_id")
60 return parts
63def _as_set(value: str | list[str] | None) -> set[str]:
64 if value is None:
65 return set()
66 return {value} if isinstance(value, str) else set(value)
69def _require_str(value: Any, what: str = "path") -> str:
70 """A clear TypeError at the boundary beats a confusing one from deep
71 inside parse_path - duplicated from objtree.py's identical helper, see
72 that module for the rationale."""
73 if not isinstance(value, str):
74 raise TypeError(f"{what} must be a str, not {type(value).__name__}")
75 return value
78class _OrderedView(Iterable[Any]):
79 """Positional access/slicing on top of a MappingView's existing order -
80 duplicated from objtree.py's identical mixin, see that module for the
81 rationale."""
83 def __getitem__(self, index: int | slice) -> Any:
84 return tuple(self)[index]
86 def __reversed__(self) -> Iterator[Any]:
87 return reversed(tuple(self))
90class OrderedKeysView(_OrderedView, KeysView, Sequence):
91 pass
94class OrderedValuesView(_OrderedView, ValuesView, Sequence):
95 pass
98class OrderedItemsView(_OrderedView, ItemsView, Sequence): # type: ignore[misc] # ty: ignore[invalid-method-override]
99 # Sequence.__contains__(value) vs ItemsView.__contains__(item: tuple) is a
100 # real static Liskov mismatch, but fine at runtime - both just delegate to
101 # tuple(self).__contains__ via _OrderedView/Iterable, deliberately getting
102 # both Set and Sequence behavior at once (see the module docstring).
103 pass
106@dataclass(frozen=True)
107class ApiEntity:
108 """One entity's worth of API client mode data - only what get_states and
109 the entity registry's partial dict expose, not the live component instance.
111 Named to match the real Entity's own properties where there's a direct
112 equivalent - state, state_attributes, name - per
113 https://developers.home-assistant.io/docs/core/entity/. `name` comes from
114 get_states' `friendly_name` attribute: that's already HA's own fully
115 resolved result of computing Entity.name (registry override, device name,
116 has_entity_name, etc.), published into the state machine for exactly this
117 kind of external consumer - cheaper and more faithful than recomputing it
118 from the pieces ourselves. `state_attributes` is everything else in that
119 attributes dict; broader than the live property of the same name (which
120 excludes things like unit_of_measurement that come from separate Entity
121 properties), since the state machine doesn't preserve that distinction.
122 """
124 entity_id: str
125 platform: str
126 domain: str
127 object_id: str
128 state: str | None
129 name: str | None
130 state_attributes: dict[str, Any]
131 area_id: str | None
132 labels: frozenset[str]
133 registry: dict[str, Any]
136class _Found(NamedTuple):
137 """Internal only - never returned from find()/find_paths()/find_names(),
138 just the shared (path, entity) pair each of them projects from
139 (.entity, .path, and .entity_id respectively) so the filtering logic in
140 _find() lives in exactly one place. Any attribute other than path/entity
141 falls through to `entity`, so e.g. found.entity_id works without
142 found.entity.entity_id."""
144 path: str
145 entity: ApiEntity
147 def __getattr__(self, name: str) -> Any:
148 return getattr(self.entity, name)
151# The two registry JSON fields that are Unix-timestamp floats rather than the
152# ISO datetime strings get_states already returns - see show()/_clean().
153_TIMESTAMP_FIELDS = frozenset({"created_at", "modified_at"})
156class Cache:
157 """Fetches and holds one snapshot of states + registries. Refresh is
158 driven by the REPL loop between commands - see the module docstring."""
160 def __init__(self, client: _ApiClient, ttl: float) -> None:
161 self.client = client
162 self.ttl = ttl
163 self._fetched_at: float | None = None
164 self.entities: dict[str, ApiEntity] = {}
165 self.areas: dict[str, dict[str, Any]] = {}
166 self.labels: dict[str, dict[str, Any]] = {}
168 def is_stale(self) -> bool:
169 return (
170 self._fetched_at is None or (time.monotonic() - self._fetched_at) > self.ttl
171 )
173 def reset(self) -> None:
174 """Mark the cache stale. Doesn't fetch - see the module docstring."""
175 self._fetched_at = None
177 async def refresh(self) -> None:
178 states = {s["entity_id"]: s for s in await self.client.call("get_states")}
179 registry_entries = await self.client.call("config/entity_registry/list")
180 devices = {
181 d["id"]: d for d in await self.client.call("config/device_registry/list")
182 }
183 self.areas = {
184 a["area_id"]: a for a in await self.client.call("config/area_registry/list")
185 }
186 self.labels = {
187 l["label_id"]: l
188 for l in await self.client.call("config/label_registry/list")
189 }
191 entities: dict[str, ApiEntity] = {}
192 for entry in registry_entries:
193 entity_id = entry["entity_id"]
194 domain, object_id = entity_id.split(".", 1)
195 state = states.get(entity_id, {})
196 area_id = entry.get("area_id")
197 if area_id is None and entry.get("device_id"):
198 device = devices.get(entry["device_id"])
199 if device is not None: 199 ↛ 205line 199 didn't jump to line 205 because the condition on line 199 was always true
200 area_id = device.get("area_id")
201 # friendly_name is already HA's own fully-resolved Entity.name,
202 # published for exactly this kind of external consumer - see
203 # ApiEntity's docstring. Copy the dict before popping: it's
204 # `states[entity_id]`'s own attributes dict, not ours to mutate.
205 attributes = dict(state.get("attributes", {}))
206 name = attributes.pop("friendly_name", None)
207 entities[entity_id] = ApiEntity(
208 entity_id=entity_id,
209 platform=entry["platform"],
210 domain=domain,
211 object_id=object_id,
212 state=state.get("state"),
213 name=name,
214 state_attributes=attributes,
215 area_id=area_id,
216 labels=frozenset(entry.get("labels", ())),
217 registry=entry,
218 )
219 self.entities = entities
220 self._fetched_at = time.monotonic()
223def _resolve_from_cache(
224 values: str | list[str] | None,
225 by_id: dict[str, dict[str, Any]],
226 kind: str,
227 id_key: str,
228) -> set[str]:
229 """Resolve each of `values` (an id or a display name) to its canonical id.
231 Raises KeyError on anything that matches neither - a typo'd area/label
232 name should fail loudly, not just quietly match nothing in find().
233 """
234 ids: set[str] = set()
235 for value in _as_set(values):
236 if value in by_id:
237 ids.add(value)
238 continue
239 match = next(
240 (
241 k
242 for k, v in by_id.items()
243 if v.get("name", "").casefold() == value.casefold()
244 ),
245 None,
246 )
247 if match is None:
248 raise KeyError(f"no such {kind}: {value!r}")
249 ids.add(match)
250 return ids
253def _normalize(value: Any) -> Any:
254 if isinstance(value, dict):
255 return _clean(value)
256 return value
259def _clean(data: dict[str, Any]) -> dict[str, Any]:
260 """Recursively drop None/empty-collection values and render the known
261 timestamp fields as local ISO 8601 strings.
263 Empty *lists* are dropped here too, not just empty sets: labels/aliases
264 etc. arrive over the wire as JSON lists (JSON has no set type), so a list
265 is this mode's wire-equivalent of the live tree's empty-set fields - and
266 keeping both modes behaving the same way is the point.
267 """
268 cleaned: dict[str, Any] = {}
269 for key, value in data.items():
270 if key in _TIMESTAMP_FIELDS and isinstance(value, (int, float)):
271 value = datetime.fromtimestamp(value).astimezone().isoformat()
272 else:
273 value = _normalize(value)
274 if value is None or value == set() or value == []:
275 continue
276 cleaned[key] = value
277 return cleaned
280@dataclass(frozen=True)
281class ApiObjTree(Mapping[str, "ApiEntity | ApiObjTree"]):
282 """API client mode's view of the object tree - see the module docstring."""
284 cache: Cache
285 integration: str | None = None
286 domain: str | None = None
288 def reset_cache(self) -> None:
289 self.cache.reset()
291 def __getitem__(self, key: str) -> ApiEntity | ApiObjTree:
292 _require_str(key, "key")
293 if "/" not in key and self.integration is None:
294 entity = self.cache.entities.get(key)
295 if entity is None: 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true
296 raise KeyError(key)
297 return entity
298 try:
299 parts = parse_path(key)
300 except ValueError as err:
301 raise KeyError(str(err)) from None
302 scope = tuple(p for p in (self.integration, self.domain) if p is not None)
303 full = scope + parts
304 if len(full) > 3: 304 ↛ 305line 304 didn't jump to line 305 because the condition on line 304 was never true
305 raise KeyError(key)
306 if len(full) < 3:
307 subtree = ApiObjTree(self.cache, *full)
308 if not subtree:
309 raise KeyError(key)
310 return subtree
311 integration, domain, object_id = full
312 entity = self.cache.entities.get(f"{domain}.{object_id}")
313 if entity is None or entity.platform != integration: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 raise KeyError(key)
315 return entity
317 def _entries(self, prefix: tuple[str, ...]) -> Iterator[ApiEntity]:
318 for entity in self.cache.entities.values():
319 if len(prefix) >= 1 and entity.platform != prefix[0]:
320 continue
321 if len(prefix) >= 2 and entity.domain != prefix[1]:
322 continue
323 if len(prefix) >= 3 and entity.object_id != prefix[2]: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 continue
325 yield entity
327 def __iter__(self) -> Iterator[str]:
328 scope = tuple(p for p in (self.integration, self.domain) if p is not None)
329 children = {
330 (entity.platform, entity.domain, entity.object_id)[len(scope)]
331 for entity in self._entries(scope)
332 }
333 for child in sorted(children):
334 yield "/" + child
336 def __len__(self) -> int:
337 return sum(1 for _ in self)
339 def keys(self) -> OrderedKeysView:
340 return OrderedKeysView(self)
342 def values(self) -> OrderedValuesView:
343 return OrderedValuesView(self)
345 def items(self) -> OrderedItemsView:
346 return OrderedItemsView(self)
348 def _find(
349 self,
350 path: str,
351 *,
352 platform: str | list[str] | None,
353 domain: str | list[str] | None,
354 area: str | list[str] | None,
355 label: str | list[str] | None,
356 ) -> Iterator[_Found]:
357 """The real search, shared by find()/find_paths()/find_names() - each
358 just projects a different field from the (path, entity) pairs this
359 yields. `domain` matches the HA domain (e.g. "light") across
360 integrations, the same way `platform` does for the owning
361 integration; `area`/`label` each take an id or a display name. Each
362 of the four ORs within itself when given a list, and they AND
363 together. No order guarantee; wrap in sorted(...) if you want one.
364 """
365 _require_str(path)
366 scope = tuple(p for p in (self.integration, self.domain) if p is not None)
367 try:
368 prefix = scope if path in ("", "/") else scope + parse_path(path)
369 except ValueError as err:
370 raise KeyError(str(err)) from None
371 if len(prefix) > 3: 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 raise KeyError(path)
374 platforms = _as_set(platform)
375 domains = _as_set(domain)
376 area_ids = _resolve_from_cache(area, self.cache.areas, "area", "area_id")
377 label_ids = _resolve_from_cache(label, self.cache.labels, "label", "label_id")
379 def _matches() -> Iterator[_Found]:
380 for entity in self._entries(prefix):
381 if platforms and entity.platform not in platforms:
382 continue
383 if domains and entity.domain not in domains:
384 continue
385 if area_ids and entity.area_id not in area_ids:
386 continue
387 if label_ids and not (entity.labels & label_ids):
388 continue
389 full_path = "/" + "/".join(
390 (entity.platform, entity.domain, entity.object_id)[len(scope) :]
391 )
392 yield _Found(full_path, entity)
394 return _matches()
396 def find(
397 self,
398 path: str = "/",
399 *,
400 platform: str | list[str] | None = None,
401 domain: str | list[str] | None = None,
402 area: str | list[str] | None = None,
403 label: str | list[str] | None = None,
404 raw: bool = False,
405 ) -> Iterator[ApiEntity] | Iterator[dict[str, Any]]:
406 """Every entity (or raw dict, if raw=True) matching the filters under
407 `path` (relative to this view, "/" meaning this view's whole
408 subtree) - flat, skipping the directory-style one-level-at-a-time
409 grouping .keys()/indexing give you. find_paths()/find_names() are the
410 same search with just the tree-path or entity_id strings, if that's
411 all you want - see _find() for the filters.
413 `raw=True` yields the object as received instead of the typed,
414 renamed-to-match-Entity ApiEntity view - exactly what show(path)
415 would return for that path (the merged, cleaned registry+state
416 dict). The dict's own "entity_id" key identifies which entity it
417 came from.
418 """
419 matches = self._find(
420 path, platform=platform, domain=domain, area=area, label=label
421 )
422 if raw:
423 return (self.show(found.path) for found in matches)
424 return (found.entity for found in matches)
426 def find_paths(
427 self,
428 path: str = "/",
429 *,
430 platform: str | list[str] | None = None,
431 domain: str | list[str] | None = None,
432 area: str | list[str] | None = None,
433 label: str | list[str] | None = None,
434 ) -> Iterator[str]:
435 """Just the tree-path strings from find() (e.g.
436 "/demo/light/kitchen_lights") - see _find() for the filters."""
437 return (
438 found.path
439 for found in self._find(
440 path, platform=platform, domain=domain, area=area, label=label
441 )
442 )
444 def find_names(
445 self,
446 path: str = "/",
447 *,
448 platform: str | list[str] | None = None,
449 domain: str | list[str] | None = None,
450 area: str | list[str] | None = None,
451 label: str | list[str] | None = None,
452 ) -> Iterator[str]:
453 """Just the HA entity_id strings from find() (e.g.
454 "light.kitchen_lights") - see _find() for the filters. Not the same
455 as find_paths(): this is the flat entity_id, not this tree's
456 /integration/domain/object_id path."""
457 return (
458 found.entity_id
459 for found in self._find(
460 path, platform=platform, domain=domain, area=area, label=label
461 )
462 )
464 def show(self, path: str) -> dict[str, Any]:
465 """Same shape as the live tree's show() - the registry entry's own
466 fields plus `state`/`state_attributes`, cleaned the same way (None/
467 empty-collection dropped, timestamps as local ISO 8601)."""
468 _require_str(path)
469 scope = tuple(p for p in (self.integration, self.domain) if p is not None)
470 try:
471 full = scope + parse_path(path)
472 except ValueError as err:
473 raise KeyError(str(err)) from None
474 if len(full) != 3: 474 ↛ 475line 474 didn't jump to line 475 because the condition on line 474 was never true
475 raise KeyError(f"{path!r} - show() needs a full entity path")
476 integration, domain, object_id = full
477 entity = self.cache.entities.get(f"{domain}.{object_id}")
478 if entity is None or entity.platform != integration: 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true
479 raise KeyError(path)
481 data: dict[str, Any] = dict(entity.registry)
482 data["name"] = entity.name
483 data["state"] = entity.state
484 data["state_attributes"] = entity.state_attributes
485 return _clean(data)
487 def __repr__(self) -> str:
488 # "(api)" is just a display hint for a human reading output - the
489 # binding name and API are identical to the live tree's on purpose.
490 scope = "/".join(p for p in (self.integration, self.domain) if p is not None)
491 label = f"obj:/{scope}" if scope else "obj"
492 kind = (
493 "entities"
494 if self.domain is not None
495 else "domains"
496 if self.integration
497 else "integrations"
498 )
499 return f"<{label}: {len(self)} {kind} (api)>"