Coverage for apps/notifier.py: 100%
142 statements
« prev ^ index » next coverage.py v7.4.3, created at 2024-03-12 12:23 +0000
1import math
3import appdaemon.plugins.hass.hassapi as hass
5"""
6Source : https://github.com/jlpouffier/home-assistant-config/blob/master/appdaemon/apps/notifier.py
8Notify is an app responsible for notifying the right occupant(s) at the right time, and making sure to discard notifications once they are not relevant anymore.
10Here is the list of parameter that you NEED to set in order to run the app:
12home_occupancy_sensor_id: the id of a binary sensor that will be true if someone is at home, and false otherwise
13proximity_threshold: A thresold in meter, bellow this threshold the app will concider the person at home. This is to avoid pinging only the first one that reaches Home if both occupant are on the same car for example (Both will be pinged)
14persons: A list of person, including
15 name: their name
16 id: the id of the person entity in home assistant
17 notification_service: the name of the notification service used to ping the phone of this person.
18 proximity_id: the id of the proximity entity linked to this person
21Here is an exmaple on how to instantiate the app:
23notifier:
24 module: notifier
25 class: Notifier
26 home_occupancy_sensor_id: binary_sensor.home_occupied
27 proximity_threshold: 1000
28 persons:
29 - name: jl
30 id: person.jenova70
31 notification_service: notify/mobile_app_pixel_6
32 proximity_id: proximity.distance_jl_home
33 - name: valentine
34 id: person.valentine
35 notification_service: notify/mobile_app_pixel_4a
36 proximity_id: proximity.distance_valentine_home
38The complete app can be called from anywhere by sending a custom event NOTIFIER with the following schema:
40action: <string>
41title: <string>
42message: <string>
43callback:
44 - title: <string>
45 event: <string>
46 destructive: <boolean>
47 icon: <string>
48 - title: <string>
49 event: <string>
50timeout: <number>
51image_url: <url>
52click_url: <url>
53icon: <string>
54color: <string>
55tag: <string>
56persistent: <boolean>
57interuption_level: <string>
58until:
59 - entity_id: <string>
60 new_state: <string>
61 - entity_id: <string>
62 new_state: <string>
64Here are detailed explanations for each field: (fields with a star * are mandatory)
66action can be the following:
67- send_to_<person_name>: Send a notification directly to the person called <person_name>
68- send_to_all: Send to all
69- send_to_present: Send a notification directly to all present occupants of the home
70- send_to_absent: Send a notification directly to all absent occupants of the home
71- send_to_nearest: Send a notification to the nearest occupant(s) of the home
72- send_when_present:
73 - if the home is occupied: Send a notification directly to all present occupant of the home
74 - if the home is empty: Stage the notification and send it once the home becomes occupied
76*title: Title of the notification
78*message: Body of the notification
80callback: Actionable buttons of the notification
81 - title: Title of the button
82 - event: a string that will be used the catch back the event when the button is pressed.
83 If event: turn_off_lights, then an event "mobile_app_notification_action" with action = "turn_off_lights" will be triggered once the button is pressed.
84 Up to the app / automation creating the notification to listen to this event and perform some action.
85 - destructive: {iOS Only} Set it to true to color the action's title red, indicating a destructive action.
86 - icon: {iOS Only} The icon to use for the callback.
88timeout: Timeout of the notification in seconds. timeout: 60 will display the notification for one minute, then discard it automatically.
90image_url: url of an image that will be embedded on the notification. Useful for cameras, vacuum maps, etc.
92click_url: url of the target location if the notification is pressed.
93If you have a lovelace view called "/lovelace/vacuums" for your vacuum, then putting click_url: "/lovelace/vacuums" will lead to this view if the notification is clicked
95icon: Icon of the notification. format mdi:<string>. Visit https://materialdesignicons.com/ for supported icons
97color: color of the notification.
98Format can be "red" or "#ff6e07"
100tag: The concept of tag is complex to understand. So I'll explain the behavior you will experience while using tags.
101 - A subsequent notification with a tag will replace an old notification with the same tag.
102 For example if you want to notify that a vacuum is starting, and then finishing: use the same tag for both (like "vacuum") and the "Cleaning complete" notification will replace the "Cleaning started" notification, as it is not relevant anymore.
103 - If you notify more than one person with the same tag:
104 - Acting on the notification (a button) on a device will discard it on every other devices
105 Example: If you notify all occupants that the lights are still on while the home is empty with an actionable button to turn off the lights, if person A clicks on "Turn off lights" then person B will see the notification disappear... Because it's not relevant anymore (it's done)
106 - The next field "until" requires the field tag to work too (See below)
108persistent: Notify the front-end of Home Assistant (with the service "notify/persistent_notification")
110interuption_level: {iOS Only} interruption level of a notification (passive/active/time-sensitive/critical)
112siri_shortcut_name: {iOS Only} Name of the shortcut to run when clicking on the notification
114until (note: "tag" is required for "until" to work)
115until dynamically creates watcher(s) to clear notification.
116I prefer to explain it with an example:
117If you want to notify all occupants that the lights are still on while the home is empty, you can specify
118until:
119 - entity_id: binary_sensor.home_occupied
120 new_state : on
121 - entity_id: light.all_lights
122 new_state : off
123This will make the notification(s) disappear as soon as the lights are off, or the home becomes occupied.
124That way, you make sure notifications are only displayed when relevant.
126"""
129class Notifier(hass.Hass):
130 def initialize(self):
131 # Listen to all NOTIFIER events
132 self.listen_event(self.callback_notifier_event_received, "NOTIFIER")
133 self.listen_event(self.callback_notifier_discard_event_received, "NOTIFIER_DISCARD")
134 self.listen_event(self.callback_button_clicked, "mobile_app_notification_action")
136 # Staged notification
137 self.staged_notifications = []
138 self.listen_state(self.callback_home_occupied, self.args["home_occupancy_sensor_id"], old="off", new="on")
140 # Temporary watchers
141 self.watchers_handles = []
143 def callback_notifier_event_received(self, event_name, data, kwargs):
144 self.log("NOTIFIER event received")
145 if "action" in data:
146 action = data["action"]
147 match action:
148 case "send_to_all":
149 # send_to_all
150 self.send_to_all(data)
151 case "send_to_present":
152 # send_to_present
153 self.send_to_present(data)
154 case "send_to_absent":
155 # send_to_absent
156 self.send_to_absent(data)
157 case "send_to_nearest":
158 # send_to_nearest
159 self.send_to_nearest(data)
160 case "send_when_present":
161 # send_when_present
162 self.send_when_present(data)
163 case _:
164 for person in self.args["persons"]:
165 if action == "send_to_" + person["name"]:
166 self.send_to_person(data, person)
168 if "persistent" in data:
169 if data["persistent"]:
170 if "tag" in data:
171 notification_id = data["tag"]
172 else:
173 notification_id = str(self.get_now_ts())
174 self.log("Persisting the notification on Home Assistant Front-end ...")
175 self.call_service(
176 "persistent_notification/create",
177 title=data["title"],
178 message=data["message"],
179 notification_id=notification_id,
180 )
182 if "until" in data and "tag" in data:
183 until = data["until"]
184 for watcher in until:
185 watcher_handle = {}
186 if "new_state" in watcher and "old_state" in watcher:
187 watcher_handle["id"] = self.listen_state(
188 self.callback_until_watcher,
189 watcher["entity_id"],
190 new=str(watcher["new_state"]),
191 old=str(watcher["old_state"]),
192 oneshot=True,
193 tag=data["tag"],
194 )
195 transition = f"from {str(watcher['old_state'])} to {str(watcher['new_state'])}"
196 elif "new_state" in watcher:
197 watcher_handle["id"] = self.listen_state(
198 self.callback_until_watcher,
199 watcher["entity_id"],
200 new=str(watcher["new_state"]),
201 oneshot=True,
202 tag=data["tag"],
203 )
204 transition = f"to {str(watcher['new_state'])}"
205 elif "old_state" in watcher:
206 watcher_handle["id"] = self.listen_state(
207 self.callback_until_watcher,
208 watcher["entity_id"],
209 old=str(watcher["old_state"]),
210 oneshot=True,
211 tag=data["tag"],
212 )
213 transition = f"from {str(watcher['old_state'])}"
214 watcher_handle["tag"] = data["tag"]
215 self.watchers_handles.append(watcher_handle)
216 self.log(
217 f"All notifications with tag {data['tag']} will be cleared "
218 f"if {watcher['entity_id']} transitions {transition}"
219 )
221 def callback_notifier_discard_event_received(self, event_name, data, kwargs):
222 self.clear_notifications(data["tag"])
224 def callback_until_watcher(self, entity, attribute, old, new, kwargs):
225 self.clear_notifications(kwargs["tag"])
227 def callback_button_clicked(self, event_name, data, kwargs):
228 if "tag" in data:
229 self.clear_notifications(data["tag"])
231 def clear_notifications(self, tag):
232 self.log("Clearing notifications with tag " + tag + " (if any) ...")
233 notification_data = {}
234 notification_data["tag"] = tag
235 for person in self.args["persons"]:
236 self.call_service(person["notification_service"], message="clear_notification", data=notification_data)
237 self.call_service("persistent_notification/dismiss", notification_id=tag)
238 self.cancel_watchers(tag)
240 def cancel_watchers(self, tag):
241 self.log("Removing watchers with tag " + tag + " (if any) ...")
242 for watcher in list(self.watchers_handles):
243 if watcher["tag"] == tag:
244 self.watchers_handles.remove(watcher)
246 def build_notification_data(self, data):
247 notification_data = {}
248 if "callback" in data:
249 notification_data["actions"] = []
250 for callback in data["callback"]:
251 action = {"action": callback["event"], "title": callback["title"]}
252 if "icon" in callback:
253 action["icon"] = "sfsymbols:" + callback["icon"]
254 if "destructive" in callback:
255 action["destructive"] = callback["destructive"]
256 notification_data["actions"].append(action)
257 if "timeout" in data:
258 notification_data["timeout"] = data["timeout"]
259 if "click_url" in data:
260 notification_data["url"] = data["click_url"]
261 if "image_url" in data:
262 notification_data["image"] = data["image_url"]
263 if "icon" in data:
264 notification_data["notification_icon"] = data["icon"]
265 if "color" in data:
266 notification_data["color"] = self.compute_color(data["color"])
267 if "tag" in data:
268 notification_data["tag"] = data["tag"]
269 if "interuption_level" in data:
270 notification_data["push"] = {"interruption-level": data["interuption_level"]}
271 if "siri_shortcut_name" in data:
272 notification_data["shortcut"] = {"name": data["siri_shortcut_name"]}
273 return notification_data
275 def send_to_person(self, data, person):
276 self.log("Sending notification to " + person["name"])
277 notification_data = self.build_notification_data(data)
278 self.call_service(
279 person["notification_service"], title=data["title"], message=data["message"], data=notification_data
280 )
282 def send_to_all(self, data):
283 for person in self.args["persons"]:
284 self.send_to_person(data, person)
286 def send_to_present(self, data):
287 for person in self.args["persons"]:
288 if (
289 self.get_state(person["id"]) == "home"
290 or float(self.get_state(person["proximity_id"])) <= self.args["proximity_threshold"]
291 ):
292 self.send_to_person(data, person)
294 def send_to_absent(self, data):
295 for person in self.args["persons"]:
296 if (
297 self.get_state(person["id"]) != "home"
298 or float(self.get_state(person["proximity_id"])) > self.args["proximity_threshold"]
299 ):
300 self.send_to_person(data, person)
302 def send_to_nearest(self, data):
303 min_proximity = float(self.get_state(self.args["persons"][0]["proximity_id"]))
304 for person in self.args["persons"]:
305 person_proximity = float(self.get_state(person["proximity_id"]))
306 if person_proximity <= min_proximity:
307 min_proximity = person_proximity
309 for person in self.args["persons"]:
310 person_proximity = float(self.get_state(person["proximity_id"]))
311 if person_proximity <= min_proximity + self.args["proximity_threshold"]:
312 self.send_to_person(data, person)
314 def send_when_present(self, data):
315 if self.get_state(self.args["home_occupancy_sensor_id"]) == "on":
316 self.send_to_present(data)
317 else:
318 self.log("Staging notification for when home becomes occupied ...")
319 self.staged_notifications.append(data)
321 def callback_home_occupied(self, entity, attribute, old, new, kwargs):
322 if len(self.staged_notifications) >= 1:
323 self.log("Home is occupied ... Sending stagged notifications now ...")
324 while len(self.staged_notifications) >= 1:
325 current_data = self.staged_notifications.pop(0)
326 self.send_to_present(current_data)
328 def compute_color(self, color_name):
329 colors = {
330 "red": "#f44336",
331 "pink": "#e91e63",
332 "purple": "#9c27b0",
333 "deep-purple": "#673ab7",
334 "indigo": "#3f51b5",
335 "blue": "#2196f3",
336 "light-blue": "#03a9f4",
337 "cyan": "#00bcd4",
338 "teal": "#009688",
339 "green": "#4caf50",
340 "light-green": "#8bc34a",
341 "lime": "#cddc39",
342 "yellow": "#ffeb3b",
343 "amber": "#ffc107",
344 "orange": "#ff9800",
345 "deep-orange": "#ff5722",
346 "brown": "#795548",
347 "grey": "#9e9e9e",
348 "blue-grey": "#607d8b",
349 "black": "#000000",
350 "white": "#ffffff",
351 "disabled": "#bdbdbd",
352 }
353 if color_name in colors:
354 return colors[color_name]
355 else:
356 return color_name