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

1import logging 

2import re 

3from typing import TYPE_CHECKING, Any 

4 

5from homeassistant.const import CONF_ENABLED 

6from homeassistant.helpers import issue_registry as ir 

7 

8from . import ATTR_MEDIA, ConditionsFunc 

9from .hass_api import HomeAssistantAPI 

10from .model import DeliveryCustomization 

11 

12if TYPE_CHECKING: 

13 from homeassistant.core import HomeAssistant 

14 from homeassistant.helpers.typing import ConfigType 

15 

16from collections.abc import Iterator 

17from contextlib import contextmanager 

18 

19import voluptuous as vol 

20 

21# type: ignore[attr-defined,unused-ignore] 

22from homeassistant.components.trace import async_store_trace 

23from homeassistant.components.trace.models import ActionTrace 

24from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_NAME, CONF_ALIAS, CONF_CONDITION, CONF_CONDITIONS 

25from homeassistant.core import Context, HomeAssistant 

26from homeassistant.helpers.typing import ConfigType 

27 

28from . import ATTR_ENABLED, CONF_ACTION_GROUP_NAMES, CONF_DELIVERY, CONF_MEDIA 

29from .delivery import Delivery, DeliveryRegistry 

30from .model import ConditionVariables 

31 

32_LOGGER = logging.getLogger(__name__) 

33 

34 

35class ScenarioRegistry: 

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

37 self._config: ConfigType = scenario_configs or {} 

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

39 

40 async def initialize( 

41 self, 

42 delivery_registry: DeliveryRegistry, 

43 mobile_actions: ConfigType, 

44 hass_api: HomeAssistantAPI, 

45 ) -> None: 

46 

47 for scenario_name, scenario_definition in self._config.items(): 

48 scenario = Scenario(scenario_name, scenario_definition, delivery_registry, hass_api) 

49 if await scenario.validate(valid_action_group_names=list(mobile_actions)): 

50 self.scenarios[scenario_name] = scenario 

51 else: 

52 _LOGGER.warning("SUPERNOTIFY Scenario %s failed to validate, ignoring", scenario.name) 

53 

54 

55class Scenario: 

56 def __init__( 

57 self, name: str, scenario_definition: dict[str, Any], delivery_registry: DeliveryRegistry, hass_api: HomeAssistantAPI 

58 ) -> None: 

59 self.hass_api: HomeAssistantAPI = hass_api 

60 self.delivery_registry = delivery_registry 

61 self.enabled: bool = scenario_definition.get(CONF_ENABLED, True) 

62 self.name: str = name 

63 self.alias: str | None = scenario_definition.get(CONF_ALIAS) 

64 self.conditions: ConditionsFunc | None = None 

65 self.conditions_config: list[ConfigType] | None = scenario_definition.get(CONF_CONDITIONS) 

66 if not scenario_definition.get(CONF_CONDITIONS) and scenario_definition.get(CONF_CONDITION): 

67 self.conditions_config = scenario_definition.get(CONF_CONDITION) 

68 self.media: dict[str, Any] | None = scenario_definition.get(CONF_MEDIA) 

69 self.action_groups: list[str] = scenario_definition.get(CONF_ACTION_GROUP_NAMES, []) 

70 self._config_delivery: dict[str, DeliveryCustomization] 

71 self.delivery_overrides: dict[str, DeliveryCustomization] = {} 

72 self._delivery_selector: dict[str, str] = {} 

73 self.last_trace: ActionTrace | None = None 

74 self.startup_issue_count: int = 0 

75 

76 delivery_data = scenario_definition.get(CONF_DELIVERY) 

77 if isinstance(delivery_data, list): 

78 # a bare list of deliveries implies enabling 

79 _LOGGER.debug("SUPERNOTIFY scenario %s delivery default enabled for list %s", self.name, delivery_data) 

80 self._config_delivery = {k: DeliveryCustomization({CONF_ENABLED: True}) for k in delivery_data} 

81 elif isinstance(delivery_data, str) and delivery_data: 

82 # a bare list of deliveries implies enabled delivery 

83 _LOGGER.debug("SUPERNOTIFY scenario %s delivery default enabled for single %s", self.name, delivery_data) 

84 self._config_delivery = {delivery_data: DeliveryCustomization({CONF_ENABLED: True})} 

85 elif isinstance(delivery_data, dict): 

86 # whereas a dict may be used to tune or restrict 

87 _LOGGER.debug("SUPERNOTIFY scenario %s delivery selection %s", self.name, delivery_data) 

88 self._config_delivery = {k: DeliveryCustomization(v) for k, v in delivery_data.items()} 

