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

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("Notification not selected for %s profile for source match on %s", profile_name, source) 

50 continue 

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

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

53 _LOGGER.debug( 

54 "Notification not selected for %s profile for state match on %s->%s", profile_name, from_state, to_state 

55 ) 

56 continue 

57 selected_profile = profile 

58 selected_profile_name = profile_name 

59 break 

60 if selected_profile is None: 

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

62 return 

63 

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

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

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

67 merged_profile = dict(base_profile) 

68 merged_profile_data = dict(base_profile_data) 

69 if selected_profile is not None: 

70 selected_profile_data: dict = selected_profile.get("data", {}) 

71 merged_profile.update(selected_profile) 

72 merged_profile_data.update(selected_profile_data) 

73 merged_profile["data"] = merged_profile_data 

74 

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

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

77 data["source"] = source 

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

79 data["profile"] = selected_profile_name 

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

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

82 

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

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

85 if notify_action is None: 

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

87 return 

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

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

90 return 

91 

92 if title is None: 

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

94 if message is None: 

95 if from_state and to_state: 

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

97 else: 

98 message = "Alarm control panel operation complete" 

99 

100 if notify_action and merged_profile: 

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

102 if notify_targets: 

103 service_data["target"] = notify_targets 

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

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

106 await self.hass.services.async_call( 

107 domain, 

108 action, 

109 service_data=service_data, 

110 ) 

111 else: 

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

113 

114 except Exception as e: 

115 self.app_health_tracker.record_runtime_error() 

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