Coverage for foxplot / node.py: 100%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# 

4# SPDX-License-Identifier: Apache-2.0 

5 

6"""Internal node used to access data in interactive mode.""" 

7 

8from typing import Any, List, Union 

9 

10from .exceptions import FoxplotError 

11from .hot_series import HotSeries 

12from .series import Series 

13 

14 

15class Node: 

16 """Series data unpacked from input dictionaries.""" 

17 

18 _label: str 

19 

20 def __init__(self, label: str): 

21 """Initialize node with a label. 

22 

23 Args: 

24 label: Node label. 

25 """ 

26 self._label = label 

27 

28 def __getitem__(self, key): 

29 """Get item from node, either a child node or an indexed series (leaf). 

30 

31 Args: 

32 key: Key that identifies the child item. 

33 """ 

34 return self.__dict__[key] 

35 

36 def __repr__(self): 

37 """String representation of the node.""" 

38 keys = ", ".join( 

39 str(key) 

40 for key in self.__dict__ 

41 if isinstance(key, int) or not key.startswith("_") 

42 ) 

43 return f"{self._label}: [{keys}]" 

44 

45 def _get_child(self, keys: List[str]) -> Series: 

46 """Get leaf descendant in the tree from a list of keys. 

47 

48 Args: 

49 keys: List of keys uniquely identifying the leaf descendant. 

50 """ 

51 child = self.__dict__[keys[0]] 

52 if len(keys) > 1: 

53 return child._get_child(keys[1:]) 

54 if not isinstance(child, Series): 

55 raise FoxplotError(f"{child._label} is not a time series") 

56 return child 

57 

58 def items(self): 

59 """Iterate over (key, child) pairs of this node.""" 

60 for key, child in self.__dict__.items(): 

61 if isinstance(key, str) and key.startswith("_"): 

62 continue 

63 yield (key, child) 

64 

65 def values(self): 

66 """Iterate over child values of this node.""" 

67 for _, child in self.items(): 

68 yield child 

69 

70 def keys(self): 

71 """Iterate over child keys of this node.""" 

72 for key, _ in self.items(): 

73 yield key 

74 

75 def __iter__(self): 

76 """Iterate over child keys of this node.""" 

77 return self.keys() 

78 

79 def __len__(self): 

80 """Return the number of children.""" 

81 return sum(1 for _ in self.items()) 

82 

83 def _list_labels(self) -> List[str]: 

84 """List all labels reachable from this node.""" 

85 labels = [] 

86 for key, child in self.__dict__.items(): 

87 if isinstance(key, int) or key.startswith("_"): 

88 continue 

89 labels.extend(child._list_labels()) 

90 return labels 

91 

92 def _update(self, index: int, unpacked: Union[None, dict, list]) -> None: 

93 """Update node from a new unpacked dictionary. 

94 

95 Args: 

96 index: Index of the unpacked dictionary in the sequential input. 

97 unpacked: Unpacked dictionary. 

98 """ 

99 if unpacked is None: 

100 return 

101 items = ( 

102 unpacked.items() 

103 if isinstance(unpacked, dict) 

104 else enumerate(unpacked) 

105 ) 

106 # Typed as Any because list-shaped records produce int keys via 

107 # enumerate(), while self.__dict__ is dict[str, Any]. 

108 self_dict: Any = self.__dict__ 

109 for key, value in items: 

110 child = self_dict.get(key) 

111 if child is None: 

112 sep = "/" if not self._label.endswith("/") else "" 

113 is_primitive = not isinstance(value, (dict, list)) 

114 ChildClass = HotSeries if is_primitive else Node 

115 child = ChildClass(label=f"{self._label}{sep}{key}") 

116 self_dict[key] = child 

117 child._update(index, value) 

118 

119 def _freeze(self, max_index: int) -> None: 

120 update = {} 

121 for key, child in self.__dict__.items(): 

122 if isinstance(child, HotSeries): 

123 update[key] = child._freeze(max_index) 

124 elif isinstance(child, Node): 

125 child._freeze(max_index) 

126 self.__dict__.update(update)