Coverage for custom_components/autoarm/notifier.py: 100%

1import logging 

2from typing import Any 

3 

4from homeassistant.components.alarm_control_panel.const import AlarmControlPanelState 

5from homeassistant.const import CONF_SERVICE, CONF_SOURCE, CONF_STATE, CONF_TARGET 

6from homeassistant.core import HomeAssistant 

7 

8from custom_components.autoarm.const import ALARM_STATES, CONF_SCENARIO, CONF_SUPERNOTIFY, NOTIFY_COMMON, ChangeSource 

9from custom_components.autoarm.helpers import AppHealthTracker 

10 

11_LOGGER = logging.getLogger(__name__) 

12 

13 

14class Notifier: 

15 def __init__( 

16 self, 

17 notify_profiles: dict[str, dict[str, Any]] | None, 

18 hass: HomeAssistant, 

19 app_health_tracker: AppHealthTracker, 

20 notify_action: str | None, 

21 notify_targets: list[str] | None = None, 

22 ) -> None: 

23 self.notify_profiles: dict[str, dict[str, Any]] = notify_profiles or {} 

24 self.hass: HomeAssistant = hass 

25 self.app_health_tracker: AppHealthTracker = app_health_tracker 

26 self.notify_action: str | None = notify_action 

27 self.notify_targets: list[str] = notify_targets or [] 

28 

29 async def notify( 

30 self, 

31 source: ChangeSource, 

32 from_state: AlarmControlPanelState | None = None, 

33 to_state: AlarmControlPanelState | None = None, 

34 message: str | None = None, 

35 title: str | None = None, 

36 ) -> None: 

37 

38 try: 

39 selected_profile: dict[str, Any] | None = None 

40 selected_profile_name: str | None = None 

41 config_by_state_pickiness = sorted( 

42 self.notify_profiles, key=lambda v: len(self.notify_profiles[v].get(CONF_STATE, ALARM_STATES)) 

43 ) 

44 for profile_name in config_by_state_pickiness: 

45 if profile_name == NOTIFY_COMMON: 

46 continue 

47 profile: dict[str, Any] = self.notify_profiles[profile_name] 

48 if profile.get(CONF_SOURCE) and source not in profile.get(CONF_SOURCE, []): 

49 _LOGGER.debug( 

50 "AUTOARM Notification not selected for %s profile for source match on %s", profile_name, source 

51 ) 

52 continue 

53 only_for_states: list[AlarmControlPanelState] | None = profile.get(CONF_STATE) 

54 if only_for_states and from_state not in only_for_states and to_state not in only_for_states: 

55 _LOGGER.debug( 

56 "AUTOARM Notification not selected for %s profile for state match on %s->%s", 

57 profile_name, 

58 from_state, 

59 to_state, 

60 ) 

61 continue 

62 selected_profile = profile 

63 selected_profile_name = profile_name 

64 break 

65 if selected_profile is None: 

66 _LOGGER.debug("AUTOARM No profile selected for %s notification: %s", source, message) 

67 return 

68 

69 # separately merge base dict and data sub-dict as cheap and nasty semi-deep-merge 

70 base_profile = self.notify_profiles.get(NOTIFY_COMMON, {}) 

71 base_profile_data = base_profile.get("data", {}) 

72 merged_profile = dict(base_profile) 

73 merged_profile_data = dict(base_profile_data) 

74 if selected_profile is not None: 

75 selected_profile_data: dict[str, Any] = selected_profile.get("data", {}) 

76 merged_profile.update(selected_profile) 

77 merged_profile_data.update(selected_profile_data) 

78 merged_profile["data"] = merged_profile_data 

79 

80 data = merged_profile.get("data", {}) 

81 if "source" in data and data["source"] is None: 

82 data["source"] = str(source) 

83 if "profile" in data and data["profile"] is None: 

84 data["profile"] = selected_profile_name 

85 if merged_profile.get(CONF_SUPERNOTIFY) and merged_profile.get(CONF_SCENARIO): 

86 data["apply_scenarios"] = merged_profile.get(CONF_SCENARIO) 

87 

88 notify_action: str | None = merged_profile.get(CONF_SERVICE, self.notify_action) 

89 notify_targets: list[str] | None = merged_profile.get(CONF_TARGET, self.notify_targets) 

90 if notify_action is None: 

91 _LOGGER.debug("AUTOARM Notifications disabled, no notification action") 

92 return 

93 if notify_action == "notify.send_message" and not notify_targets: 

94 _LOGGER.debug("AUTOARM Notifications disabled, no targets for notify.send_message") 

95 return 

96 

97 if title is None: 

98 title = f"Alarm now {to_state}" if to_state else "Alarm Panel Change" 

99 if message is None: 

100 if from_state and to_state: 

101 message = f"Alarm state changed from {from_state} to {to_state} by {source.capitalize()}" 

102 else: 

103 message = "Alarm control panel operation complete" 

104 

105 if notify_action and merged_profile: 

106 service_data: dict[str, Any] = {"message": message, "title": title} 

107 if data and notify_action != "notify.send_message": 

108 # new notify entities can't handle nested data section 

109 service_data["data"] = data 

110 domain, action = notify_action.split(".", 1) 

111 target: dict[str, Any] | None = None 

112 if notify_targets: 

113 if domain == "notify" and action == "send_message": 

114 # notify.send_message is entity-based and targets via entity_id, 

115 # not a "target" key in service_data like legacy notify services 

116 target = {"entity_id": notify_targets} 

117 else: 

118 service_data["target"] = notify_targets 

119 

120 _LOGGER.debug("AUTOARM Notifying %s.%s with %s target=%s", domain, action, service_data, target) 

121 try: 

122 await self.hass.services.async_call( 

123 domain, 

124 action, 

125 service_data=service_data, 

126 target=target, 

127 ) 

128 except Exception as e: 

129 _LOGGER.error("AUTOARM notify failure, %s.%s: %s [%s]", domain, action, service_data, e) 

130 raise 

131 else: 

132 _LOGGER.debug("AUTOARM Skipped notification, service: %s, data: %s", self.notify_action, merged_profile) 

133 

134 except Exception: 

135 self.app_health_tracker.record_runtime_error() 

136 _LOGGER.exception("AUTOARM %s failed", self.notify_action)