HA-Addon APPDaemon-Programme Python

Intro

Dans mes publications, je fais de plus en plus référence à des applications écrites en python dans l’addon « Appdaemon ».

Il me semblait utile de détailler simplement son installation dans un article.

Cet Addon, pour ceux qui ne connaissent pas, vient compléter HA en permettant d’écrire et utiliser des programmes en Python très élaborés, voir impossibles à réaliser avec l’automatisme de base de HA.

Il permet également de développer un tableau de bord, mais ce n’est pas son point fort, j’ai testé mais je ne suis pas fan de son utilisation, j’ai préféré utilisé le « Dwain lovelace », voir mon article sur cette excellente intégration HACS.

Mise à jour

Du 1 juin 2022:

  • Ajout de l’application « Alerte meteo »

Installation

l’installation de l’Addon est simple en tout cas avec HASS, il suffit de le chercher dans la boutique des modules complémentaires et de l’installer. Pour les autres configurations, il faut se reporter à la documentation de l’addon.

Classiquement dans un Addon, vous avez accès aux informations, à la documentation, la configuration et le journal.

Dans un premier temps vous laissez la configuration par défaut, vous pourrez toujours y revenir par la suite. Veuillez noter le port 5050 qui vous permettra d’accéder à l’UI de AppDaeamon.

Si tout va bien vous devez visualiser un journal qui ressemble à minima celui-ci:

Configuration

Après installation, nous devons configurer AppDaemon, pour cela vous trouverez un dossier « /config/appdaemon » contenant plusieurs sous-dossiers:

AppDaemon.yaml

Le premier fichier qui nous interresse est le « /config/www/appdaemon.yaml », il est installé par défaut avec l’Addon, mais il nécessite d’être personnalisé.

Commençons par les coordonnées géographiques permettent de déterminer la date et l’heure locale, vous pouvez utiliser le fichier « secrets.yaml » déclaré dans votre HA.

Ensuite nous redirigeons les journaux dans un dossier spécifique « /config/log/.. » précédemment créé. Par défaut les « logs » sont envoyés dans le « main_log » de l’addon.

Pour ma part je créé un fichier spécifique à chacune de mes applications, c’est plus facile de s’y retrouver, par exemple:

  • piscine_log: #Nom du log que vous utilisez dans le programme python
    • name: PiscineLog # Nom obligatoire créé par l’utilisateur
    • filename: /config/log/piscine.log # chemin et nom des fichiers logs
    • log_generations: 3 #nombre de fichier logs maximum (les anciens sont écrasés)
    • log_size: 100000 # Taille des fichiers logs quasiment égales au nombre de caractères(Par défaut ils font 1 000 0000 de caractères)
    • Vous pouvez aussi personnaliser le format, se reporter à la documentation de l’addon.

La dernière version de mon fichier « appdaemon.yaml » est téléchargeable ici.

---
secrets: /config/secrets.yaml
appdaemon:
  latitude: !secret latitude
  longitude: !secret longitude
  elevation: 220
  time_zone: Europe/Paris
  plugins:
    HASS:
      type: hass
http:
  url: http://192.168.0.37:5050
admin:
api:
hadashboard:

logs:
  main_log:
    filename: /config/log/appdaemon.log
    
  access_log:
    filename: /config/log/access.log

  error_log:
    filename: /config/log/error.log

  diag_log:
    filename: /config/log/diag.log
    format: "{asctime} {levelname:<8} {appname:<10}: {message}"

  piscine_log:
    name: PiscineLog
    filename: /config/log/piscine.log
    log_generations: 3
    log_size: 100000

  linky_log:
    name: LinkyLog
    filename: /config/log/linky.log
    log_generations: 3
    log_size: 100000

  surveille_log:
    name: SurveilleLog
    filename: /config/log/surveille_log.log
    log_generations: 3
    log_size: 1024

  groupealerte_log:
    name: GroupeAlertLog
    filename: /config/log/groupealerte_log.log
    log_generations: 3
    log_size: 1024

  test_log:
    name: TestLog
    filename: /config/log/test_log.log
    log_generations: 3
    log_size: 1024

Il ne vous reste qu’a personnaliser vos log si vous le souhaitez.

Nota: Après chaque modification du fichier « .yaml », il faut redémarrer l’addon depuis son interface:

