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

1import logging 

2from collections.abc import Iterator 

3from contextlib import contextmanager 

4from dataclasses import asdict 

5from typing import Any 

6 

7import voluptuous as vol 

8from homeassistant.components.trace import async_setup, async_store_trace # type: ignore[attr-defined] 

9from homeassistant.components.trace.const import DATA_TRACE 

10from homeassistant.components.trace.models import ActionTrace 

11from homeassistant.const import ( 

12 CONF_ALIAS, 

13 CONF_CONDITION, 

14) 

15from homeassistant.core import Context, HomeAssistant 

16from homeassistant.helpers import condition 

17from homeassistant.helpers import issue_registry as ir 

18from homeassistant.helpers.trace import trace_get, trace_path 

19from homeassistant.helpers.typing import ConfigType 

20from voluptuous import Invalid 

21 

22from . import ( 

23 ATTR_DEFAULT, 

24 CONF_ACTION_GROUP_NAMES, 

25 CONF_DELIVERY, 

26 CONF_DELIVERY_SELECTION, 

27 CONF_MEDIA, 

28 DOMAIN, 

29 ConditionVariables, 

30) 

31 

32_LOGGER = logging.getLogger(__name__) 

33 

34 

35class Scenario: 

36 def __init__(self, name: str, scenario_definition: dict[str, Any], hass: HomeAssistant) -> None: 

37 self.hass: HomeAssistant = hass 

38 self.name: str = name 

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

40 self.condition: ConfigType | None = scenario_definition.get(CONF_CONDITION) 

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

42 self.delivery_selection: str | None = scenario_definition.get(CONF_DELIVERY_SELECTION) 

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

44 self.delivery: dict[str, Any] = scenario_definition.get(CONF_DELIVERY) or {} 

45 self.default: bool = self.name == ATTR_DEFAULT 

46 self.last_trace: ActionTrace | None = None 

47 self.condition_func = None 

48 

49 async def validate(self, valid_deliveries: list[str] | None = None, valid_action_groups: list[str] | None = None) -> bool: 

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

51 if self.condition: 

52 error: str | None = None 

53 try: 

54 cond: ConfigType = await condition.async_validate_condition_config(self.hass, self.condition) 

55 if await condition.async_from_config(self.hass, cond) is None: 

56 _LOGGER.warning("SUPERNOTIFY Disabling scenario %s with failed condition %s", self.name, self.condition) 

57 error = "Unable to build condition from definition" 

58 except vol.Invalid as vi: 

59 _LOGGER.error( 

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

61 ) 

62 error = f"Schema error {vi}" 

63 except Exception as e: 

64 _LOGGER.error("SUPERNOTIFY Disabling scenario %s with error validating %s: %s", self.name, self.condition, e) 

65 error = f"Unknown error {e}" 

66 if error is not None: 

67 ir.async_create_issue( 

68 self.hass, 

69 DOMAIN, 

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

71 is_fixable=False, 

72 translation_key="scenario_condition", 

73 translation_placeholders={"scenario": self.name, "error": error}, 

74 severity=ir.IssueSeverity.ERROR, 

75 learn_more_url="https://jeyrb.github.io/hass_supernotify/#scenarios", 

76 ) 

77 return False 

78 

79 if valid_deliveries is not None: 

80 invalid_deliveries: list[str] = [] 

81 for delivery_name in self.delivery: 

82 if delivery_name not in valid_deliveries: 

83 _LOGGER.error(f"SUPERNOTIFY Unknown delivery {delivery_name} removed from scenario {self.name}") 

84 invalid_deliveries.append(delivery_name) 

85 ir.async_create_issue( 

86 self.hass, 

87 DOMAIN, 

88 f"scenario_{self.name}_delivery_{delivery_name}", 

89 is_fixable=False, 

90 translation_key="scenario_delivery", 

91 translation_placeholders={"scenario": self.name, "delivery": delivery_name}, 

92 severity=ir.IssueSeverity.WARNING, 

93 learn_more_url="https://jeyrb.github.io/hass_supernotify/#scenarios", 

94 ) 

95 for delivery_name in invalid_deliveries: 

96 del self.delivery[delivery_name] 

97 

98 if valid_action_groups is not None: 

99 invalid_action_groups: list[str] = [] 

100 for action_group_name in self.action_groups: 

101 if action_group_name not in valid_action_groups: 

102 _LOGGER.error(f"SUPERNOTIFY Unknown delivery {action_group_name} removed from scenario {self.name}") 

103 invalid_action_groups.append(action_group_name) 

104 ir.async_create_issue( 

105 self.hass, 

106 DOMAIN, 

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

108 is_fixable=False, 

109 translation_key="scenario_delivery", 

110 translation_placeholders={"scenario": self.name, "action_group": action_group_name}, 

111 severity=ir.IssueSeverity.WARNING, 

112 learn_more_url="https://jeyrb.github.io/hass_supernotify/#scenarios", 

113 ) 

114 for action_group_name in invalid_action_groups: 

115 self.action_groups.remove(action_group_name) 

116 return True 

117 

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

119 """Return scenario attributes""" 

120 attrs = { 

121 "name": self.name, 

122 "alias": self.alias, 

123 "media": self.media, 

124 "delivery_selection": self.delivery_selection, 

125 "action_groups": self.action_groups, 

126 "delivery": self.delivery, 

127 "default": self.default, 

128 } 

129 if include_condition: 

130 attrs["condition"] = self.condition 

131 if include_trace and self.last_trace: 

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

133 return attrs 

134 

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

136 """Archive friendly view of scenario""" 

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

138 

139 async def evaluate(self, condition_variables: ConditionVariables | None = None) -> bool: 

140 """Evaluate scenario conditions""" 

141 if self.condition: 

142 try: 

143 test = await condition.async_from_config(self.hass, self.condition) 

144 if test is None: 

145 raise Invalid(f"Empty condition generated for {self.name}") 

146 except Exception as e: 

147 _LOGGER.error("SUPERNOTIFY Scenario %s condition create failed: %s", self.name, e) 

148 return False 

149 try: 

150 if test(self.hass, asdict(condition_variables) if condition_variables else None): 

151 return True 

152 except Exception as e: 

153 _LOGGER.error( 

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

155 e, 

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

157 ) 

158 return False 

159 

160 async def trace(self, condition_variables: ConditionVariables | None = None, config: ConfigType | None = None) -> bool: 

161 """Trace scenario delivery""" 

162 result = None 

163 config = {} if config is None else config 

164 if DATA_TRACE not in self.hass.data: 

165 await async_setup(self.hass, config) 

166 with trace_action(self.hass, f"scenario_{self.name}", config) as scenario_trace: 

167 scenario_trace.set_trace(trace_get()) 

168 self.last_trace = scenario_trace 

169 with trace_path(["condition", "conditions"]) as _tp: 

170 result = await self.evaluate(condition_variables) 

171 _LOGGER.info(scenario_trace.as_dict()) 

172 return result 

173 

174 

175@contextmanager 

176def trace_action( 

177 hass: HomeAssistant, 

178 item_id: str, 

179 config: dict[str, Any], 

180 context: Context | None = None, 

181 stored_traces: int = 5, 

182) -> Iterator[ActionTrace]: 

183 """Trace execution of a scenario.""" 

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

185 async_store_trace(hass, trace, stored_traces) 

186 

187 try: 

188 yield trace 

189 except Exception as ex: 

190 if item_id: 

191 trace.set_error(ex) 

192 raise 

193 finally: 

194 if item_id: 

195 trace.finished()