Coverage for custom_components/supernotify/scenario.py: 90%

1from __future__ import annotations 

2 

3import logging 

4import re 

5from typing import TYPE_CHECKING, Any 

6 

7from homeassistant.const import CONF_ENABLED 

8from homeassistant.helpers import issue_registry as ir 

9 

10from .const import ATTR_MEDIA 

11from .model import DeliveryCustomization 

12 

13if TYPE_CHECKING: 

14 from collections.abc import Iterator 

15 

16 from homeassistant.core import HomeAssistant 

17 from homeassistant.helpers.typing import ConfigType 

18 

19 from .delivery import Delivery, DeliveryRegistry 

20 from .hass_api import HomeAssistantAPI 

21 from .schema import ConditionsFunc 

22 

23from contextlib import contextmanager 

24 

25import voluptuous as vol 

26 

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 

32 

33from .const import ATTR_ENABLED, CONF_ACTION_GROUP_NAMES, CONF_DELIVERY, CONF_MEDIA 

34from .model import ConditionVariables 

35 

36_LOGGER = logging.getLogger(__name__) 

37 

38 

39class ScenarioRegistry: 

40 def __init__(self, scenario_configs: ConfigType) -> None: 

41 self._config: ConfigType = scenario_configs or {} 

42 self.scenarios: dict[str, Scenario] = {} 

43 

44 async def initialize( 

45 self, 

46 delivery_registry: DeliveryRegistry, 

47 mobile_actions: ConfigType, 

48 hass_api: HomeAssistantAPI, 

49 ) -> None: 

50 

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) 

57 

58 

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 