Code des fichiers python

Nos fichiers d’applications écrits en python sont stockés dans le fichier « /config/appdaemon/apps »

Chaque application comprends deux fichiers:

  • Fichier  » .yaml » dans lequel nous déclarons les appels vers le fichier python « .py » associé et les entités HA utilisées dans le programme.
  • Fichier « .py« : C’est le fichier programme python principal déclaré dans le . »yaml. » Je ne vais pas en détailler le contenu car d’une part la documentation de l’addon est très complète et d’autre part il existe de nombreux sites expliquant la programmation en python.

Par défaut, un seul fichier « .yaml » peut regrouper les déclarations de toutes les applications « .py », mais afin d’améliorer la visibilité, je préfère le séparer par application.

Après modification d’un de ces fichiers, toutes les applications sont automatiquement réinitialisées.

Exemple d’applications AppDaemon:

Surveille l’activité d’une entité

Cette application permet de « surveiller » le changement d’état d’une variable dans un temps donné, une fois le temps écoulé, si la variable n’a pas évolué, une notification et une action sont déclenchées.

Fichier appdaemon.yaml

Ajout du log correspondant:

  surveille_log:
    name: SurveilleLog
    filename: /config/log/surveille_log.log
    log_generations: 3
    log_size: 100000

Fichier « Surveille_activite.yaml »

La configuration minimale est:

  • surveille_com_linky: # nom du module que vous retrouvez dans AppDaemon
    • module: surveille_activite  » Nom du fichier « .py » associé
    • class: SurveilleActivite # Class à déclarer dans le fichier « .py »
    • entite: sensor.linky_papp # L’entité HA surveillée
    • tempo: 3600 # Tempo de surveillance en secondes
    • activite_ok: input_boolean.com_linky # l’entité HA actionnée si dépassement

En quelques lignes, vous pouvez faire appel plusieurs fois à une même instance, dans mon cas je surveille les communications entre HA et mon compteur Linky et ma centrale Météo.

La dernière version du fichier est téléchargeable ici.

surveille_com_linky:
  module: surveille_activite
  class: SurveilleActivite
  entite: sensor.linky_umoy #sensor.linky_i_inst # sensor.linky_papp #sensor.linky_umoy #
  #entite: input_text.linky_test_status
  tempo: 3600 # Tempo de surveillance en secondes
  activite_ok: input_boolean.com_linky #switch.test_com_out

surveille_com_vp2:
  module: surveille_activite
  class: SurveilleActivite
  entite: sensor.vp2_datetime
  tempo: 360 # Tempo de surveillance en secondes
  activite_ok: input_boolean.com_vp2

surveille_alim_tablette:
  module: surveille_activite
  class: SurveilleActivite
  entite: sensor.tablette_niveau_charge
  tempo: 600 # Tempo de surveillance en secondes
  activite_ok: input_boolean.alim_tablette_ok  

Fichier « Surveille_activite.py »

Dans les grandes lignes, vous avez:

  • les class python à importer: à minima « import hassapi as hass »
  • la class déclarée dans le fichier « .yaml »
  • Une fonction initialisation obligatoire
  • Une fonction appelée sur changement d’état
  • Une fonction de notification

La dernière version du fichier est téléchargeable ici.

import hassapi as hass

