Coverage for custom_components/supernotify/scenario.py: 90%
155 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
1from __future__ import annotations
3import logging
4import re
5from typing import TYPE_CHECKING, Any
7from homeassistant.const import CONF_ENABLED
8from homeassistant.helpers import issue_registry as ir
10from .const import ATTR_MEDIA
11from .model import DeliveryCustomization
13if TYPE_CHECKING:
14 from collections.abc import Iterator
16 from homeassistant.core import HomeAssistant
17 from homeassistant.helpers.typing import ConfigType
19 from .delivery import Delivery, DeliveryRegistry
20 from .hass_api import HomeAssistantAPI
21 from .schema import ConditionsFunc
23from contextlib import contextmanager
25import voluptuous as vol
27# type: ignore[attr-defined,unused-ignore]
28from homeassistant.components.trace import async_store_trace
29from homeassistant.components.trace.models import ActionTrace
30from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_NAME, CONF_ALIAS, CONF_CONDITIONS
31from homeassistant.core import Context, HomeAssistant
33from .const import ATTR_ENABLED, CONF_ACTION_GROUP_NAMES, CONF_DELIVERY, CONF_MEDIA
34from .model import ConditionVariables
36_LOGGER = logging.getLogger(__name__)
39class ScenarioRegistry:
40 def __init__(self, scenario_configs: ConfigType) -> None:
41 self._config: ConfigType = scenario_configs or {}
42 self.scenarios: dict[str, Scenario] = {}
44 async def initialize(
45 self,
46 delivery_registry: DeliveryRegistry,
47 mobile_actions: ConfigType,
48 hass_api: HomeAssistantAPI,
49 ) -> None:
51 for scenario_name, scenario_definition in self._config.items():
52 scenario = Scenario(scenario_name, scenario_definition, delivery_registry, hass_api)
53 if await scenario.validate(valid_action_group_names=list(mobile_actions)):
54 self.scenarios[scenario_name] = scenario
55 else:
56 _LOGGER.warning("SUPERNOTIFY Scenario %s failed to validate, ignoring", scenario.name)
59class Scenario:
60 def __init__(
61 self, name: str, scenario_definition: dict[str, Any], delivery_registry: DeliveryRegistry, hass_api: HomeAssistantAPI
62 ) -> None:
63 self.hass_api: HomeAssistantAPI = hass_api
64 self.delivery_registry = delivery_registry
65 self.enabled: bool = scenario_definition.get(CONF_ENABLED, True)
66 self.name: str = name
67 self.alias: str | None = scenario_definition.get(CONF_ALIAS)
68 self.conditions: ConditionsFunc | None = None
69 self.conditions_config: list[ConfigType] | None = scenario_definition.get(CONF_CONDITIONS)
70 self.media: dict[str, Any] | None = scenario_definition.get(CONF_MEDIA)
71 self.action_groups: list[str] = scenario_definition.get(CONF_ACTION_GROUP_NAMES, [])
72 self._config_delivery: dict[str, DeliveryCustomization]
73 self.delivery_overrides: dict[str, DeliveryCustomization] = {}
74 self._delivery_selector: dict[str, str] = {}
75 self.last_trace: ActionTrace | None = None
76 self.startup_issue_count: int = 0
78 delivery_data = scenario_definition.get(CONF_DELIVERY)
79 if isinstance(delivery_data, list):
80 # a bare list of deliveries implies enabling
81 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery default enabled for list %s", self.name, delivery_data)
82 self._config_delivery = {k: DeliveryCustomization(config=None, default_enabled=True) for k in delivery_data}
83 elif isinstance(delivery_data, str) and delivery_data:
84 # a bare list of deliveries implies enabled delivery
85 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery default enabled for single %s", self.name, delivery_data)
86 self._config_delivery = {delivery_data: DeliveryCustomization(config=None, default_enabled=True)}
87 elif isinstance(delivery_data, dict):
88 # whereas a dict may be used to tune or restrict
89 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery selection %s", self.name, delivery_data)
90 self._config_delivery = {}
91 for k, v in delivery_data.items():
92 # a wildcard/regex pattern with no explicit enabled: only apply as an
93 # override to deliveries already selected elsewhere, don't force-enable
94 # every delivery it happens to match (e.g. selection: scenario deliveries)
95 self._config_delivery[k] = DeliveryCustomization(
96 config=v, default_enabled=True if k in delivery_registry.deliveries else None
97 )
98 elif delivery_data:
99 _LOGGER.warning("SUPERNOTIFY Unable to interpret scenario %s delivery data %s", self.name, delivery_data)
100 self._config_delivery = {}
101 else:
102 _LOGGER.warning("SUPERNOTIFY No delivery definitions for scenario %s", self.name)
103 self._config_delivery = {}
105 async def validate(self, valid_action_group_names: list[str] | None = None) -> bool:
106 """Validate Home Assistant conditiion definition at initiation"""
107 if self.conditions_config:
108 error: str | None = None
109 try:
110 # note: basic template syntax within conditions already validated by voluptuous checks
111 self.conditions = await self.hass_api.build_conditions(self.conditions_config, strict=True, validate=True)
112 except vol.Invalid as vi:
113 _LOGGER.error(
114 f"SUPERNOTIFY Condition definition for scenario {self.name} fails Home Assistant schema check {vi}"
115 )
116 error = f"Schema error {vi}"
117 except Exception as e:
118 _LOGGER.error(
119 "SUPERNOTIFY Disabling scenario %s with error validating %s: %s", self.name, self.conditions_config, e
120 )
121 error = f"Unknown error {e}"
122 if error is not None:
123 self.startup_issue_count += 1
124 self.hass_api.raise_issue(
125 f"scenario_{self.name}_condition",
126 is_fixable=False,
127 issue_key="scenario_condition",
128 issue_map={"scenario": self.name, "error": error},
129 severity=ir.IssueSeverity.ERROR,
130 learn_more_url="https://supernotify.rhizomatics.org.uk/scenarios/",
131 )
133 for name_or_pattern, config in self._config_delivery.items():
134 matched: bool = False
135 delivery: Delivery | None = self.delivery_registry.deliveries.get(name_or_pattern)
136 if delivery:
137 self.delivery_overrides[delivery.name] = config
138 self._delivery_selector[delivery.name] = name_or_pattern
139 matched = True
140 else:
141 # look for a wildcard match instead
142 for delivery_name in self.delivery_registry.deliveries:
143 if re.fullmatch(name_or_pattern, delivery_name):
144 if self._delivery_selector.get(delivery_name) == delivery_name:
145 _LOGGER.info(
146 f"SUPERNOTIFY Scenario {self.name} ignoring '{name_or_pattern}' shadowing explicit delivery {delivery_name}"
147 )
148 else:
149 _LOGGER.debug(
150 f"SUPERNOTIFY Scenario {self.name} delivery '{name_or_pattern}' matched {delivery_name}"
151 )
152 self.delivery_overrides[delivery_name] = config
153 self._delivery_selector[delivery_name] = name_or_pattern
154 matched = True
155 if not matched:
156 _LOGGER.error(f"SUPERNOTIFY Scenario {self.name} has delivery {name_or_pattern} not found")
157 self.startup_issue_count += 1
158 self.hass_api.raise_issue(
159 f"scenario_{self.name}_delivery_{name_or_pattern.replace('.', 'DOT').replace('*', 'STAR')}",
160 is_fixable=False,
161 issue_key="scenario_delivery",
162 issue_map={"scenario": self.name, "delivery": name_or_pattern},
163 severity=ir.IssueSeverity.WARNING,
164 learn_more_url="https://supernotify.rhizomatics.org.uk/scenarios/",
165 )
167 if valid_action_group_names is not None:
168 invalid_action_groups: list[str] = []
169 for action_group_name in self.action_groups:
170 if action_group_name not in valid_action_group_names:
171 _LOGGER.error(f"SUPERNOTIFY Unknown action group {action_group_name} removed from scenario {self.name}")
172 invalid_action_groups.append(action_group_name)
173 self.startup_issue_count += 1
174 self.hass_api.raise_issue(
175 f"scenario_{self.name}_action_group_{action_group_name}",
176 is_fixable=False,
177 issue_key="scenario_delivery",
178 issue_map={"scenario": self.name, "action_group": action_group_name},
179 severity=ir.IssueSeverity.WARNING,
180 learn_more_url="https://supernotify.rhizomatics.org.uk/scenarios/",
181 )
182 for action_group_name in invalid_action_groups:
183 self.action_groups.remove(action_group_name)
185 return self.startup_issue_count == 0
187 def enabling_deliveries(self) -> list[str]:
188 # default_enabled (see __init__/DeliveryCustomization) already resolves whether an
189 # omitted `enabled` key should count as enabling: True for a directly-named delivery
190 # (matching the list/string delivery config forms), None for a wildcard/regex match
191 # so it doesn't force-select every delivery it happens to match. An explicit
192 # `enabled: None` (e.g. just to carry a priority override) is left as None too, not
193 # upgraded to the default - only a real `enabled: true` should land here.
194 return [del_name for del_name, del_config in self.delivery_overrides.items() if del_config.enabled is True]
196 def relevant_deliveries(self) -> list[str]:
197 return [
198 del_name
199 for del_name, del_config in self.delivery_overrides.items()
200 if del_config.enabled or del_config.enabled is None
201 ]
203 def disabling_deliveries(self) -> list[str]:
204 return [del_name for del_name, del_config in self.delivery_overrides.items() if del_config.enabled is False]
206 def delivery_customization(self, delivery_name: str) -> DeliveryCustomization | None:
207 return self.delivery_overrides.get(delivery_name)
209 def attributes(self, include_condition: bool = True, include_trace: bool = False) -> dict[str, Any]:
210 """Return scenario attributes"""
211 attrs = {
212 ATTR_NAME: self.name,
213 ATTR_ENABLED: self.enabled,
214 ATTR_MEDIA: self.media,
215 "action_groups": self.action_groups,
216 "delivery": self.delivery_overrides,
217 }
218 if self.alias:
219 attrs[ATTR_FRIENDLY_NAME] = self.alias
220 if include_condition:
221 attrs["conditions"] = self.conditions_config
222 if include_trace and self.last_trace:
223 attrs["trace"] = self.last_trace.as_extended_dict()
224 return attrs
226 def delivery_config(self, delivery_name: str) -> DeliveryCustomization | None:
227 return self.delivery_overrides.get(delivery_name)
229 def contents(self, minimal: bool = False, **_kwargs: Any) -> dict[str, Any]:
230 """Archive friendly view of scenario"""
231 return self.attributes(include_condition=False, include_trace=not minimal)
233 def evaluate(self, condition_variables: ConditionVariables) -> bool:
234 """Evaluate scenario conditions"""
235 result: bool | None = False
236 if self.enabled and self.conditions:
237 try:
238 result = self.hass_api.evaluate_conditions(self.conditions, condition_variables)
239 if result is None:
240 _LOGGER.warning(f"SUPERNOTIFY Scenario {self.name} condition empty result")
241 except Exception as e:
242 _LOGGER.error(
243 "SUPERNOTIFY Scenario %s condition eval failed: %s, vars: %s",
244 self.name,
245 e,
246 condition_variables.as_dict() if condition_variables else {},
247 )
248 return result if result is not None else False
250 async def trace(self, condition_variables: ConditionVariables) -> bool:
251 """Trace scenario condition execution"""
252 result: bool | None = False
253 trace: ActionTrace | None = None
254 if self.enabled and self.conditions:
255 result, trace = await self.hass_api.trace_conditions(
256 self.conditions, condition_variables, trace_name=f"scenario_{self.name}"
257 )
258 if trace:
259 self.last_trace = trace
260 return result if result is not None else False
263@contextmanager
264def trace_action(
265 hass: HomeAssistant,
266 item_id: str,
267 config: dict[str, Any],
268 context: Context | None = None,
269 stored_traces: int = 5,
270) -> Iterator[ActionTrace]:
271 """Trace execution of a scenario."""
272 trace = ActionTrace(item_id, config, None, context or Context())
273 async_store_trace(hass, trace, stored_traces)
275 try:
276 yield trace
277 except Exception as ex:
278 if item_id:
279 trace.set_error(ex)
280 raise
281 finally:
282 if item_id:
283 trace.finished()