Coverage for custom_components/supernotify/scenario.py: 93%
254 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +0000
1from __future__ import annotations
3import logging
4import re
5from typing import TYPE_CHECKING, Any
7from homeassistant.const import CONF_ENABLED, STATE_OFF, STATE_ON, STATE_UNKNOWN
8from homeassistant.core import (
9 HomeAssistant,
10 callback,
11)
12from homeassistant.helpers import condition
13from homeassistant.helpers import issue_registry as ir
15from custom_components.supernotify.people import PeopleRegistry
17from .const import (
18 ATTR_MEDIA,
19 CONF_EXPOSE_STATE,
20 CONF_REFRESH,
21 CONF_REFRESH_INTERVAL,
22 PRIORITY_MEDIUM,
23 SCENARIO_STATE_REFRESH_DEFAULT,
24)
25from .model import DeliveryCustomization
27if TYPE_CHECKING:
28 from collections.abc import Iterator
30 from homeassistant.helpers.typing import ConfigType
32 from .binary_sensor import SupernotifyScenarioBinarySensor
33 from .delivery import Delivery, DeliveryRegistry
34 from .hass_api import HomeAssistantAPI
35 from .schema import ConditionsFunc
37from contextlib import contextmanager
39import voluptuous as vol
41# type: ignore[attr-defined,unused-ignore]
42from homeassistant.components.trace import async_store_trace
43from homeassistant.components.trace.models import ActionTrace
44from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_NAME, CONF_ALIAS, CONF_CONDITIONS
45from homeassistant.core import Context
47from .const import ATTR_ENABLED, CONF_ACTION_GROUP_NAMES, CONF_DELIVERY, CONF_MEDIA
48from .model import ConditionVariables
50_LOGGER = logging.getLogger(__name__)
53class ScenarioRegistry:
54 def __init__(
55 self, scenario_configs: ConfigType, scenario_control: ConfigType | None, people_registry: PeopleRegistry
56 ) -> None:
57 self._config: ConfigType = scenario_configs or {}
58 self.scenarios: dict[str, Scenario] = {}
59 self.scenario_control = scenario_control or {}
60 self._people_registry: PeopleRegistry = people_registry
61 # Populated by binary_sensor.py's async_setup_entry once the platform is loaded (after
62 # initialize() below) - see register_entity/unregister_entity. Empty (and harmless to
63 # look up against) before then, e.g. in tests that build ScenarioRegistry directly
64 # without a config entry.
65 self._entities: dict[str, SupernotifyScenarioBinarySensor] = {}
66 # Shared occupancy/ConditionVariables snapshot for the scenario currently being batch
67 # refreshed - set for the duration of async_refresh_scenario_states()'s loop, read by
68 # scenario_is_on() so determine_occupancy() runs once per refresh instead of once per
69 # scenario. None outside of a batch refresh (each scenario_is_on() call then computes
70 # its own, e.g. a single entity being read on demand).
71 self._batch_cvars: ConditionVariables | None = None
73 async def initialize(
74 self,
75 delivery_registry: DeliveryRegistry,
76 mobile_actions: ConfigType,
77 hass_api: HomeAssistantAPI,
78 ) -> None:
80 for scenario_name, scenario_definition in self._config.items():
81 scenario = Scenario(scenario_name, scenario_definition, delivery_registry, hass_api)
82 if await scenario.validate(valid_action_group_names=list(mobile_actions)):
83 self.scenarios[scenario_name] = scenario
84 else:
85 _LOGGER.warning("SUPERNOTIFY Scenario %s failed to validate, ignoring", scenario.name)
86 self._hass_api: HomeAssistantAPI = hass_api
87 self._scenario_cond_entities = self._collect_scenario_condition_entities()
88 self._scenario_by_entity = self._index_scenarios_by_entity()
90 # Keep the scenario binary_sensors' state current: react to their condition entities
91 # (immediate, and only for the scenarios that depend on the entity that changed), plus
92 # a periodic sweep for conditions no entity change announces - time windows, sun, and
93 # templates whose dependencies could not be extracted.
94 # Evaluating conditions costs whatever the conditions cost, so the whole mechanism is
95 # switchable: `scenario_control: {enabled: false}` subscribes to nothing and starts no
96 # timer, and `refresh_interval: 0` keeps the reactive path without the sweep.
97 if self.scenario_state_enabled:
98 scenario_watch: set[str] = set(self._scenario_by_entity)
99 if scenario_watch:
100 hass_api.subscribe_state(sorted(scenario_watch), self.async_refresh_scenario_states)
101 if self.scenario_state_interval:
102 hass_api.subscribe_interval(self.scenario_state_interval, self.async_refresh_scenario_states)
104 def register_entity(self, name: str, entity: SupernotifyScenarioBinarySensor) -> None:
105 """Called by SupernotifyScenarioBinarySensor.async_added_to_hass()."""
106 self._entities[name] = entity
108 def unregister_entity(self, name: str) -> None:
109 """Called by SupernotifyScenarioBinarySensor.async_will_remove_from_hass()."""
110 self._entities.pop(name, None)
112 @callback
113 def async_refresh_entity(self, name: str) -> None:
114 """Re-publish one scenario's binary_sensor now, whether or not periodic refresh is on"""
115 entity = self._entities.get(name)
116 if entity is not None:
117 entity.async_write_ha_state()
119 def scenario_has_state(self, scenario: Scenario) -> bool:
120 """Whether a scenario has any state to report - anything that hasn't opted out with
121 expose_state. That is the evaluated state of its conditions if it has any, otherwise
122 a manual state that something outside Supernotify sets, see Scenario.manual_active."""
123 return scenario.expose_state
125 def scenario_is_on(self, scenario: Scenario) -> bool | None:
126 """`is_on` for SupernotifyScenarioBinarySensor - None maps to STATE_UNKNOWN."""
127 if scenario.is_manual:
128 return scenario.manual_active
129 state = self._scenario_state(scenario, self._batch_cvars)
130 if state == STATE_UNKNOWN:
131 return None
132 return state == STATE_ON
134 def _collect_scenario_condition_entities(self) -> dict[str, set[str]]:
135 """Entities referenced by each scenario's conditions.
137 A scenario whose conditions reference no Home Assistant entity depends
138 only on the per-notification variables (notification_priority /
139 applied_scenarios). Such a scenario is 'transient': it has no meaningful
140 state between notifications, so it is left as STATE_UNKNOWN. Extraction is
141 best-effort (templates are opaque); the periodic refresh is the safety net.
142 """
143 mapping: dict[str, set[str]] = {}
144 for name, scenario in self.scenarios.items():
145 ents: set[str] = set()
146 for cond in scenario.conditions_config or []:
147 try:
148 ents |= condition.async_extract_entities(cond)
149 except Exception:
150 _LOGGER.debug("SUPERNOTIFY could not extract entities for scenario %s", name)
151 mapping[name] = ents
152 return mapping
154 @property
155 def scenario_state_enabled(self) -> bool:
156 return bool(self.scenario_control.get(CONF_REFRESH, True))
158 @property
159 def scenario_state_interval(self) -> int:
160 return int(self.scenario_control.get(CONF_REFRESH_INTERVAL, SCENARIO_STATE_REFRESH_DEFAULT))
162 def _index_scenarios_by_entity(self) -> dict[str, set[str]]:
163 """Reverse of _collect_scenario_condition_entities: entity -> scenarios depending on it.
165 Used to re-evaluate only the scenarios a state change can actually affect, instead of
166 the whole registry on every event.
167 """
168 index: dict[str, set[str]] = {}
169 for name, entities in self._scenario_cond_entities.items():
170 scenario = self.scenarios.get(name)
171 if scenario is not None and not scenario.expose_state:
172 continue
173 for entity_id in entities:
174 index.setdefault(entity_id, set()).add(name)
175 return index
177 def _scenario_state(self, scenario: Scenario, cvars: ConditionVariables | None = None) -> str:
178 """State to expose for a scenario binary_sensor.
180 - no conditions at all (manual/emergency-only scenario) -> STATE_UNKNOWN
181 (state is undefined outside of a notification);
182 - otherwise ON/OFF from a neutral evaluation (current occupancy, medium
183 priority), the same basis as enquire_active_scenarios() - including a
184 scenario whose conditions reference no HA entity (e.g. a pure now()/date
185 template): those are exactly what the periodic sweep in initialize() exists
186 to keep current, so they get evaluated for real rather than stuck at UNKNOWN.
187 A condition keyed off notification_priority/applied_scenarios/message/title
188 instead will always evaluate the same way here, since those are neutral
189 placeholders rather than real values - a known, accepted limitation rather
190 than something this method tries to detect and suppress.
191 """
192 if not scenario.expose_state:
193 return STATE_UNKNOWN
194 if not scenario.conditions_config:
195 return STATE_UNKNOWN
196 if cvars is None:
197 occupiers = self._people_registry.determine_occupancy()
198 cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
199 return STATE_ON if scenario.evaluate(cvars) else STATE_OFF
201 @callback
202 def async_refresh_scenario_states(self, *args: Any) -> None:
203 """Ask each affected scenario's binary_sensor entity to re-read and re-publish its state.
205 Triggered by the 1-minute timer (time/date scenarios and any dependency
206 not captured by entity extraction) and by state changes of the scenarios'
207 condition entities (immediate reactivity). The entity's own `is_on`
208 property (via scenario_is_on() above) does the actual (pure, in-memory)
209 evaluation on read; this only decides which entities need to refresh, and
210 is a no-op for a scenario with no entity registered yet (e.g. before the
211 binary_sensor platform has finished loading).
212 """
213 if not self.scenario_state_enabled:
214 return
215 names: set[str] | None = None
216 if args:
217 event = args[0]
218 entity_id = getattr(event, "data", {}).get("entity_id") if hasattr(event, "data") else None
219 if entity_id is not None:
220 names = self._scenario_by_entity.get(entity_id, set())
221 if not names:
222 return
224 # Computed once for the whole batch (see _batch_cvars/scenario_is_on) rather than once
225 # per scenario below - determine_occupancy() is only dict lookups, not real I/O, but
226 # doing it once per refresh instead of once per scenario is the correct scale for a
227 # mechanism meant to run on every relevant state change and every periodic sweep.
228 occupiers = self._people_registry.determine_occupancy()
229 self._batch_cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
230 try:
231 for name in self.scenarios if names is None else names:
232 entity = self._entities.get(name)
233 if entity is not None:
234 entity.async_write_ha_state()
235 finally:
236 self._batch_cvars = None
239class Scenario:
240 def __init__(
241 self, name: str, scenario_definition: dict[str, Any], delivery_registry: DeliveryRegistry, hass_api: HomeAssistantAPI
242 ) -> None:
243 self.hass_api: HomeAssistantAPI = hass_api
244 self.delivery_registry = delivery_registry
245 self.enabled: bool = scenario_definition.get(CONF_ENABLED, True)
246 # as configured, which enabled can be overridden from at runtime by the scenario switch
247 self.config_enabled: bool = self.enabled
248 self.expose_state: bool = scenario_definition.get(CONF_EXPOSE_STATE, True)
249 self.name: str = name
250 self.alias: str | None = scenario_definition.get(CONF_ALIAS)
251 self.conditions: ConditionsFunc | None = None
252 self.conditions_config: list[ConfigType] | None = scenario_definition.get(CONF_CONDITIONS)
253 # With no conditions to evaluate, whether the scenario applies is set from outside, by
254 # the state of its binary_sensor - see SupernotifyScenarioManualBinarySensor
255 self.manual_active: bool = False
256 self.media: dict[str, Any] | None = scenario_definition.get(CONF_MEDIA)
257 self.action_groups: list[str] = scenario_definition.get(CONF_ACTION_GROUP_NAMES, [])
258 self._config_delivery: dict[str, DeliveryCustomization]
259 self.delivery_overrides: dict[str, DeliveryCustomization] = {}
260 self._delivery_selector: dict[str, str] = {}
261 self.last_trace: ActionTrace | None = None
262 self.startup_issue_count: int = 0
264 delivery_data = scenario_definition.get(CONF_DELIVERY)
265 if isinstance(delivery_data, list):
266 # a bare list of deliveries implies enabling
267 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery default enabled for list %s", self.name, delivery_data)
268 self._config_delivery = {k: DeliveryCustomization(config=None, default_enabled=True) for k in delivery_data}
269 elif isinstance(delivery_data, str) and delivery_data:
270 # a bare list of deliveries implies enabled delivery
271 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery default enabled for single %s", self.name, delivery_data)
272 self._config_delivery = {delivery_data: DeliveryCustomization(config=None, default_enabled=True)}
273 elif isinstance(delivery_data, dict):
274 # whereas a dict may be used to tune or restrict
275 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery selection %s", self.name, delivery_data)
276 self._config_delivery = {}
277 for k, v in delivery_data.items():
278 # a wildcard/regex pattern with no explicit enabled: only apply as an
279 # override to deliveries already selected elsewhere, don't force-enable
280 # every delivery it happens to match (e.g. selection: scenario deliveries)
281 self._config_delivery[k] = DeliveryCustomization(
282 config=v, default_enabled=True if k in delivery_registry.deliveries else None
283 )
284 elif delivery_data:
285 _LOGGER.warning("SUPERNOTIFY Unable to interpret scenario %s delivery data %s", self.name, delivery_data)
286 self._config_delivery = {}
287 else:
288 _LOGGER.warning("SUPERNOTIFY No delivery definitions for scenario %s", self.name)
289 self._config_delivery = {}
291 @property
292 def is_manual(self) -> bool:
293 """A scenario with no conditions, which only applies when its manual state is on"""
294 return not self.conditions_config
296 async def validate(self, valid_action_group_names: list[str] | None = None) -> bool:
297 """Validate Home Assistant conditiion definition at initiation"""
298 if self.conditions_config:
299 error: str | None = None
300 try:
301 # note: basic template syntax within conditions already validated by voluptuous checks
302 self.conditions = await self.hass_api.build_conditions(self.conditions_config, strict=True, validate=True)
303 except vol.Invalid as vi:
304 _LOGGER.error(
305 f"SUPERNOTIFY Condition definition for scenario {self.name} fails Home Assistant schema check {vi}"
306 )
307 error = f"Schema error {vi}"
308 except Exception as e:
309 _LOGGER.error(
310 "SUPERNOTIFY Disabling scenario %s with error validating %s: %s", self.name, self.conditions_config, e
311 )
312 error = f"Unknown error {e}"
313 if error is not None:
314 self.startup_issue_count += 1
315 self.hass_api.raise_issue(
316 f"scenario_{self.name}_condition",
317 is_fixable=False,
318 issue_key="scenario_condition",
319 issue_map={"scenario": self.name, "error": error},
320 severity=ir.IssueSeverity.ERROR,
321 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/scenarios/",
322 )
324 for name_or_pattern, config in self._config_delivery.items():
325 matched: bool = False
326 delivery: Delivery | None = self.delivery_registry.deliveries.get(name_or_pattern)
327 if delivery:
328 self.delivery_overrides[delivery.name] = config
329 self._delivery_selector[delivery.name] = name_or_pattern
330 matched = True
331 else:
332 # look for a wildcard match instead
333 for delivery_name in self.delivery_registry.deliveries:
334 if re.fullmatch(name_or_pattern, delivery_name):
335 if self._delivery_selector.get(delivery_name) == delivery_name:
336 _LOGGER.info(
337 f"SUPERNOTIFY Scenario {self.name} ignoring '{name_or_pattern}' shadowing explicit delivery {delivery_name}"
338 )
339 else:
340 _LOGGER.debug(
341 f"SUPERNOTIFY Scenario {self.name} delivery '{name_or_pattern}' matched {delivery_name}"
342 )
343 self.delivery_overrides[delivery_name] = config
344 self._delivery_selector[delivery_name] = name_or_pattern
345 matched = True
346 if not matched:
347 _LOGGER.error(f"SUPERNOTIFY Scenario {self.name} has delivery {name_or_pattern} not found")
348 self.startup_issue_count += 1
349 self.hass_api.raise_issue(
350 f"scenario_{self.name}_delivery_{name_or_pattern.replace('.', 'DOT').replace('*', 'STAR')}",
351 is_fixable=False,
352 issue_key="scenario_delivery",
353 issue_map={
354 "scenario": self.name,
355 "delivery": name_or_pattern,
356 "deliveries": ", ".join(self.delivery_registry.deliveries),
357 },
358 severity=ir.IssueSeverity.WARNING,
359 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/scenarios/",
360 )
362 if valid_action_group_names is not None:
363 invalid_action_groups: list[str] = []
364 for action_group_name in self.action_groups:
365 if action_group_name not in valid_action_group_names:
366 _LOGGER.error(f"SUPERNOTIFY Unknown action group {action_group_name} removed from scenario {self.name}")
367 invalid_action_groups.append(action_group_name)
368 self.startup_issue_count += 1
369 self.hass_api.raise_issue(
370 f"scenario_{self.name}_action_group_{action_group_name}",
371 is_fixable=False,
372 issue_key="scenario_action_group",
373 issue_map={
374 "scenario": self.name,
375 "action_group": action_group_name,
376 "action_groups": ", ".join(valid_action_group_names),
377 },
378 severity=ir.IssueSeverity.WARNING,
379 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/scenarios/",
380 )
381 for action_group_name in invalid_action_groups:
382 self.action_groups.remove(action_group_name)
384 return self.startup_issue_count == 0
386 def enabling_deliveries(self) -> list[str]:
387 # default_enabled (see __init__/DeliveryCustomization) already resolves whether an
388 # omitted `enabled` key should count as enabling: True for a directly-named delivery
389 # (matching the list/string delivery config forms), None for a wildcard/regex match
390 # so it doesn't force-select every delivery it happens to match. An explicit
391 # `enabled: None` (e.g. just to carry a priority override) is left as None too, not
392 # upgraded to the default - only a real `enabled: true` should land here.
393 return [del_name for del_name, del_config in self.delivery_overrides.items() if del_config.enabled is True]
395 def relevant_deliveries(self) -> list[str]:
396 return [
397 del_name
398 for del_name, del_config in self.delivery_overrides.items()
399 if del_config.enabled or del_config.enabled is None
400 ]
402 def disabling_deliveries(self) -> list[str]:
403 return [del_name for del_name, del_config in self.delivery_overrides.items() if del_config.enabled is False]
405 def delivery_customization(self, delivery_name: str) -> DeliveryCustomization | None:
406 return self.delivery_overrides.get(delivery_name)
408 def attributes(self, include_condition: bool = True, include_trace: bool = False) -> dict[str, Any]:
409 """Return scenario attributes"""
410 attrs = {
411 ATTR_NAME: self.name,
412 ATTR_ENABLED: self.enabled,
413 ATTR_MEDIA: self.media,
414 "action_groups": self.action_groups,
415 "delivery": self.delivery_overrides,
416 }
417 if self.alias:
418 attrs[ATTR_FRIENDLY_NAME] = self.alias
419 if include_condition:
420 attrs["conditions"] = self.conditions_config
421 if include_trace and self.last_trace:
422 attrs["trace"] = self.last_trace.as_extended_dict()
423 return attrs
425 def delivery_config(self, delivery_name: str) -> DeliveryCustomization | None:
426 return self.delivery_overrides.get(delivery_name)
428 def contents(self, minimal: bool = False, **_kwargs: Any) -> dict[str, Any]:
429 """Archive friendly view of scenario"""
430 return self.attributes(include_condition=False, include_trace=not minimal)
432 def evaluate(self, condition_variables: ConditionVariables) -> bool:
433 """Evaluate scenario conditions"""
434 result: bool | None = False
435 if self.enabled and self.is_manual:
436 return self.manual_active
437 if self.enabled and self.conditions:
438 try:
439 result = self.hass_api.evaluate_conditions(self.conditions, condition_variables)
440 if result is None:
441 _LOGGER.warning(f"SUPERNOTIFY Scenario {self.name} condition empty result")
442 except Exception as e:
443 _LOGGER.error(
444 "SUPERNOTIFY Scenario %s condition eval failed: %s, vars: %s",
445 self.name,
446 e,
447 condition_variables.as_dict() if condition_variables else {},
448 )
449 return result if result is not None else False
451 async def trace(self, condition_variables: ConditionVariables) -> bool:
452 """Trace scenario condition execution"""
453 result: bool | None = False
454 trace: ActionTrace | None = None
455 if self.enabled and self.is_manual:
456 return self.manual_active
457 if self.enabled and self.conditions:
458 result, trace = await self.hass_api.trace_conditions(
459 self.conditions, condition_variables, trace_name=f"scenario_{self.name}"
460 )
461 if trace:
462 self.last_trace = trace
463 return result if result is not None else False
466@contextmanager
467def trace_action(
468 hass: HomeAssistant,
469 item_id: str,
470 config: dict[str, Any],
471 context: Context | None = None,
472 stored_traces: int = 5,
473) -> Iterator[ActionTrace]:
474 """Trace execution of a scenario."""
475 trace = ActionTrace(item_id, config, None, context or Context())
476 async_store_trace(hass, trace, stored_traces)
478 try:
479 yield trace
480 except Exception as ex:
481 if item_id:
482 trace.set_error(ex)
483 raise
484 finally:
485 if item_id:
486 trace.finished()