89 elif delivery_data: 

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

91 self._config_delivery = {} 

92 else: 

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

94 self._config_delivery = {} 

95 

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

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

98 if self.conditions_config: 

99 error: str | None = None 

100 try: 

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

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

103 except vol.Invalid as vi: 

104 _LOGGER.error( 

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

106 ) 

107 error = f"Schema error {vi}" 

108 except Exception as e: 

109 _LOGGER.error( 

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

111 ) 

112 error = f"Unknown error {e}" 

113 if error is not None: 

114 self.startup_issue_count += 1 

115 self.hass_api.raise_issue( 

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

117 is_fixable=False, 

118 issue_key="scenario_condition", 

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

120 severity=ir.IssueSeverity.ERROR, 

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

122 ) 

123 

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

125 matched: bool = False 

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

127 if delivery: 

128 self.delivery_overrides[delivery.name] = config 

129 self._delivery_selector[delivery.name] = name_or_pattern 

130 matched = True 

131 else: 

132 # look for a wildcard match instead 

133 for delivery_name in self.delivery_registry.deliveries: 

134 if re.fullmatch(name_or_pattern, delivery_name): 

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

136 _LOGGER.info( 

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

138 ) 

139 else: 

140 _LOGGER.debug( 

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

142 ) 

143 self.delivery_overrides[delivery_name] = config 

144 self._delivery_selector[delivery_name] = name_or_pattern 

145 matched = True 

146 if not matched: 

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

148 self.startup_issue_count += 1 

149 self.hass_api.raise_issue( 

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

151 is_fixable=False, 

152 issue_key="scenario_delivery", 

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

154 severity=ir.IssueSeverity.WARNING, 

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

156 ) 

157 

158 if valid_action_group_names is not None: 

159 invalid_action_groups: list[str] = [] 

160 for action_group_name in self.action_groups: 

161 if action_group_name not in valid_action_group_names: 

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

163 invalid_action_groups.append(action_group_name) 

164 self.startup_issue_count += 1 

165 self.hass_api.raise_issue( 

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

167 is_fixable=False, 

168 issue_key="scenario_delivery", 

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

170 severity=ir.IssueSeverity.WARNING, 

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

172 ) 

173 for action_group_name in invalid_action_groups: 

174 self.action_groups.remove(action_group_name) 

175 

176 return self.startup_issue_count == 0 

177 

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

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

180 

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

182 return [ 

183 del_name 

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

185 if del_config.enabled or del_config.enabled is None 

186 ] 

187 

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

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

190 

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

192 return self.delivery_overrides.get(delivery_name) 

193 

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

195 """Return scenario attributes""" 

196 attrs = { 

197 ATTR_NAME: self.name, 

198 ATTR_ENABLED: self.enabled, 

199 ATTR_MEDIA: self.media, 

200 "action_groups": self.action_groups, 

201 "delivery": self.delivery_overrides, 

202 } 

203 if self.alias: 

204 attrs[ATTR_FRIENDLY_NAME] = self.alias 

205 if include_condition: 

206 attrs["conditions"] = self.conditions_config 

207 if include_trace and self.last_trace: 

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

209 return attrs 

210 

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

212 return self.delivery_overrides.get(delivery_name) 

213 

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

215 """Archive friendly view of scenario""" 

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

217 

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

219 """Evaluate scenario conditions""" 

220 result: bool | None = False 

221 if self.enabled and self.conditions: 

222 try: 

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

224 if result is None: 

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

226 except Exception as e: 

227 _LOGGER.error( 

228 f"SUPERNOTIFY Scenario {self.name} condition eval failed: %s, vars: %s", 

229 e, 

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

231 ) 

232 return result if result is not None else False 

233 

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

235 """Trace scenario condition execution""" 

236 result: bool | None = False 

237 trace: ActionTrace | None = None 

238 if self.enabled and self.conditions: 

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

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

241 ) 

242 if trace: 

243 self.last_trace = trace 

244 return result if result is not None else False 

245 

246 

247@contextmanager 

248def trace_action( 

249 hass: HomeAssistant, 

250 item_id: str, 

251 config: dict[str, Any], 

252 context: Context | None = None, 

253 stored_traces: int = 5, 

254) -> Iterator[ActionTrace]: 

255 """Trace execution of a scenario.""" 

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

257 async_store_trace(hass, trace, stored_traces) 

258 

259 try: 

260 yield trace 

261 except Exception as ex: 

262 if item_id: 

263 trace.set_error(ex) 

264 raise 

265 finally: 

266 if item_id: 

267 trace.finished()