class SurveilleActivite(hass.Hass):
    def initialize(self):
        self.listen_state(self.change,self.args["entite"])
        nom_entité = self.get_state(self.args["entite"],attribute="entity_id")
        duree_tempo=int(self.args["tempo"])
        self.turn_on(self.args["activite_ok"])
        self.tempo=self.run_in(self.notification, duree_tempo,entité=nom_entité,temps=duree_tempo)
        self.log(f'Initialisation de: {nom_entité} pour {duree_tempo}s et {self.tempo}.', log="surveille_log")

    def change(self, entity, attribute, old, new, kwargs):
        heure = str(self.time())[:8]
        duree_tempo=int(self.args["tempo"])
        tempo_on = str(entity)
        nom_entité = str(entity)
        nouvelle_valeur = new
        # Mise à on de Com_Ok
        self.turn_on(self.args["activite_ok"])
        self.log(f'Nouvelle valeur de {entity}: {nouvelle_valeur}-Tempo={duree_tempo}', log="surveille_log")
        cle_tempo = self.tempo
        if cle_tempo != None:
            self.tempo = self.cancel_timer(cle_tempo) 
            #self.log(f'Info tempo: {self.info_timer(cle_tempo)}', log="surveille_log")
            #self.log(f'Fin tempo {cle_tempo}', log="surveille_log")

        self.tempo = self.run_in(self.notification, duree_tempo,entité=nom_entité,temps=duree_tempo)

    def notification(self, kwargs):
        heure = str(self.time())[:8]
        nom_entité = kwargs["entité"]
        duree_temps= kwargs["temps"]
        # Mise à off de Com_Ok
        self.turn_off(self.args["activite_ok"])
        self.log(f'Alerte! {nom_entité} est out depuis {duree_temps} sec.', log="surveille_log")
        message_notification=format(heure)+"Alerte!"+ format(nom_entité)+"est out depuis: "+format(duree_temps)+" sec."
        self.call_service('notify/telegram', message_notification)
        self.call_service('dwains_dashboard/notification_create', message=message_notification)

Surveille le délai de validité d’un certificat Lets

L’intégration « Expiration du certificat » permet de surveiller l’état d’un certificat « Lest’enscript ». Les miens sont gérés par mon NAS OpenMediaVault.

Le but de cette application « AppDaemon » est de notifier que le délai du certificat expire dans un nombre de jours paramétrables ou si le certificat est devenu invalide.

La notification sera relancée chaque jour à 00:00 tant le certificat n’aura pas été renouvelé.

Le « state » indique la date-time à laquelle le certificat expire, l’attribut « is_valid » indique si le certificat est encore valide.

Fichier appdaemon.yaml

Ajout du log correspondant:

  fin_certificats_log:
    name: fincertificats
    filename: /config/log/fin_certificats_log.log
    log_generations: 3
    log_size: 100000

Fichier « alerte_fin_certificats.yaml »

Déclaration des entités à surveiller et du seuil bas en nombre de Jours en dessous duquel la notification est activée.

La dernière version du fichier est téléchargeable ici:

alert_fin_certificats:
  class: AlerteFinCertificats
  module: alerte_fin_certificats
  certif: sensor.cert_expiry_timestamp_domo_rem81_com,sensor.cert_expiry_timestamp_vtt_rem81_com,sensor.cert_expiry_timestamp_meteo_rem81_com,sensor.cert_expiry_timestamp_motioneye_rem81_com,sensor.cert_expiry_timestamp_kobold_rem81_com,sensor.cert_expiry_timestamp_meteo81000_rem81_com,sensor.cert_expiry_timestamp_ha_rem81_com,sensor.cert_expiry_timestamp_nextcloud_rem81_com
  seuil_bas: 25 # En Jours

Fichier « alerte_fin_certificats.py »

Le programme python correspondant:

La dernière version du fichier est téléchargeable ici:

import hassapi as hass
import datetime
from datetime import datetime
from datetime import timedelta
import time

#  Niveau de JOURNALISATION (log): 0=rien ou 1 =info ou 2=debug 
JOURNAL=2 

