Coverage for custom_components / supernotify / common.py: 96%

1"""Miscellaneous helper functions. 

2 

3No same pkg dependencies permitted 

4""" 

5 

6from __future__ import annotations 

7 

8import datetime as dt 

9import logging 

10from abc import abstractmethod 

11from collections.abc import KeysView 

12from dataclasses import dataclass, field 

13from enum import Enum 

14from typing import TYPE_CHECKING, Any 

15 

16from cachetools import TTLCache 

17 

18from .const import ( 

19 ATTR_DUPE_POLICY_MT, 

20 ATTR_DUPE_POLICY_MTSLP, 

21 ATTR_DUPE_POLICY_NONE, 

22 CONF_DUPE_POLICY, 

23 CONF_SIZE, 

24 CONF_TTL, 

25 PRIORITY_VALUES, 

26) 

27 

28if TYPE_CHECKING: 

29 from homeassistant.helpers.typing import ConfigType 

30 

31_LOGGER = logging.getLogger(__name__) 

32 

33_FALSY_STRINGS = frozenset({"false", "0", "no", "off", ""}) 

34 

35 

36def safe_get(probably_a_dict: dict[Any, Any] | None, key: Any, default: Any = None) -> Any: 

37 probably_a_dict = probably_a_dict or {} 

38 return probably_a_dict.get(key, default) 

39 

40 

41def safe_extend(target: list[Any], extension: list[Any] | tuple[Any] | Any) -> list[Any]: 

42 if target is None: 

43 target = [] 

44 elif not isinstance(target, list): 

45 target = [target] 

46 if isinstance(extension, list | tuple): 

47 target.extend(extension) 

48 elif extension: 

49 target.append(extension) 

50 return target 

51 

52 

53def nullable_ensure_list(v: Any) -> list[Any] | None: 

54 if v is None: 

55 return None 

56 return ensure_list(v) 

57 

58 

59def ensure_list(v: Any) -> list[Any]: 

60 if v is None: 

61 return [] 

62 if isinstance(v, list): 

63 return v 

64 if isinstance(v, tuple): 

65 return list(v) 

66 return [v] 

67 

68 

69def ensure_dict(v: Any, default: Any = None) -> dict[Any, Any]: 

70 if v is None: 

71 return {} 

72 if isinstance(v, dict): 

73 return v 

74 if isinstance(v, set | list): 

75 return dict.fromkeys(v, default) 

76 return {v: default} 

77 

78 

79def sanitize(v: Any, minimal: bool = True, top_level_keys_only: bool = False, **kwargs) -> Any: 

80 if isinstance(v, dt.datetime | dt.time | dt.date): 

81 return v.isoformat() 

82 if isinstance(v, str | int | float | bool): 

83 return v 

84 if isinstance(v, list | KeysView): 

85 return [sanitize(vv, minimal=minimal, **kwargs) for vv in v] 

86 if isinstance(v, tuple): 

87 return (sanitize(vv, minimal=minimal, **kwargs) for vv in v) 

88 if isinstance(v, dict): 

89 if top_level_keys_only: 

90 return [sanitize(k, minimal, **kwargs) for k in v] 

91 return {k: sanitize(vv, minimal=minimal, **kwargs) for k, vv in v.items()} 

92 if isinstance(v, Enum): 

93 return str(v) 

94 if isinstance(v, object): 

95 if hasattr(v, "contents"): 

96 return sanitize(v.contents(**kwargs), minimal=minimal, **kwargs) 

97 if hasattr(v, "as_dict"): 

98 return v.as_dict(**kwargs) 

99 return None 

100 

101 

102@dataclass 

103class CallRecord: 

104 timestamp: dt.datetime | None = field() 

105 elapsed: float = field() 

106 domain: str | None = field(default=None) 

107 action: str | None = field(default=None) 

108 action_data: dict[str, Any] | None = field(default=None) 

109 target_data: dict[str, Any] | None = field(default=None) 

110 exception: str | None = field(default=None) 

111 debug: bool = field(default=False) 

112 service_response: dict[str, Any] | None = field(default=None) 

113 

114 def contents(self, **_kwargs: Any) -> dict[str, Any]: 

115 result = { 

116 "timestamp": self.timestamp.isoformat() if self.timestamp else None, 

117 "domain": self.domain, 

118 "action": self.action, 

119 "action_data": self.action_data, 

120 "elapsed": self.elapsed, 

121 "debug": self.debug, 

122 } 

123 if self.target_data is not None: 

124 result["target_data"] = self.target_data 

125 if self.exception is not None: 

126 result["exception"] = self.exception 

127 if self.service_response is not None: 

128 result["service_response"] = self.service_response 

129 return result 

130 

131 

132class DupeCheckable: 

133 id: str 

134 priority: str 

135 

136 @abstractmethod 

137 def hash(self) -> int: 

138 raise NotImplementedError 

139 

140 

141class DupeChecker: 

142 """Apply the duplicate checking policy to duplicate candidates""" 

143 

144 def __init__(self, dupe_check_config: ConfigType) -> None: 

145 self.policy = dupe_check_config.get(CONF_DUPE_POLICY, ATTR_DUPE_POLICY_MTSLP) 

146 # dupe check cache, key is (priority, message hash) 

147 self.cache: TTLCache[tuple[int, int], str] = TTLCache( 

148 maxsize=dupe_check_config.get(CONF_SIZE, 100), ttl=dupe_check_config.get(CONF_TTL, 120) 

149 ) 

150 

151 def check(self, dupe_candidate: DupeCheckable) -> bool: 

152 if self.policy == ATTR_DUPE_POLICY_NONE: 

153 return False 

154 hashed: int = dupe_candidate.hash() 

155 ranked_priority: int = PRIORITY_VALUES.get(dupe_candidate.priority, 3) 

156 if self.policy == ATTR_DUPE_POLICY_MTSLP: 

157 dupe: bool = any(prev_hash == hashed and prev_prior >= ranked_priority for prev_hash, prev_prior in self.cache) 

158 elif self.policy == ATTR_DUPE_POLICY_MT: 

159 dupe = any(prev_hash == hashed for prev_hash, _prev_prior in self.cache) 

160 else: 

161 dupe = False 

162 if dupe: 

163 _LOGGER.debug("SUPERNOTIFY Detected dupe: %s", dupe_candidate.id) 

164 self.cache[hashed, ranked_priority] = dupe_candidate.id 

165 return dupe 

166 

167 

168def boolify(value: Any, default: bool) -> bool: 

169 """Convert a value to bool, correctly handling string 'false'/'true'. 

170 

171 Python's built-in bool() treats any non-empty string as True, so 

172 bool("false") == True. This helper avoids that pitfall for values 

173 that may arrive as YAML strings or Jinja2 template results. 

174 """ 

175 if value is None: 

176 return default 

177 if isinstance(value, bool): 

178 return value 

179 if isinstance(value, str): 

180 return value.strip().lower() not in _FALSY_STRINGS 

181 return bool(value)