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.exceptions import ConditionError, ConditionErrorContainer 

9from homeassistant.helpers import condition as condition_helper 

10from homeassistant.helpers import issue_registry as ir 

11from homeassistant.helpers.template import Template 

12 

13from .const import DOMAIN, ConditionVariables 

14 

15if TYPE_CHECKING: 

16 from collections.abc import Callable 

17 

18 from homeassistant.core import HomeAssistant 

19 from homeassistant.helpers.typing import ConfigType, TemplateVarsType 

20 

21 

22_LOGGER = logging.getLogger(__name__) 

23 

24 

25class HomeAssistantAPI: 

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

27 self._hass = hass 

28 

29 def raise_issue( 

30 self, 

31 issue_id: str, 

32 issue_key: str, 

33 issue_map: dict[str, str], 

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

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

36 is_fixable: bool = False, 

37 ) -> None: 

38 if not self._hass: 

39 return 

40 ir.async_create_issue( 

41 self._hass, 

42 DOMAIN, 

43 issue_id, 

44 translation_key=issue_key, 

45 translation_placeholders=issue_map, 

46 severity=severity, 

47 learn_more_url=learn_more_url, 

48 is_fixable=is_fixable, 

49 ) 

50 

51 async def build_condition( 

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

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

54 if self._hass is None: 

55 raise ValueError("HomeAssistant not available") 

56 capturing_logger: ConditionErrorLoggingAdaptor = ConditionErrorLoggingAdaptor(_LOGGER) 

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

58 cond_list: list[ConfigType] 

59 try: 

60 if validate: 

61 cond_list = cast( 

62 "list[ConfigType]", await condition_helper.async_validate_conditions_config(self._hass, condition_config) 

63 ) 

64 else: 

65 cond_list = condition_config 

66 except Exception: 

67 _LOGGER.exception("AUTOARM Condition validation failed") 

68 raise 

69 try: 

70 if strict: 

71 force_strict_template_mode(cond_list, undo=False) 

72 

73 test: Callable[[TemplateVarsType], bool] = await condition_helper.async_conditions_from_config( 

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

75 ) 

76 if test is None: 

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

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

79 if strict and capturing_logger.condition_errors: 

80 for exception in capturing_logger.condition_errors: 

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

82 raise capturing_logger.condition_errors[0] 

83 return test 

84 except Exception: 

85 _LOGGER.exception("AUTOARM Condition eval failed") 

86 raise 

87 finally: 

88 if strict: 

89 force_strict_template_mode(condition_config, undo=True) 

90 

91 def evaluate_condition( 

92 self, 

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

94 condition_variables: ConditionVariables | None = None, 

95 ) -> bool | None: 

96 if self._hass is None: 

97 raise ValueError("HomeAssistant not available") 

98 try: 

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

100 except Exception as e: 

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

102 raise 

103 

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

105 if self._hass is not None: 

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

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

108 

109 

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

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

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

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

114 

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

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

117 for arg in args: 

118 if isinstance(arg, ConditionErrorContainer): 

119 self.condition_errors.extend(arg.errors) 

120 elif isinstance(arg, ConditionError): 

121 self.condition_errors.append(arg) 

122 

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

124 self.capture(args) 

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

126 

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

128 self.capture(args) 

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

130 

131 

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

133 class TemplateWrapper: 

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

135 self._obj = obj 

136 

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

138 if name == "async_render_to_info": 

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

140 return getattr(self._obj, name) 

141 

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

143 super().__setattr__(name, value) 

144 

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

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

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

148 cond[key] = TemplateWrapper(val) 

149 elif undo and isinstance(val, TemplateWrapper): 

150 cond[key] = val._obj 

151 elif isinstance(val, dict): 

152 wrap_template(val, undo) 

153 return cond 

154 

155 if conditions is not None: 

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