class AlerteFinCertificats(hass.Hass):
    def initialize(self):
        if "certif" in self.args:
            for certif in self.split_device_list(self.args["certif"]):
                self.notification('Surveillance de:'+certif,2,"")
        tempo_j = self.run_daily(self.bilan_jour, "00:05:00")
        self.bilan_jour(self)
        #tempo_j = self.run_every(self.bilan_jour, "now", 1 * 60)
        self.log("Initialisation Alert_fin_certificat.py", log="fin_certificats_log")
        self.log("Initialisation Alert_fin_certificat.py", log="error_log")
            
    def bilan_jour(self,kwargs):
        s_bas= int(self.args["seuil_bas"])
        tousvalides=1 # Indicateur que tous les certificats sont valables
        for certif in self.split_device_list(self.args["certif"]):
            self.notification('Lecture de:'+certif,2,"")
            nom_entité =  self.friendly_name(certif)
            
            etat = self.get_state(certif, attribute="state")
            validité = self.get_state(certif,attribute="is_valid")
            self.notification("Friendly_name= "+nom_entité,2,"")
            self.notification("Etat= "+etat,2,"")
            # Vérifie si le certificat est valide
            if etat !="unknown" and validité == True:
                # Création entités dans HA
                binarysensorname="binary_sensor.cert_"+certif[29:]+"_validite"  #binary_sensor.certificat_ha
                #binarysensorname="binary_sensor.cert_"+nom_entité+"_validite"  #binary_sensor.certificat_ha
                self.set_state(binarysensorname, state="on", replace=True, attributes= {"icon": "mdi:check","device_class": "connectivity"})
                self.notification("Binary_SensorName:" + binarysensorname,2,"")
                ce_jour=datetime.strptime(time.strftime('%Y:%m:%d', time.localtime()),'%Y:%m:%d')
                self.notification("ce Jour:" + str(ce_jour),2,"")
                date_de_fin=datetime.strptime(etat[:10],'%Y-%m-%d')
                self.notification("Date de Fin:" + str(date_de_fin),2,"")
                if date_de_fin<ce_jour:
                    nb_jour=(date_de_fin-ce_jour).days
                    self.notification("nb Jour negatifs:" + str(nb_jour),2,"")
                else:
                    nb_jour=(date_de_fin-ce_jour).days
                    self.notification("nb Jour:" + str(nb_jour),2,"")
                    sensorname="sensor.cert_"+certif[29:]+"_fin"
                    #sensorname="sensor.cert_"+nom_entité+"_fin"
                    self.notification("SensorName:" + sensorname,2,"")

                    # Vérifie si le nombre de jours est inférieur au seuil bas
                    if nb_jour < s_bas:
                        # Mise à jour entités HA
                        message_notification= "Attention: Fin du certificat <"+ format(nom_entité)+"> dans "+ format(nb_jour)+" J."
                        self.set_state(sensorname, state=nb_jour, replace=True, attributes= {"icon": "mdi:alert-octagram", "unit_of_measurement": "J"})
                        self.notification(message_notification,0,"teleg")
                    else:
                        # Mise à jour entités HA
                        message_notification= "Le certificat <"+ format(nom_entité)+"> est encore valable "+ format(nb_jour)+" J."
                        self.set_state(sensorname, state=nb_jour, replace=True, attributes= {"icon": "mdi:check", "unit_of_measurement": "J"})
                        self.notification(message_notification,2,"")
            else:
                tousvalides=0 # Indicateur qu'au moins un certificat n'est plus valable
                # Mise à jour entité HA
                binarysensorname="binary_sensor.cert_"+nom_entité+"_validite"
                self.set_state(binarysensorname, state="off", replace=True, attributes= {"icon": "mdi:alert-octagram","device_class": "connectivity"})
                message_notification= " Attention: Le certificat <"+ format(nom_entité)+"> n'est plus valide."
                self.notification(message_notification,0,"teleg")

        #  mise à jour dans HA de l'indicateur synthèse que les certificats sont valables
        if tousvalides==1:
            self.set_state("binary_sensor.certificat_tous_valides",state="on", replace=True, attributes= {"icon": "mdi:check","device_class": "connectivity"})
        else:
            self.set_state("binary_sensor.certificat_tous_valides",state="off", replace=True, attributes= {"icon": "mdi:alert-octagram","device_class": "connectivity"})



                
    # Fonction Notification
    # message =  Texte à afficher
    # niveau = niveau de journalisation 0,1,2
    # si notif == "teleg" on notifie aussi sur Télégram
    def notification(self,texte_message,niveau,notif):
        global JOURNAL
        heure = str(self.time())[:8]
        if niveau <= JOURNAL:
            message_notification= format(heure)+": "+ texte_message
            self.log(message_notification, log="fin_certificats_log")
            if notif=="teleg":
                self.call_service('notify/telegram', message=message_notification)
                self.call_service('persistent_notification/create', message=message_notification)
                self.call_service('dwains_dashboard/notification_create', message=message_notification)
    

Affichage du résultat

Alerte météo

Ce programme notifie le début et la fin de alertes météo délivrée par Meteo France ainsi que l’arrivée de la pluie dans la prochaine heure. En pré requis il faut déclarer l’intégration « meteo france ».

Dans le menu « parametre/intégration » cliquer en bas à droite « +ajouter integration »

Chercher « méteo-france ».

