J'ai tenté une évolution du module notifier pour intégrer la logique de groupe. Au sein d'une famille cela permet de gérer des notifications différentes pour les parents ou pour les enfants.
Dans le fichier apps.yaml pour chaque personne je lui déclare 1 ou plusieurs groupes.
Ensuite au niveau de l'appel dans HA j'ajoute le mot clé group.
Et j'ai ajouté une fonction send_to_group_
il faudrait tester les cas absents et when present, je n'ai pas fini les tests
import hassapi as hass
import math
class notifier(hass.Hass):
def initialize(self):
# Listen to all NOTIFIER events
self.listen_event(self.callback_notifier_event_received , "NOTIFIER")
self.listen_event(self.callback_notifier_discard_event_received , "NOTIFIER_DISCARD")
self.listen_event(self.callback_button_clicked, "mobile_app_notification_action")
# Staged notification
self.staged_notifications = []
self.listen_state(self.callback_home_occupied , self.args["home_occupancy_sensor_id"] , old = "off" , new = "on")
# Temporary watchers
self.watchers_handles = []
# Créer un dictionnaire pour un accès rapide aux personnes
self.persons_dict = {person["name"]: person for person in self.args["persons"]}
# Chargement des groupes
self.groups = {person['name']: person.get('group', []) for person in self.args['persons']}
def callback_notifier_event_received(self, event_name, data, kwargs):
self.log("NOTIFIER event received")
if "action" in data:
action = data["action"]
group = data.get("group", None)
if action.startswith("send_to_group_"):
group_name = action[len("send_to_group_"):]
self.send_to_group(data, group_name)
elif action == "send_to_all":
self.send_to_all(data)
elif action in ["send_to_present", "send_to_absent", "send_to_nearest", "send_when_present"]:
getattr(self, action)(data, group)
elif action.startswith("send_to_"):
person_name = action[len("send_to_"):]
person = self.persons_dict.get(person_name)
if person:
self.send_to_person(data, person)
else:
# Gérer les cas où `action` ne correspond à aucun des cas traités
self.log(f"Action non reconnue : {action}", level="WARNING")
if "persistent" in data:
if data["persistent"]:
if 'tag' in data:
notification_id = data['tag']
else:
notification_id = str(self.get_now_ts())
self.log("Persisting the notification on Home Assistant Front-end ...")
self.call_service("persistent_notification/create", title = data["title"], message = data["message"], notification_id = notification_id)
if "until" in data and 'tag' in data:
until = data["until"]
for watcher in until:
watcher_handle = {}
watcher_handle["id"] = self.listen_state(self.callback_until_watcher, watcher["entity_id"], new = str(watcher["new_state"]), oneshot = True, tag = data["tag"])
watcher_handle["tag"] = data["tag"]
self.watchers_handles.append(watcher_handle)
self.log("All notifications with tag " + data["tag"] + " will be cleared if " + watcher["entity_id"] + " transitions to " + str(watcher["new_state"]))
def callback_notifier_discard_event_received(self, event_name, data, kwargs):
self.clear_notifications(data["tag"])
def callback_until_watcher(self, entity, attribute, old, new, kwargs):
self.clear_notifications(kwargs["tag"])
def callback_button_clicked(self, event_name, data, kwargs):
if "tag" in data:
self.clear_notifications(data["tag"])
def clear_notifications(self, tag):
self.log("Clearing notifications with tag " + tag + " (if any) ...")
notification_data = {}
notification_data["tag"] = tag
for person in self.args["persons"]:
self.call_service(person["notification_service"], message = "clear_notification", data = notification_data)
self.call_service("persistent_notification/dismiss", notification_id = tag)
self.cancel_watchers(tag)
def cancel_watchers(self, tag):
self.log("Removing watchers with tag " + tag + " (if any) ...")
for watcher in list(self.watchers_handles):
if watcher["tag"] == tag:
self.watchers_handles.remove(watcher)
def build_notification_data(self, data):
notification_data = {}
if "callback" in data:
notification_data["actions"] = []
for callback in data["callback"]:
action = {
"action":callback["event"],
"title":callback["title"]
}
if "icon" in callback:
action["icon"] = "sfsymbols:" + callback["icon"]
if "destructive" in callback:
action["destructive"] = callback["destructive"]
notification_data["actions"].append(action)
if "timeout" in data:
notification_data["timeout"] = data["timeout"]
if "click_url" in data:
notification_data["url"] = data["click_url"]
if "image_url" in data:
notification_data["image"] = data["image_url"]
if "icon" in data:
notification_data["notification_icon"] = data["icon"]
if "color" in data:
notification_data["color"] = self.compute_color(data["color"])
if "tag" in data:
notification_data["tag"] = data["tag"]
if "interuption_level" in data:
notification_data["push"] = {
"interruption-level": data["interuption_level"]
}
if "siri_shortcut_name" in data:
notification_data["shortcut"] = {
"name": data["siri_shortcut_name"]
}
return notification_data
def send_to_person(self, data, person):
notification_data = self.build_notification_data(data)
self.call_service(person["notification_service"], title = data["title"], message = data["message"], data = notification_data)
self.log("Sending notification to " + person["name"])
def send_to_group(self, data, group_name):
self.log(f"Sending notification to group {group_name}")
group_members = self.filtered_persons(group_name)
#self.log(f"Members in group {group_name}: {group_members}")
for person in group_members:
#self.log(f"Attempting to send notification to {person['name']} in group {group_name}")
self.send_to_person(data, person)
def send_to_all(self, data):
self.log("Sending notification to all")
for person in self.args["persons"]:
self.send_to_person(data, person)
def send_to_present(self, data, group):
self.log("Sending notification to present")
target_persons = self.filtered_persons(group)
for person in target_persons:
if self.get_state(person["id"]) == "home" or float(self.get_state(person["distance_sensor"])) <= self.args["proximity_threshold"]:
self.send_to_person(data, person)
def send_to_absent(self, data, group):
self.log("Sending notification to absent")
target_persons = self.filtered_persons(group)
for person in target_persons:
if self.get_state(person["id"]) != "home" or float(self.get_state(person["distance_sensor"])) > self.args["proximity_threshold"]:
self.send_to_person(data, person)
def send_to_nearest(self, data, group):
self.log("Sending notification to nearest")
target_persons = self.filtered_persons(group)
min_proximity = float("inf")
for person in target_persons:
person_proximity = float(self.get_state(person["distance_sensor"]))
if person_proximity <= min_proximity:
min_proximity = person_proximity
for person in target_persons:
person_proximity = float(self.get_state(person["distance_sensor"]))
if person_proximity <= min_proximity + self.args["proximity_threshold"]:
self.send_to_person(data, person)
def send_when_present(self, data, group):
self.log("Sending notification when present")
if self.get_state(self.args["home_occupancy_sensor_id"]) == "on":
self.send_to_present(data, group)
else:
self.log("Staging notification for when home becomes occupied ...")
self.staged_notifications.append(data)
def filtered_persons(self, group_name):
"""
Renvoie une liste des personnes appartenant au groupe spécifié.
Si aucun groupe n'est spécifié (group_name est None), renvoie toutes les personnes.
"""
if group_name is None:
return self.args["persons"]
return [person for person in self.args["persons"] if group_name in person.get('group', [])]
def callback_home_occupied(self, entity, attribute, old, new, kwargs):
if len(self.staged_notifications) >= 1:
self.log("Home is occupied ... Sending stagged notifications now ...")
while len(self.staged_notifications) >= 1:
current_data = self.staged_notifications.pop(0)
self.send_to_present(current_data)
def compute_color(self, color_name):
colors = {
"red":"#f44336",
"pink":"#e91e63",
"purple":"#9c27b0",
"deep-purple":"#673ab7",
"indigo":"#3f51b5",
"blue":"#2196f3",
"light-blue":"#03a9f4",
"cyan":"#00bcd4",
"teal":"#009688",
"green":"#4caf50",
"light-green":"#8bc34a",
"lime":"#cddc39",
"yellow":"#ffeb3b",
"amber":"#ffc107",
"orange":"#ff9800",
"deep-orange":"#ff5722",
"brown":"#795548",
"grey":"#9e9e9e",
"blue-grey":"#607d8b",
"black":"#000000",
"white":"#ffffff",
"disabled":"#bdbdbd"
}
if color_name in colors:
return colors[color_name]
else:
return color_name
J'ai tenté une évolution du module notifier pour intégrer la logique de groupe. Au sein d'une famille cela permet de gérer des notifications différentes pour les parents ou pour les enfants.
Dans le fichier apps.yaml pour chaque personne je lui déclare 1 ou plusieurs groupes.
Ensuite au niveau de l'appel dans HA j'ajoute le mot clé group.
Et j'ai ajouté une fonction send_to_group_
il faudrait tester les cas absents et when present, je n'ai pas fini les tests