Coverage for custom_components/autoarm/hass_api.py: 93%

1from __future__ import annotations 

2 

3import logging 

4from functools import partial 

5from typing import TYPE_CHECKING, Any, cast 

6 

7from homeassistant.components.alarm_control_panel.const import AlarmControlPanelState 

8from homeassistant.core import HomeAssistant 

9from homeassistant.exceptions import ConditionError, ConditionErrorContainer 

10from homeassistant.helpers import condition as condition 

11from homeassistant.helpers import issue_registry as ir 

12from homeassistant.helpers.template import Template 

13from homeassistant.helpers.typing import ConfigType 

14 

15from .const import DOMAIN, ConditionVariables 

16 

17if TYPE_CHECKING: 

18 from collections.abc import Callable 

19 

20 from homeassistant.core import HomeAssistant 

21 from homeassistant.helpers.typing import ConfigType, TemplateVarsType 

22 

23 

24_LOGGER = logging.getLogger(__name__) 

25 

26 

27class HomeAssistantAPI: 

28 def __init__(self, hass: HomeAssistant | None = None) -> None: 

29 self._hass = hass 

30 

31 def raise_issue( 

32 self, 

33 issue_id: str, 

34 issue_key: str, 

35 issue_map: dict[str, str], 

36 severity: ir.IssueSeverity = ir.IssueSeverity.WARNING, 

37 learn_more_url: str = "https://autoarm.rhizomatics.org.uk", 

38 is_fixable: bool = False, 

39 ) -> None: 

40 if not self._hass: 

41 return 

42 ir.async_create_issue( 

43 self._hass, 

44 DOMAIN, 

45 issue_id, 

46 translation_key=issue_key, 

47 translation_placeholders=issue_map, 

48 severity=severity, 

49 learn_more_url=learn_more_url, 

50 is_fixable=is_fixable, 

51 ) 

52 

53 async def build_condition( 

54 self, condition_config: list[ConfigType], strict: bool = False, validate: bool = False, name: str = DOMAIN 

55 ) -> Callable[[TemplateVarsType], bool] | None: 

56 if self._hass is None: 

57 raise ValueError("HomeAssistant not available") 

58 capturing_logger: ConditionErrorLoggingAdaptor = ConditionErrorLoggingAdaptor(_LOGGER) 

59 condition_variables: ConditionVariables = ConditionVariables(None, None, False, AlarmControlPanelState.PENDING, {}) 

60 cond_list: list[ConfigType] 

61 try: 

62 if validate: 

63 cond_list = cast( 

64 "list[ConfigType]", await condition.async_validate_conditions_config(self._hass, condition_config) 

65 ) 

66 else: 

67 cond_list = condition_config 

68 except Exception as e: 

69 _LOGGER.exception("AUTOARM Condition validation failed: %s", e) 

70 raise 

71 try: 

72 if strict: 

73 force_strict_template_mode(cond_list, undo=False) 

74 

75 test: Callable[[TemplateVarsType], bool] = await condition.async_conditions_from_config( 

76 self._hass, cond_list, cast("logging.Logger", capturing_logger), name 

77 ) 

78 if test is None: 

79 raise ValueError(f"Invalid condition {condition_config}") 

80 test({DOMAIN: condition_variables.as_dict()}) 

81 if strict and capturing_logger.condition_errors: 

82 for exception in capturing_logger.condition_errors: 

83 _LOGGER.warning("AUTOARM Invalid condition %s:%s", condition_config, exception) 

84 raise capturing_logger.condition_errors[0] 

85 return test 

86 except Exception as e: 

87 _LOGGER.exception("AUTOARM Condition eval failed: %s", e) 

88 raise 

89 finally: 

90 if strict: 

91 force_strict_template_mode(condition_config, undo=True) 

92 

93 def evaluate_condition( 

94 self, 

95 checker: Callable[[TemplateVarsType], bool], 

96 condition_variables: ConditionVariables | None = None, 

97 ) -> bool | None: 

98 if self._hass is None: 

99 raise ValueError("HomeAssistant not available") 

100 try: 

101 return checker({DOMAIN: condition_variables.as_dict()} if condition_variables else None) 

102 except Exception as e: 

103 _LOGGER.error("AUTOARM Condition eval failed: %s", e) 

104 raise 

105 

106 def fire_event(self, event_name: str, event_data: dict[str, Any] | None = None) -> None: 

107 if self._hass is not None: 

108 _LOGGER.debug("AUTOARM Firing %s event: %s", event_name, event_data) 

109 self._hass.bus.async_fire(f"{DOMAIN}_{event_name}", event_data) 

110 

111 

112class ConditionErrorLoggingAdaptor(logging.LoggerAdapter["logging.Logger"]): 

113 def __init__(self, *args: Any, **kwargs: Any) -> None: 

114 super().__init__(*args, **kwargs) 

115 self.condition_errors: list[ConditionError] = [] 

116 

117 def capture(self, args: Any) -> None: 

118 if args and isinstance(args, (list, tuple)): 

119 for arg in args: 

120 if isinstance(arg, ConditionErrorContainer): 

121 self.condition_errors.extend(arg.errors) 

122 elif isinstance(arg, ConditionError): 

123 self.condition_errors.append(arg) 

124 

125 def error(self, msg: Any, *args: object, **kwargs: Any) -> None: 

126 self.capture(args) 

127 self.logger.error(msg, args, kwargs) 

128 

129 def warning(self, msg: Any, *args: Any, **kwargs: Any) -> None: 

130 self.capture(args) 

131 self.logger.warning(msg, args, kwargs) 

132 

133 

134def force_strict_template_mode(conditions: list[ConfigType], undo: bool = False) -> None: 

135 class TemplateWrapper: 

136 def __init__(self, obj: Template) -> None: 

137 self._obj = obj 

138 

139 def __getattr__(self, name: str) -> Any: 

140 if name == "async_render_to_info": 

141 return partial(self._obj.async_render_to_info, strict=True) 

142 return getattr(self._obj, name) 

143 

144 def __setattr__(self, name: str, value: Any) -> None: 

145 super().__setattr__(name, value) 

146 

147 def wrap_template(cond: ConfigType, undo: bool) -> ConfigType: 

148 for key, val in cond.items(): 

149 if not undo and isinstance(val, Template) and hasattr(val, "_env"): 

150 cond[key] = TemplateWrapper(val) 

151 elif undo and isinstance(val, TemplateWrapper): 

152 cond[key] = val._obj 

153 elif isinstance(val, dict): 

154 wrap_template(val, undo) 

155 return cond 

156 

157 if conditions is not None: 

158 conditions = [wrap_template(condition, undo) for condition in conditions]