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

1"""The notification for each decision of the tracker, through Supernotify, or a plain one without it""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from typing import TYPE_CHECKING, Any 

7 

8from homeassistant.const import ATTR_ENTITY_ID, CONF_TYPE 

9from homeassistant.exceptions import HomeAssistantError 

10from homeassistant.helpers import entity_registry as er 

11 

12from .const import ( 

13 ACTION_VIEW_SNAPSHOT, 

14 CONF_CUSTOM_TARGET, 

15 CONF_DELIVERIES, 

16 CONF_SOURCE_ENTRY_ID, 

17 CONF_TARGET, 

18 CONF_TARGETS, 

19 MESSAGE_ENDED, 

20 MESSAGE_IN_ZONES, 

21 MESSAGE_ON_CAMERA, 

22 MESSAGE_SEEN, 

23 SUPERNOTIFY_DOMAIN, 

24 TYPE_ZONE, 

25) 

26from .frigate import base_url, camera_entity_id, hls_url, snapshot_url 

27from .supernotify_api import supernotify_available 

28 

29if TYPE_CHECKING: 

30 from homeassistant.config_entries import ConfigEntry 

31 from homeassistant.core import Context, HomeAssistant 

32 

33 from .tracker import Decision 

34 

35_LOGGER = logging.getLogger(__name__) 

36 

37MOBILE_APP_PLATFORM = "mobile_app" 

38# by severity, an alert to be noticed and a detection not to interrupt, both left to Supernotify to map to each 

39# platform, as iOS's time sensitive or Android's high importance 

40PRIORITY_ALERT = "high" 

41PRIORITY_DETECTION = "medium" 

42# an update with alert once, no sound or interruption, iOS's passive 

43PRIORITY_SILENT = "low" 

44SEVERITY_ALERT = "alert" 

45 

46 

47def _friendly(name: str) -> str: 

48 """A zone or camera as Frigate writes it, `Barn_Door`, as said in a notification""" 

49 return name.replace("_", " ").title() 

50 

51 

52class Notifier: 

53 """Sends each notification for a camera's or a zone's entry""" 

54 

55 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: 

56 self.hass = hass 

57 self.entry = entry 

58 self.name: str = entry.title 

59 self.is_zone: bool = entry.data.get(CONF_TYPE) == TYPE_ZONE 

60 self.source_entry_id: str = entry.data[CONF_SOURCE_ENTRY_ID] 

61 

62 def message(self, decision: Decision) -> str: 

63 review = decision.review 

64 if self.is_zone: 

65 message = MESSAGE_ON_CAMERA.format(label=review.label, camera=_friendly(review.camera)) 

66 elif review.zones: 

67 message = MESSAGE_IN_ZONES.format(label=review.label, zones=", ".join(_friendly(z) for z in review.zones)) 

68 else: 

69 message = MESSAGE_SEEN.format(label=review.label) 

70 return MESSAGE_ENDED.format(message=message) if decision.is_final else message 

71 

72 def supernotify_data(self, decision: Decision) -> dict[str, Any]: 

73 """The `supernotify.notify` call: the snapshot, tapped for the clip as HLS, and never `clip_url`, which iOS 

74 fails to show, even the image""" 

75 review = decision.review 

76 if decision.silent: 

77 priority = PRIORITY_SILENT 

78 else: 

79 priority = PRIORITY_ALERT if review.severity == SEVERITY_ALERT else PRIORITY_DETECTION 

80 data: dict[str, Any] = { 

81 "title": self.name, 

82 "message": self.message(decision), 

83 "priority": priority, 

84 "extra_data": { 

85 # each update of a review takes the place of its last notification 

86 "mobile_push_notification_tag": f"frigate_{review.id}", 

87 "group": f"frigate_{review.camera}", 

88 }, 

89 } 

90 media: dict[str, Any] = {} 

91 if camera := camera_entity_id(self.hass, self.source_entry_id, review.camera): 

92 # for Supernotify's camera snooze, the snapshot still being the image 

93 media["camera_entity_id"] = camera 

94 base = base_url(self.hass) 

95 if base and decision.detection_id: 

96 snapshot = snapshot_url(base, decision.detection_id) 

97 media["snapshot_url"] = snapshot 

98 data["actions"] = [ 

99 {"level": "notification", "uri": hls_url(base, decision.detection_id, review.camera)}, 

100 {"action": "URI", "title": ACTION_VIEW_SNAPSHOT, "uri": snapshot}, 

101 ] 

102 if media: 

103 data["media"] = media 

104 if deliveries := self.entry.options.get(CONF_DELIVERIES): 

105 data["delivery"] = list(deliveries) 

106 data["delivery_selection"] = "explicit" 

107 for key in (CONF_TARGET, CONF_CUSTOM_TARGET): 

108 if value := self.entry.options.get(key): 

109 data[key] = value 

110 return data 

111 

112 async def async_notify(self, decision: Decision, context: Context) -> None: 

113 if not supernotify_available(self.hass): 

114 await self._send_message(decision, context) 

115 return 

116 try: 

117 await self.hass.services.async_call( 

118 SUPERNOTIFY_DOMAIN, "notify", self.supernotify_data(decision), blocking=True, context=context 

119 ) 

120 except HomeAssistantError: 

121 _LOGGER.exception("CAMERA_AUTONOTIFY Supernotify notification failed for %s", self.name) 

122 

123 async def _send_message(self, decision: Decision, context: Context) -> None: 

124 """Plain text only, with no snapshot or link, Supernotify being what's needed for those""" 

125 if decision.is_update: 

126 # each would be a new notification, with nothing to take the place of the last 

127 return 

128 # with none chosen, every phone with the mobile app is told 

129 targets: list[str] = self.entry.options.get(CONF_TARGETS) or [ 

130 entity.entity_id 

131 for entity in er.async_get(self.hass).entities.values() 

132 if entity.domain == "notify" and entity.platform == MOBILE_APP_PLATFORM and not entity.disabled 

133 ] 

134 if not targets: 

135 _LOGGER.warning("CAMERA_AUTONOTIFY Nothing to notify for %s, choose targets in its settings", self.name) 

136 return 

137 try: 

138 await self.hass.services.async_call( 

139 "notify", 

140 "send_message", 

141 {ATTR_ENTITY_ID: targets, "title": self.name, "message": self.message(decision)}, 

142 blocking=True, 

143 context=context, 

144 ) 

145 except HomeAssistantError: 

146 _LOGGER.exception("CAMERA_AUTONOTIFY Notification failed for %s", self.name)