Coverage for custom_components/dev_shell_server/objtree.py: 20%
201 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-10-03 09:02 +0000
1"""`obj[...]` - a live, read-only lookup over Home Assistant's entities.
3Entities only for now; the full integration/domain/area/label tree from the
4README's Object Browser design is a later step. Lookup forms:
6 obj["light.kitchen_lights"] # plain entity_id
7 obj["/demo/light/kitchen_lights"] # /integration/domain/object_id
9A path that doesn't reach all the way to an object_id instead returns an
10ObjTree restricted to that part of the tree, so the object browser's
11directories can be navigated a level at a time:
13 alexa = obj["/alexa_devices"] # restricted to that integration
14 alexa["media_player"] # -> restricted to that domain too
15 alexa["media_player/kitchen_show"] # -> the live entity
17`.keys()`/iteration mirror that: `alexa.keys()` lists just the domains directly
18under `/alexa_devices` (one level, like `ls` on a directory), not every entity
19underneath it recursively. `obj.find()` is the flat alternative - every
20(path, entity) pair under a subtree, optionally filtered by
21platform/domain/area/label; `obj.find_paths()` is the same search with just
22the tree path strings (e.g. "/demo/light/kitchen_lights"), and
23`obj.find_names()` with just the HA entity_id strings (e.g.
24"light.kitchen_lights") instead. `obj.show(path)` renders one entity's
25registry entry plus its live state.
27Both return the live Entity object - the actual LightEntity/SensorEntity/etc.
28instance a component wrote, not the hass.states.get() State snapshot - since this
29is strict mode: you use it exactly as you would from component code.
31Only entities in the entity registry are reachable by the /integration/... path,
32since that's where the owning integration ("platform") is recorded; legacy YAML
33entities without a unique_id aren't registered there. Both lookup forms require a
34live Entity object backing the entity - true for everything set up the normal way
35through an EntityPlatform (YAML or config-entry), but not for a bare state set
36directly via hass.states.async_set() with no Entity subclass behind it (rare
37outside of tests/templates). See the README's Object Browser enhancements for the
38registry-less follow-up.
39"""
41from __future__ import annotations
43from collections.abc import (
44 Callable,
45 ItemsView,
46 Iterable,
47 Iterator,
48 KeysView,
49 Mapping,
50 Sequence,
51 ValuesView,
52)
53from dataclasses import dataclass
54from datetime import datetime
55from typing import Any, NamedTuple
57from homeassistant.core import HomeAssistant
58from homeassistant.helpers import area_registry as ar
59from homeassistant.helpers import device_registry as dr
60from homeassistant.helpers import entity_registry as er
61from homeassistant.helpers import label_registry as lr
62from homeassistant.helpers.entity import Entity
63from homeassistant.helpers.entity_platform import DATA_DOMAIN_ENTITIES
64from homeassistant.util import dt as dt_util
66from .paths import parse_path
69def _get_entity(hass: HomeAssistant, entity_id: str) -> Entity | None:
70 """The live Entity instance behind entity_id, or None.
72 hass.data[DATA_DOMAIN_ENTITIES] is maintained by EntityPlatform itself (for
73 every platform, YAML or config-entry) as {domain: {entity_id: Entity}} - the
74 one lookup that's both centralized and independent of which integration or
75 setup style owns the entity. It's a private implementation detail of
76 homeassistant.helpers.entity_platform, not public API, so this is written
77 defensively (plain dict .get chains) rather than assuming its shape.
78 """
79 domain = entity_id.partition(".")[0]
80 return hass.data.get(DATA_DOMAIN_ENTITIES, {}).get(domain, {}).get(entity_id)
83def _as_set(value: str | list[str] | None) -> set[str]:
84 if value is None:
85 return set()
86 return {value} if isinstance(value, str) else set(value)
89def _require_str(value: Any, what: str = "path") -> str:
90 """A clear TypeError at the boundary beats a confusing one from deep
91 inside parse_path (e.g. passing an Entity instead of its path - an easy
92 slip, since indexing returns one and it's natural to then pass it back
93 in somewhere a path string was wanted)."""
94 if not isinstance(value, str):
95 raise TypeError(f"{what} must be a str, not {type(value).__name__}")
96 return value
99class _Found(NamedTuple):
100 """Internal only - never returned from find()/find_paths()/find_names(),
101 just the shared (path, entity) pair each of them projects from
102 (.entity, .path, and .entity_id respectively) so the filtering logic in
103 _find() lives in exactly one place. Any attribute other than path/entity
104 falls through to `entity`, so e.g. found.entity_id works without
105 found.entity.entity_id."""
107 path: str
108 entity: Entity
110 def __getattr__(self, name: str) -> Any:
111 return getattr(self.entity, name)
114def _resolve_ids(
115 values: str | list[str] | None,
116 kind: str,
117 by_id: Callable[[str], Any],
118 by_name: Callable[[str], Any],
119 id_attr: str,
120) -> set[str]:
121 """Resolve each of `values` (an id or a display name) to its canonical id.
123 Raises KeyError on anything that matches neither - a typo'd area/label
124 name should fail loudly, not just quietly match nothing in find().
125 """
126 ids: set[str] = set()
127 for value in _as_set(values):
128 found = by_id(value) or by_name(value)
129 if found is None:
130 raise KeyError(f"no such {kind}: {value!r}")
131 ids.add(getattr(found, id_attr))
132 return ids
135def _entity_area_id(entry: er.RegistryEntry, devices: dr.DeviceRegistry) -> str | None:
136 """The entry's own area, falling back to its device's - the same rule the
137 frontend uses to decide which area an entity "is in"."""
138 if entry.area_id is not None:
139 return entry.area_id
140 if entry.device_id is not None:
141 device = devices.async_get(entry.device_id)
142 if device is not None:
143 return device.area_id
144 return None
147def _public_attrs(obj: Any) -> dict[str, Any]:
148 """Every attrs-declared field on `obj`, minus _cache and anything else
149 marked private by the leading-underscore convention attrs itself uses for
150 fields like RegistryEntry._cache."""
151 return {
152 a.name: getattr(obj, a.name)
153 for a in getattr(obj, "__attrs_attrs__", ())
154 if not a.name.startswith("_")
155 }
158def _normalize(value: Any) -> Any:
159 if isinstance(value, datetime):
160 return dt_util.as_local(value).isoformat()
161 if isinstance(value, Mapping):
162 return _clean(dict(value))
163 return value
166def _clean(data: dict[str, Any]) -> dict[str, Any]:
167 """Recursively drop None/empty-set values and render datetimes as local
168 ISO 8601 strings - applied throughout, not just at the top level, since
169 things like state_attributes are themselves dicts that can hold either."""
170 cleaned = {}
171 for key, value in data.items():
172 value = _normalize(value)
173 if value is None or value == set():
174 continue
175 cleaned[key] = value
176 return cleaned
179class _OrderedView(Iterable[Any]):
180 """Adds positional access/slicing on top of a MappingView's existing order.
182 ObjTree.__iter__ already yields children in a meaningful (alphanumeric)
183 order, so the views built on it can afford to be real Sequences too - not
184 just the usual Collection (ValuesView) or Set (KeysView/ItemsView) - without
185 giving up set arithmetic or `isinstance(x, KeysView)` duck-typing: this is a
186 mixin, not a replacement, so a class using it keeps its MappingView base's
187 behaviour and only adds `__getitem__`/`__reversed__` on top.
189 Inherits Iterable (abstract, no `__iter__` of its own) only so type checkers
190 know `self` supports `tuple(self)` here - the concrete classes below supply
191 the real `__iter__` via their MappingView base, same as at runtime.
192 """
194 def __getitem__(self, index: int | slice) -> Any:
195 return tuple(self)[index]
197 def __reversed__(self) -> Iterator[Any]:
198 return reversed(tuple(self))
201class OrderedKeysView(_OrderedView, KeysView, Sequence):
202 pass
205class OrderedValuesView(_OrderedView, ValuesView, Sequence):
206 pass
209class OrderedItemsView(_OrderedView, ItemsView, Sequence): # type: ignore[misc] # ty: ignore[invalid-method-override]
210 # Sequence.__contains__(value) vs ItemsView.__contains__(item: tuple) is a
211 # real static Liskov mismatch, but fine at runtime - both just delegate to
212 # tuple(self).__contains__ via _OrderedView/Iterable, deliberately getting
213 # both Set and Sequence behavior at once.
214 pass
217@dataclass(frozen=True)
218class ObjTree(Mapping[str, "Entity | ObjTree"]):
219 """A view of the object tree, optionally restricted to a subtree.
221 `integration` and/or `domain` pin this view to that part of the tree -
222 the way `obj["/alexa_devices"]` or `obj["/alexa_devices/media_player"]`
223 does. Indexing a restricted view only needs the remaining path segments,
224 given either as a single "a/b" string or one segment at a time.
226 Implementing collections.abc.Mapping (on top of __getitem__, __iter__ and
227 __len__) gets `in`, .get(), and real KeysView/ItemsView/ValuesView from
228 .keys()/.items()/.values() for free, consistent with any other dict-like
229 object - overridden below to also be Sequences, since every level in this
230 tree is naturally a list (see _OrderedView): `obj["/alexa_devices"].keys()[0]`
231 and slicing work, alongside the usual Set/Collection behaviour.
232 """
234 hass: HomeAssistant
235 integration: str | None = None
236 domain: str | None = None
238 def __getitem__(self, key: str) -> Entity | ObjTree:
239 _require_str(key, "key")
240 if "/" not in key and self.integration is None:
241 entity = _get_entity(self.hass, key)
242 if entity is None:
243 raise KeyError(key)
244 return entity
245 try:
246 parts = parse_path(key)
247 except ValueError as err:
248 raise KeyError(str(err)) from None
249 scope = tuple(p for p in (self.integration, self.domain) if p is not None)
250 full = scope + parts
251 if len(full) > 3:
252 raise KeyError(key)
253 if len(full) < 3:
254 subtree = ObjTree(self.hass, *full)
255 # Without this, `in`/.get() (Mapping's default __contains__ tries
256 # self[key]) would say yes to any made-up integration/domain name,
257 # disagreeing with .keys() - which only ever lists ones with entities.
258 if not subtree:
259 raise KeyError(key)
260 return subtree
261 entity = self._entity_at(*full)
262 if entity is None:
263 raise KeyError(key)
264 return entity
266 def _entity_at(
267 self, integration: str, domain: str, object_id: str
268 ) -> Entity | None:
269 entity_id = f"{domain}.{object_id}"
270 entry = er.async_get(self.hass).async_get(entity_id)
271 if entry is None or entry.platform != integration:
272 return None
273 return _get_entity(self.hass, entity_id)
275 def _entries(
276 self, prefix: tuple[str, ...]
277 ) -> Iterator[tuple[er.RegistryEntry, str, str]]:
278 """Every (registry entry, domain, object_id) whose (platform, domain,
279 object_id) matches `prefix` position by position - `prefix` may have
280 0-3 elements, a narrower prefix just matching fewer positions. The
281 shared scan both __iter__ (one level, deduped) and find() (full depth,
282 flat) are built on.
283 """
284 registry = er.async_get(self.hass)
285 domain_entities = self.hass.data.get(DATA_DOMAIN_ENTITIES, {})
286 for domain, entities in domain_entities.items():
287 if len(prefix) >= 2 and domain != prefix[1]:
288 continue
289 for entity_id in entities:
290 entry = registry.async_get(entity_id)
291 if entry is None:
292 continue
293 if len(prefix) >= 1 and entry.platform != prefix[0]:
294 continue
295 object_id = entity_id.split(".", 1)[1]
296 if len(prefix) >= 3 and object_id != prefix[2]:
297 continue
298 yield entry, domain, object_id
300 def __iter__(self) -> Iterator[str]:
301 """The immediate child names one level down, alphanumeric order - a
302 directory listing, not a recursive flattening down to every entity
303 (that's find(), a different, explicitly flat view, not this one)."""
304 scope = tuple(p for p in (self.integration, self.domain) if p is not None)
305 children = {
306 (entry.platform, domain, object_id)[len(scope)]
307 for entry, domain, object_id in self._entries(scope)
308 }
309 yield from sorted(children)
311 def _find(
312 self,
313 path: str,
314 *,
315 platform: str | list[str] | None,
316 domain: str | list[str] | None,
317 area: str | list[str] | None,
318 label: str | list[str] | None,
319 ) -> Iterator[_Found]:
320 """The real search, shared by find()/find_paths()/find_names() - each
321 just projects a different field from the (path, entity) pairs this
322 yields. Skips anything with no live Entity object backing it - same
323 requirement as indexing - so an entity id from here always works if
324 passed back into this view.
326 `platform` matches the registry entry's platform (owning integration)
327 directly; `domain` matches the HA domain (e.g. "light"), letting you
328 search across integrations without fixing `path` to one; `area`/`label`
329 each take an id or a display name, and an entity's area falls back to
330 its device's when it has none of its own - the same rule the frontend
331 uses. Each of the four ORs within itself when given a list, and they
332 AND together. No order guarantee; wrap in sorted(...) if you want one.
333 """
334 _require_str(path)
335 scope = tuple(p for p in (self.integration, self.domain) if p is not None)
336 try:
337 prefix = scope if path in ("", "/") else scope + parse_path(path)
338 except ValueError as err:
339 raise KeyError(str(err)) from None
340 if len(prefix) > 3:
341 raise KeyError(path)
343 platforms = _as_set(platform)
344 domains = _as_set(domain)
345 areas = ar.async_get(self.hass)
346 labels = lr.async_get(self.hass)
347 area_ids = _resolve_ids(
348 area, "area", areas.async_get_area, areas.async_get_area_by_name, "id"
349 )
350 label_ids = _resolve_ids(
351 label,
352 "label",
353 labels.async_get_label,
354 labels.async_get_label_by_name,
355 "label_id",
356 )
357 devices = dr.async_get(self.hass)
359 def _matches() -> Iterator[_Found]:
360 for entry, dom, object_id in self._entries(prefix):
361 if platforms and entry.platform not in platforms:
362 continue
363 if domains and dom not in domains:
364 continue
365 if area_ids and _entity_area_id(entry, devices) not in area_ids:
366 continue
367 if label_ids and not (entry.labels & label_ids):
368 continue
369 entity = _get_entity(self.hass, entry.entity_id)
370 if entity is None:
371 continue
372 full_path = "/" + "/".join(
373 (entry.platform, dom, object_id)[len(scope) :]
374 )
375 yield _Found(full_path, entity)
377 return _matches()
379 def find(
380 self,
381 path: str = "/",
382 *,
383 platform: str | list[str] | None = None,
384 domain: str | list[str] | None = None,
385 area: str | list[str] | None = None,
386 label: str | list[str] | None = None,
387 raw: bool = False,
388 ) -> Iterator[Entity] | Iterator[dict[str, Any]]:
389 """Every entity (or raw dict, if raw=True) matching the filters under
390 `path` (relative to this view, "/" meaning this view's whole
391 subtree) - flat, skipping the directory-style one-level-at-a-time
392 grouping .keys()/indexing give you. find_paths()/find_names() are the
393 same search with just the tree-path or entity_id strings, if that's
394 all you want - see _find() for the filters.
396 `raw=True` yields the object as received instead of the live Entity -
397 exactly what show(path) would return for that path (the registry
398 entry's fields plus state/state_attributes, cleaned the same way).
399 The dict's own "entity_id" key identifies which entity it came from.
400 """
401 matches = self._find(
402 path, platform=platform, domain=domain, area=area, label=label
403 )
404 if raw:
405 return (self.show(found.path) for found in matches)
406 return (found.entity for found in matches)
408 def find_paths(
409 self,
410 path: str = "/",
411 *,
412 platform: str | list[str] | None = None,
413 domain: str | list[str] | None = None,
414 area: str | list[str] | None = None,
415 label: str | list[str] | None = None,
416 ) -> Iterator[str]:
417 """Just the tree-path strings from find() (e.g.
418 "/demo/light/kitchen_lights") - see _find() for the filters."""
419 return (
420 found.path
421 for found in self._find(
422 path, platform=platform, domain=domain, area=area, label=label
423 )
424 )
426 def find_names(
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 HA entity_id strings from find() (e.g.
436 "light.kitchen_lights") - see _find() for the filters. Not the same
437 as find_paths(): this is the flat entity_id, not this tree's
438 /integration/domain/object_id path."""
439 return (
440 found.entity_id
441 for found in self._find(
442 path, platform=platform, domain=domain, area=area, label=label
443 )
444 )
446 def show(self, path: str) -> dict[str, Any]:
447 """The registry entry's own fields (minus _cache and other data this
448 hides by the leading-underscore convention) plus the live entity's
449 `state` and `state_attributes` - meant to be the trailing expression
450 at the REPL, so the usual Pretty-printed echo renders it; this doesn't
451 print anything itself. Datetimes come back as local ISO 8601 strings,
452 and None/empty-set values are dropped throughout, including inside
453 nested dicts like state_attributes.
454 """
455 _require_str(path)
456 scope = tuple(p for p in (self.integration, self.domain) if p is not None)
457 try:
458 full = scope + parse_path(path)
459 except ValueError as err:
460 raise KeyError(str(err)) from None
461 if len(full) != 3:
462 raise KeyError(f"{path!r} - show() needs a full entity path")
463 integration, domain, object_id = full
464 entity = self._entity_at(integration, domain, object_id)
465 if entity is None:
466 raise KeyError(path)
468 entry = er.async_get(self.hass).async_get(f"{domain}.{object_id}")
469 data: dict[str, Any] = _public_attrs(entry) if entry is not None else {}
470 # Overrides the registry's own (often-None, user-override-only) "name"
471 # field with the live, fully-resolved display name - the direct
472 # equivalent per https://developers.home-assistant.io/docs/core/entity/.
473 data["name"] = entity.name
474 data["state"] = entity.state
475 data["state_attributes"] = entity.state_attributes
476 return _clean(data)
478 def __len__(self) -> int:
479 return sum(1 for _ in self)
481 def keys(self) -> OrderedKeysView:
482 return OrderedKeysView(self)
484 def values(self) -> OrderedValuesView:
485 return OrderedValuesView(self)
487 def items(self) -> OrderedItemsView:
488 return OrderedItemsView(self)
490 def __repr__(self) -> str:
491 scope = "/".join(p for p in (self.integration, self.domain) if p is not None)
492 label = f"obj:/{scope}" if scope else "obj"
493 kind = (
494 "entities"
495 if self.domain is not None
496 else "domains"
497 if self.integration
498 else "integrations"
499 )
500 return f"<{label}: {len(self)} {kind}>"