Coverage for custom_components/supernotify/common.py: 97%
150 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1"""Miscellaneous helper functions.
3No same pkg dependencies permitted
4"""
6from __future__ import annotations
8import datetime as dt
9import logging
10import time
11from abc import abstractmethod
12from collections.abc import Callable, KeysView
13from dataclasses import dataclass, field
14from enum import Enum
15from typing import TYPE_CHECKING, Any
17from .const import (
18 ATTR_DUPE_POLICY_MT,
19 ATTR_DUPE_POLICY_MTSLP,
20 ATTR_DUPE_POLICY_NONE,
21 CONF_DUPE_POLICY,
22 CONF_SIZE,
23 CONF_TTL,
24 PRIORITY_SYNONYMS,
25 PRIORITY_VALUES,
26)
28if TYPE_CHECKING:
29 from homeassistant.helpers.typing import ConfigType
31_LOGGER = logging.getLogger(__name__)
33_FALSY_STRINGS = frozenset({"false", "0", "no", "off", ""})
36def safe_get(probably_a_dict: dict[Any, Any] | None, key: Any, default: Any = None) -> Any: # ruff: ignore[any-type]
37 probably_a_dict = probably_a_dict or {}
38 return probably_a_dict.get(key, default)
41def safe_extend(target: list[Any], extension: list[Any] | tuple[Any] | Any) -> list[Any]: # ruff: ignore[any-type]
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
53def nullable_ensure_list(v: Any) -> list[Any] | None: # ruff: ignore[any-type]
54 if v is None:
55 return None
56 return ensure_list(v)
59def canonical_priority(priority: Any) -> Any: # ruff: ignore[any-type]
60 """Supernotify's own name for a priority given by a synonym, such as `highest` for `critical`"""
61 return PRIORITY_SYNONYMS.get(priority, priority) if isinstance(priority, str) else priority
64def spoken_name(name: str) -> str:
65 """A name as it would be said, ignoring case, underscores and extra spaces, for matching spoken tags"""
66 return " ".join(name.replace("_", " ").casefold().split())
69def ensure_list(v: Any) -> list[Any]: # ruff: ignore[any-type]
70 if v is None:
71 return []
72 if isinstance(v, list):
73 return v
74 if isinstance(v, tuple):
75 return list(v)
76 return [v]
79def ensure_dict(v: Any, default: Any = None) -> dict[Any, Any]: # ruff: ignore[any-type]
80 if v is None:
81 return {}
82 if isinstance(v, dict):
83 return v
84 if isinstance(v, set | list):
85 return dict.fromkeys(v, default)
86 return {v: default}
89def sanitize(v: Any, minimal: bool = True, top_level_keys_only: bool = False, **kwargs: Any) -> Any: # ruff: ignore[any-type]
90 if isinstance(v, dt.datetime | dt.time | dt.date):
91 return v.isoformat()
92 if isinstance(v, str | int | float | bool):
93 return v
94 if isinstance(v, list | KeysView):
95 return [sanitize(vv, minimal=minimal, **kwargs) for vv in v]
96 if isinstance(v, tuple):
97 return (sanitize(vv, minimal=minimal, **kwargs) for vv in v)
98 if isinstance(v, dict):
99 if top_level_keys_only:
100 return [sanitize(k, minimal, **kwargs) for k in v]
101 return {k: sanitize(vv, minimal=minimal, **kwargs) for k, vv in v.items()}
102 if isinstance(v, Enum):
103 return str(v)
104 if isinstance(v, object):
105 if hasattr(v, "contents"):
106 return sanitize(v.contents(**kwargs), minimal=minimal, **kwargs)
107 if hasattr(v, "as_dict"):
108 return v.as_dict(**kwargs)
109 return None
112@dataclass
113class CallRecord:
114 timestamp: dt.datetime | None = field()
115 elapsed: float = field()
116 domain: str | None = field(default=None)
117 action: str | None = field(default=None)
118 action_data: dict[str, Any] | None = field(default=None)
119 target_data: dict[str, Any] | None = field(default=None)
120 exception: str | None = field(default=None)
121 debug: bool = field(default=False)
122 service_response: dict[str, Any] | None = field(default=None)
124 def contents(self, **_kwargs: Any) -> dict[str, Any]:
125 result = {
126 "timestamp": self.timestamp.isoformat() if self.timestamp else None,
127 "domain": self.domain,
128 "action": self.action,
129 "action_data": self.action_data,
130 "elapsed": self.elapsed,
131 "debug": self.debug,
132 }
133 if self.target_data is not None:
134 result["target_data"] = self.target_data
135 if self.exception is not None:
136 result["exception"] = self.exception
137 if self.service_response is not None:
138 result["service_response"] = self.service_response
139 return result
142class DupeCheckable:
143 id: str
144 priority: str
146 @abstractmethod
147 def hash(self) -> int:
148 raise NotImplementedError
151class DupeChecker:
152 """Apply the duplicate checking policy to duplicate candidates"""
154 def __init__(self, dupe_check_config: ConfigType, timer: Callable[[], float] = time.monotonic) -> None:
155 self.policy = dupe_check_config.get(CONF_DUPE_POLICY, ATTR_DUPE_POLICY_MTSLP)
156 self.size: int = dupe_check_config.get(CONF_SIZE, 100)
157 self.ttl: int = dupe_check_config.get(CONF_TTL, 120)
158 self.timer = timer
159 # when each live send expires, key is (message hash, priority), oldest first
160 self.cache: dict[tuple[int, int], float] = {}
161 # separate cache for dry runs, so a dry run never suppresses the live send that follows it
162 self.dry_run_cache: dict[tuple[int, int], float] = {}
164 def check(self, dupe_candidate: DupeCheckable, dry_run: bool = False) -> bool:
165 """Return True if the candidate is a dupe of one recorded within the TTL.
167 A live send checks only the live cache. A dry run checks both caches, so it
168 reports a dupe a live send would suppress.
169 """
170 if self.policy == ATTR_DUPE_POLICY_NONE:
171 return False
172 hashed: int = dupe_candidate.hash()
173 ranked_priority: int = PRIORITY_VALUES.get(dupe_candidate.priority, 3)
174 caches = (self.cache, self.dry_run_cache) if dry_run else (self.cache,)
175 dupe: bool = any(self._matches(cache, hashed, ranked_priority) for cache in caches)
176 if dupe:
177 _LOGGER.debug("SUPERNOTIFY Detected dupe: %s", dupe_candidate.id)
178 return dupe
180 def record(self, dupe_candidate: DupeCheckable, dry_run: bool = False) -> None:
181 """Remember a candidate that was sent, so a repeat within the TTL is a dupe.
183 Only what was delivered is recorded, so a failed send can be retried, and a
184 suppressed dupe doesn't keep resetting the TTL.
185 """
186 if self.policy == ATTR_DUPE_POLICY_NONE:
187 return
188 cache = self.dry_run_cache if dry_run else self.cache
189 self._expire(cache)
190 key = (dupe_candidate.hash(), PRIORITY_VALUES.get(dupe_candidate.priority, 3))
191 cache.pop(key, None)
192 while len(cache) >= self.size:
193 del cache[next(iter(cache))]
194 cache[key] = self.timer() + self.ttl
196 def _expire(self, cache: dict[tuple[int, int], float]) -> None:
197 now: float = self.timer()
198 for key in [k for k, expires in cache.items() if expires <= now]:
199 del cache[key]
201 def _matches(self, cache: dict[tuple[int, int], float], hashed: int, ranked_priority: int) -> bool:
202 self._expire(cache)
203 if self.policy == ATTR_DUPE_POLICY_MTSLP:
204 return any(prev_hash == hashed and prev_prior >= ranked_priority for prev_hash, prev_prior in cache)
205 if self.policy == ATTR_DUPE_POLICY_MT:
206 return any(prev_hash == hashed for prev_hash, _prev_prior in cache)
207 return False
210def boolify(value: Any, default: bool) -> bool: # ruff: ignore[any-type]
211 """Convert a value to bool, correctly handling string 'false'/'true'.
213 Python's built-in bool() treats any non-empty string as True, so
214 bool("false") == True. This helper avoids that pitfall for values
215 that may arrive as YAML strings or Jinja2 template results.
216 """
217 if value is None:
218 return default
219 if isinstance(value, bool):
220 return value
221 if isinstance(value, str):
222 return value.strip().lower() not in _FALSY_STRINGS
223 return bool(value)
226def int_or_none(v: Any) -> int | None: # ruff: ignore[any-type]
227 if v is not None:
228 try:
229 return int(v)
230 except ValueError:
231 _LOGGER.debug("SUPERNOTIFY Invalid int value %s", v)
232 return None