77 

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({CONF_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({CONF_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 = {k: DeliveryCustomization(v) for k, v in delivery_data.items()} 

91 elif delivery_data: 

92 _LOGGER.warning("SUPERNOTIFY Unable to interpret scenario %s delivery data %s", self.name, delivery_data) 

93 self._config_delivery = {} 

94 else: 

95 _LOGGER.warning("SUPERNOTIFY No delivery definitions for scenario %s", self.name) 

96 self._config_delivery = {} 

97 

98 async def validate(self, valid_action_group_names: list[str] | None = None) -> bool: 

99 """Validate Home Assistant conditiion definition at initiation""" 

100 if self.conditions_config: 

101 error: str | None = None 

102 try: 

103 # note: basic template syntax within conditions already validated by voluptuous checks 

104 self.conditions = await self.hass_api.build_conditions(self.conditions_config, strict=True, validate=True) 

105 except vol.Invalid as vi: 

106 _LOGGER.error( 

107 f"SUPERNOTIFY Condition definition for scenario {self.name} fails Home Assistant schema check {vi}" 

108 ) 

109 error = f"Schema error {vi}" 

110 except Exception as e: 

111 _LOGGER.error( 

112 "SUPERNOTIFY Disabling scenario %s with error validating %s: %s", self.name, self.conditions_config, e 

113 ) 

114 error = f"Unknown error {e}" 

115 if error is not None: 

116 self.startup_issue_count += 1 

117 self.hass_api.raise_issue( 

118 f"scenario_{self.name}_condition", 

119 is_fixable=False, 

120 issue_key="scenario_condition", 

121 issue_map={"scenario": self.name, "error": error}, 

122 severity=ir.IssueSeverity.ERROR, 

123 learn_more_url="https://supernotify.rhizomatics.org.uk/scenarios/", 

124 ) 

125 

126 for name_or_pattern, config in self._config_delivery.items(): 

127 matched: bool = False 

128 delivery: Delivery | None = self.delivery_registry.deliveries.get(name_or_pattern) 

129 if delivery: 

130 self.delivery_overrides[delivery.name] = config 

131 self._delivery_selector[delivery.name] = name_or_pattern 

132 matched = True 

133 else: 

134 # look for a wildcard match instead 

135 for delivery_name in self.delivery_registry.deliveries: 

136 if re.fullmatch(name_or_pattern, delivery_name): 

137 if self._delivery_selector.get(delivery_name) == delivery_name: 

138 _LOGGER.info( 

139 f"SUPERNOTIFY Scenario {self.name} ignoring '{name_or_pattern}' shadowing explicit delivery {delivery_name}" # noqa: E501 

140 ) 

141 else: 

142 _LOGGER.debug( 

143 f"SUPERNOTIFY Scenario {self.name} delivery '{name_or_pattern}' matched {delivery_name}" 

144 ) 

145 self.delivery_overrides[delivery_name] = config 

146 self._delivery_selector[delivery_name] = name_or_pattern 

147 matched = True 

148 if not matched: 

149 _LOGGER.error(f"SUPERNOTIFY Scenario {self.name} has delivery {name_or_pattern} not found") 

150 self.startup_issue_count += 1 

151 self.hass_api.raise_issue( 

152 f"scenario_{self.name}_delivery_{name_or_pattern.replace('.', 'DOT').replace('*', 'STAR')}", 

153 is_fixable=False, 

154 issue_key="scenario_delivery", 

155 issue_map={"scenario": self.name, "delivery": name_or_pattern}, 

156 severity=ir.IssueSeverity.WARNING, 

157 learn_more_url="https://supernotify.rhizomatics.org.uk/scenarios/", 

158 ) 

159 

160 if valid_action_group_names is not None: 

161 invalid_action_groups: list[str] = [] 

162 for action_group_name in self.action_groups: 

163 if action_group_name not in valid_action_group_names: 

164 _LOGGER.error(f"SUPERNOTIFY Unknown action group {action_group_name} removed from scenario {self.name}") 

165 invalid_action_groups.append(action_group_name) 

166 self.startup_issue_count += 1 

167 self.hass_api.raise_issue( 

168 f"scenario_{self.name}_action_group_{action_group_name}", 

169 is_fixable=False, 

170 issue_key="scenario_delivery", 

171 issue_map={"scenario": self.name, "action_group": action_group_name}, 

172 severity=ir.IssueSeverity.WARNING, 

173 learn_more_url="https://supernotify.rhizomatics.org.uk/scenarios/", 

174 ) 

175 for action_group_name in invalid_action_groups: 

176 self.action_groups.remove(action_group_name) 

177 

178 return self.startup_issue_count == 0 

179 

180 def enabling_deliveries(self) -> list[str]: 

181 return [del_name for del_name, del_config in self.delivery_overrides.items() if del_config.enabled] 

182 

183 def relevant_deliveries(self) -> list[str]: 

184 return [ 

185 del_name 

186 for del_name, del_config in self.delivery_overrides.items() 

187 if del_config.enabled or del_config.enabled is None 

188 ] 

189 

190 def disabling_deliveries(self) -> list[str]: 

191 return [del_name for del_name, del_config in self.delivery_overrides.items() if del_config.enabled is False] 

192 

193 def delivery_customization(self, delivery_name: str) -> DeliveryCustomization | None: 

194 return self.delivery_overrides.get(delivery_name) 

195 

196 def attributes(self, include_condition: bool = True, include_trace: bool = False) -> dict[str, Any]: 

197 """Return scenario attributes""" 

198 attrs = { 

199 ATTR_NAME: self.name, 

200 ATTR_ENABLED: self.enabled, 

201 ATTR_MEDIA: self.media, 

202 "action_groups": self.action_groups, 

203 "delivery": self.delivery_overrides, 

204 } 

205 if self.alias: 

206 attrs[ATTR_FRIENDLY_NAME] = self.alias 

207 if include_condition: 

208 attrs["conditions"] = self.conditions_config 

209 if include_trace and self.last_trace: 

210 attrs["trace"] = self.last_trace.as_extended_dict() 

211 return attrs 

212 

213 def delivery_config(self, delivery_name: str) -> DeliveryCustomization | None: 

214 return self.delivery_overrides.get(delivery_name) 

215 

216 def contents(self, minimal: bool = False, **_kwargs: Any) -> dict[str, Any]: 

217 """Archive friendly view of scenario""" 

218 return self.attributes(include_condition=False, include_trace=not minimal) 

219 

220 def evaluate(self, condition_variables: ConditionVariables) -> bool: 

221 """Evaluate scenario conditions""" 

222 result: bool | None = False 

223 if self.enabled and self.conditions: 

224 try: 

225 result = self.hass_api.evaluate_conditions(self.conditions, condition_variables) 

226 if result is None: 

227 _LOGGER.warning(f"SUPERNOTIFY Scenario {self.name} condition empty result") 

228 except Exception as e: 

229 _LOGGER.error( 

230 "SUPERNOTIFY Scenario %s condition eval failed: %s, vars: %s", 

231 self.name, 

232 e, 

233 condition_variables.as_dict() if condition_variables else {}, 

234 ) 

235 return result if result is not None else False 

236 

237 async def trace(self, condition_variables: ConditionVariables) -> bool: 

238 """Trace scenario condition execution""" 

239 result: bool | None = False 

240 trace: ActionTrace | None = None 

241 if self.enabled and self.conditions: 

242 result, trace = await self.hass_api.trace_conditions( 

243 self.conditions, condition_variables, trace_name=f"scenario_{self.name}" 

244 ) 

245 if trace: 

246 self.last_trace = trace 

247 return result if result is not None else False 

248 

249 

250@contextmanager 

251def trace_action( 

252 hass: HomeAssistant, 

253 item_id: str, 

254 config: dict[str, Any], 

255 context: Context | None = None, 

256 stored_traces: int = 5, 

257) -> Iterator[ActionTrace]: 

258 """Trace execution of a scenario.""" 

259 trace = ActionTrace(item_id, config, None, context or Context()) 

260 async_store_trace(hass, trace, stored_traces) 

261 

262 try: 

263 yield trace 

264 except Exception as ex: 

265 if item_id: 

266 trace.set_error(ex) 

267 raise 

268 finally: 

269 if item_id: 

270 trace.finished()