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

1import logging 

2from typing import Any 

3 

4from homeassistant.auth import HomeAssistant 

5from homeassistant.components.alarm_control_panel.const import AlarmControlPanelState 

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

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, "data": data} 

107 if notify_targets: 

108 service_data["target"] = notify_targets 

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

110 _LOGGER.debug("AUTOARM Notifying %s.%s with %s", domain, action, service_data) 

111 await self.hass.services.async_call( 

112 domain, 

113 action, 

114 service_data=service_data, 

115 ) 

116 else: 

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

118 

119 except Exception as e: 

120 self.app_health_tracker.record_runtime_error() 

121 _LOGGER.exception("AUTOARM %s failed: %s", self.notify_action, e)