Fichier « alerte_meteo.yaml »

La dernière version du fichier est téléchargeable ici:

Personnaliser « entité » et « alerte » avec ceux de votre intégration « meteo-france »

# Version du 1 juin 2022
alerte_meteo:
  class: AlerteMeteo
  module: alerte_meteo
  entité: sensor.albi_next_rain #h_arrivee_pluie
  alerte: sensor.81_weather_alert
  

Fichier « alerte_meteo.py »

La dernière version du fichier est téléchargeable ici.

Vous pouvez définir le niveau de journalisation « Niveau de JOURNALISATION (log): 0=aucune notif ou 1 =info ou 2=debug »

De même, vous devez adapter vos services de notification dans les trois dernières lignes du fichier.

# Version du 1 juin 2022
import hassapi as hass
import datetime
from datetime import datetime
from datetime import timedelta
import time

# Niveau de JOURNALISATION (log): 0=aucune notif ou 1 =info ou 2=debug 
JOURNAL=2 
h_forcast={}
FLAG=0

class AlerteMeteo(hass.Hass):
    def initialize(self):
        self.notification("Initialisation Alerte Meteo..",1,"")
        # Ecoute l'arrivée de la pluie imminente
        self.listen_state(self.change_pluie, self.args["entité"])
        
        forcast = self.get_state(self.args["entité"],attribute="forecast_time_ref")
        self.log(f"Forecast: {forcast}", log="test_log")
        
        # Ecoute une alerte méteo
        self.listen_state(self.change_alerte, self.args["alerte"])

    def change_pluie(self, entity, attribute, old, new, kwargs):
        global FLAG
        message_notification= "New: "+new+" / Old: "+old+" / Flag="+ str(FLAG)
        self.notification(message_notification,2,"")
        dic_forcast = self.get_state(self.args["entité"],attribute="1_hour_forecast")
        # Calcul décalage UTC <-> heure local
        decalage_utc=self.get_tz_offset()/60
        self.notification("Decalage UTC: "+str(decalage_utc),2,"")
        # Affiche l'heure locale
        #h_locale=time.strftime('%H:%M:%S', time.localtime())[:5]
        #self.notification("Heure Locale:" + str(h_locale),2,"")
        if new!="unavailable":
            if new!="unknown":
                # Ajoute le décalage UTC/Heure locale à l'heure transmise par Meteo France
                h_pluie=datetime.strftime(datetime.strptime(new[11:19],'%H:%M:%S')+timedelta(hours=decalage_utc),"%H:%M:%S")
                h_forcast[0] = dic_forcast['0 min']
                h_forcast[1] = dic_forcast['5 min']
                h_forcast[2] = dic_forcast['10 min']
                h_forcast[3] = dic_forcast['15 min']
                h_forcast[4] = dic_forcast['20 min']
                h_forcast[5] = dic_forcast['25 min']
                h_forcast[6] = dic_forcast['35 min']
                h_forcast[7] = dic_forcast['45 min']
                h_forcast[8] = dic_forcast['55 min']
                for h in range(8):
                    message_notification= "h"+str(h)+"_Forecast:"+h_forcast[h]
                    self.notification(message_notification,2,"")
                    
                if FLAG==0: # Permet d'afficher le message une seule fois
                    message_notification= " La pluie est attendue a "+ format(h_pluie)
                    self.notification(message_notification,1,"teleg")
                    FLAG = 1

            if new=="unknown" and h_forcast[0] == "Temps sec":
                message_notification= " Plus de pluie attendue."
                FLAG=0
                self.notification(message_notification,1,"teleg")
            
            message_notification= "Flag="+ str(FLAG)
            self.notification(message_notification,2,"")

    def change_alerte(self, entity, attribute, old, new, kwargs):
        heure = str(self.time())[:8]
        alerte_w = self.get_state(self.args["alerte"],attribute="state")
        Inondation= self.get_state(self.args["alerte"],attribute="Inondation")
