Coverage for src/bayernwerk_client/map/instances.py: 100%

1"""Schema-less helpers for the order-instances tree (`MapClient.get_order_instances`). 

2 

3That tree (the "Anschluss" hierarchy in the portal UI - meter, its 

4consumers/generators, inverters, batteries, ...) nests items under 

5different key names depending on `itemType` (`meters`, `actors`, 

6`inverterGroups`, `inverters`, ...), and each `schemaType` 

7(`WALLBOX`, `HEATING_PUMP`, `BATTERY`, `PV`, ...) has its own `formData` 

8shape. Bayernwerk can add new equipment types or nesting at any time, so 

9modelling every current shape as a dataclass would mean constantly chasing 

10schema changes for little benefit. `iter_instance_items` instead walks the 

11raw JSON generically: any dict with an `itemType` key is a node, and every 

12nested list/dict is searched for more nodes, regardless of the key it's 

13stored under. 

14 

15Some actors appear twice in the raw data (once under an `inverterGroup`'s 

16own `actors`, again nested inside that group's `inverters[].actors`) - the 

17same physical component referenced via two different nesting paths. 

18`iter_instance_items` dedups by `(itemType, ivyId)` (present on every node 

19type observed so far) so callers see each real component once. 

20""" 

21 

22from __future__ import annotations 

23 

24from collections.abc import Iterator 

25from typing import Any 

26 

27 

28def iter_instance_items(tree: Any) -> Iterator[dict[str, Any]]: 

29 """Recursively yield every node (instance/meter/actor/inverter/...) in an 

30 order-instances tree, regardless of which key it's nested under.""" 

31 yield from _walk(tree, seen=set()) 

32 

33 

34def _walk(node: Any, *, seen: set[tuple[Any, Any]]) -> Iterator[dict[str, Any]]: 

35 if isinstance(node, list): 

36 for item in node: 

37 yield from _walk(item, seen=seen) 

38 return 

39 if not isinstance(node, dict): 

40 return 

41 

42 if "itemType" in node: 

43 ivy_id = node.get("ivyId") 

44 key = (node.get("itemType"), ivy_id) 

45 already_seen = ivy_id is not None and key in seen 

46 if not already_seen: 

47 if ivy_id is not None: 

48 seen.add(key) 

49 yield node 

50 

51 for value in node.values(): 

52 if isinstance(value, (list, dict)): 

53 yield from _walk(value, seen=seen)