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 .const import ATTR_MEDIA 

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 

16 from .schema import ConditionsFunc 

17 

18from collections.abc import Iterator 

19from contextlib import contextmanager 

20 

21import voluptuous as vol 

22 

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

24from homeassistant.components.trace import async_store_trace 

25from homeassistant.components.trace.models import ActionTrace 

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

27from homeassistant.core import Context, HomeAssistant 

28from homeassistant.helpers.typing import ConfigType 

29 

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

31from .delivery import Delivery, DeliveryRegistry 

32from .model import ConditionVariables 

33 

34_LOGGER = logging.getLogger(__name__) 

35 

36 

37class ScenarioRegistry: 

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

39 self._config: ConfigType = scenario_configs or {} 

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

41 

42 async def initialize( 

43 self, 

44 delivery_registry: DeliveryRegistry, 

45 mobile_actions: ConfigType, 

46 hass_api: HomeAssistantAPI, 

47 ) -> None: 

48 

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

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

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

52 self.scenarios[scenario_name] = scenario 

53 else: 

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

55 

56 

57class Scenario: 

58 def __init__( 

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

60 ) -> None: 

61 self.hass_api: HomeAssistantAPI = hass_api 

62 self.delivery_registry = delivery_registry 

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

64 self.name: str = name 

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

66 self.conditions: ConditionsFunc | None = None 

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

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

69 self.conditions_config = scenario_definition.get(CONF_CONDITION) 

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 f"SUPERNOTIFY Scenario {self.name} condition eval failed: %s, vars: %s", 

231 e, 

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

233 ) 

234 return result if result is not None else False 

235 

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

237 """Trace scenario condition execution""" 

238 result: bool | None = False 

239 trace: ActionTrace | None = None 

240 if self.enabled and self.conditions: 

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

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

243 ) 

244 if trace: 

245 self.last_trace = trace 

246 return result if result is not None else False 

247 

248 

249@contextmanager 

250def trace_action( 

251 hass: HomeAssistant, 

252 item_id: str, 

253 config: dict[str, Any], 

254 context: Context | None = None, 

255 stored_traces: int = 5, 

256) -> Iterator[ActionTrace]: 

257 """Trace execution of a scenario.""" 

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

259 async_store_trace(hass, trace, stored_traces) 

260 

261 try: 

262 yield trace 

263 except Exception as ex: 

264 if item_id: 

265 trace.set_error(ex) 

266 raise 

267 finally: 

268 if item_id: 

269 trace.finished()