#        Grand_Froid= self.get_state(self.args["alerte"],attribute="Grand-froid")
        Neige_Verglas= self.get_state(self.args["alerte"],attribute="Neige-verglas")
        Orages= self.get_state(self.args["alerte"],attribute="Orages")
        Pluie_inondation= self.get_state(self.args["alerte"],attribute="Pluie-inondation")
        Vent_violent=self.get_state(self.args["alerte"],attribute="Vent violent")
        attribution= self.get_state(self.args["alerte"],attribute="attribution")
        
        if alerte_w != 'Vert':
            if alerte_w != 'unavailable':
                if Inondation != 'Vert':
                    message_notification= ": Alerte Innondation :"+ Inondation
                    self.notification(message_notification,1,"teleg")

                if Grand_Froid != 'Vert':
                    message_notification= ": Alerte Grand-froid :"+Grand_Froid
                    self.notification(message_notification,1,"teleg")

                if Neige_Verglas != 'Vert':
                    message_notification= ": Alerte Neige Verglas :"+Neige_Verglas
                    self.notification(message_notification,1,"teleg")

                if Orages != 'Vert':
                    message_notification= ": Alerte Orages :"+Orages
                    self.notification(message_notification,1,"teleg")

                if Pluie_inondation != 'Vert':
                    message_notification= ": Alerte Pluie Inondation :"+Pluie_inondation
                    self.notification(message_notification,1,"teleg")

                if Vent_violent != 'Vert':
                    message_notification= ": Alerte vents Violents :"+Vent_violent
                self.notification(message_notification,1,"teleg")
            else:
                message_notification= ": Fin Alerte Météo "
                self.notification(message_notification,1,"teleg")
                
    # Fonction Notification
    # message =  Texte à afficher
    # niveau = niveau de journalisation 0,1,2
    # si notif == "teleg" on notifie aussi sur Télégram
    def notification(self,texte_message,niveau,notif):
        global JOURNAL
        heure = str(self.time())[:8]
        if niveau <= JOURNAL:
            message_notification= format(heure)+"-"+ texte_message
            self.log(message_notification, log="test_log")
            if notif=="teleg":
                self.call_service('notify/telegram', message=message_notification)
                self.call_service('dwains_dashboard/notification_create', message=message_notification)
                #self.call_service('persistent_notification.create', message=message_notification)

Tableau de bord- exemple

Pour pouvoir l’utiliser, installer « meteo-france-weather-card » avec HACS.

Le sensor UV est indépendant de Meteo France.

type: vertical-stack
cards:
  - type: custom:meteo-france-weather-card
    name: Albi
    entity: weather.albi
    rainForecastEntity: sensor.albi_next_rain
    freezeChanceEntity: sensor.albi_freeze_chance
    rainChanceEntity: sensor.albi_rain_chance
    snowChanceEntity: sensor.albi_snow_chance
    thunderChanceEntity: sensor.albi_thunder_chance
    alertEntity: sensor.albi_weather_alert
    uvEntity: sensor.esp134_uv
    cloudCoverEntity: sensor.albi_cloud_cover

Interface Utilisateur Web

Vous y avez accès via le port de « 5050 » de HA (par exemple http://192.168.0.37:5050/). le port « 5050 » est celui déclaré dans la configuration de l’Addon:

Vous retrouverez dans l’UI de AppDaemon l’etat, les logs etc…

Dans l’ onglet « State », vous retrouvez vos applications App, le nom correspond à celui de la class de votre fichier « *.py ».

« State Idle » signifie que tout va bien. « compile error » signifié que votre programme fonctionne pas, dans ce cas vérifiez le programme python.

L’onglet « log » vous permet de sélectionner votre fichier log précédemment déclaré le « .yaml »

Le log « main_log » est commun à toutes les applications, c’est aussi le log par défaut.

Dans le log « error_log », vous retrouverez les erreurs de programmation, fonctionnement, etc..

Conclusion

J’ai fait le plus simple et le plus détaillé possible dans l’installation et l’utilisation de AppDaemon.

Vous l’avez compris, à travers cet Addon et sa puissance de feu, vous offrez de nombreuses fonctionnalités à votre HA.

N’hésitez pas à commenter, vos critiques positives et négatives constructrices seront les bienvenues.

Publication en lien avec cet article:

Références

Voici quelques liens qui m’ont aidé dans la découverte de Appdaemon:

Laisser un commentaire

Votre adresse e-mail ne sera pas publiée.

Ce site utilise Akismet pour réduire les indésirables. En savoir plus sur comment les données de vos commentaires sont utilisées.