HELP Créer une automatisation pour détecter une fuite d'eau via l'utilisation du courant électrique de la pompe de surpression

mowgly13 · Mars 12, 2025, 1:46

Bonjour à toute la comu,

Je viens à peine de me lancer dans l’aventure Home Assistant et j’aurais besoin d’aide pour la création d’une automatisation. Je solicite donc aimablement votre aide…

J’ai une pompe de surpression d’eau qui s’enclenche quand c’est nécéssaire pour maintenir la pression dans ma tuyauterie et qui est connectée à un switch Wifi qui permet de monitorer sa consomation électrique en temps réel.
J’ai pu integrer ce switch dans Home Assistant et sa valeur (en Watt) comme entitée → sensor.power_pompe

J’aimerais maintenant créer une automatisation qui calcule le temps entre chaque utilisation (plus de 500w sur sensor.power_pompe) et qui, si elle détecte par exemple 3 ou 4 déclanchements consecutifs MAIS avec le même temps entre chaque déclanchement (intervalles identiques), m’enverrai une notification.

Ca serait en faite la preuve d’une fuite d’eau dans mon sisteme de tuyauterie, un robinet mal fermé ou une chasse d’eau qui laisse passer un peu d’eau…

Aussi et dans l’idéal, j’aurais besoin de pouvoir modifier assez simplement dans le code le nombre de déclanchement consécutifs avant d’envoyer la notif, par exemple après 3, 4 ou 10 declanchements si ca devenait nécéssaire…

Concernant les intervalles de temps, j’aimerais aussi pouvoir définir une tolérance en secondes que je pourrais changer au fur et à mesure de mes test…

La seule chose de plus ou moins similaire trouvé sur le net est cette page en anglais:

Un énorme merci

Home Assistant Raspberry Pi 3b

System Information

version	core-2025.3.1
installation_type	Home Assistant OS
dev	false
hassio	true
docker	true
user	root
virtualenv	false
python_version	3.13.2
os_name	Linux
os_version	6.6.62-haos-raspi
arch	aarch64
timezone	America/Cancun
config_dir	/config

Home Assistant Community Store

GitHub API	ok
GitHub Content	ok
GitHub Web	ok
HACS Data	ok
GitHub API Calls Remaining	5000
Installed Version	2.0.5
Stage	running
Available Repositories	1581
Downloaded Repositories	9

Home Assistant Cloud

logged_in	false
can_reach_cert_server	ok
can_reach_cloud_auth	ok
can_reach_cloud	ok

Home Assistant Supervisor

host_os	Home Assistant OS 14.2
update_channel	stable
supervisor_version	supervisor-2025.03.2
agent_version	1.6.0
docker_version	27.2.0
disk_total	28.6 GB
disk_used	4.0 GB
healthy	true
supported	true
host_connectivity	true
supervisor_connectivity	true
ntp_synchronized	true
virtualization
board	rpi3-64
supervisor_api	ok
version_api	ok
installed_addons	Matter Server (7.0.0), Terminal & SSH (9.16.0)

Dashboards

dashboards	2
resources	0
views	1
mode	storage

Network Configuration

adapters	lo (disabled), enu1u1 (enabled, default, auto), wlan0 (disabled), docker0 (disabled), hassio (disabled), vethb6e238a (disabled), vethb880551 (disabled), veth1d60b47 (disabled), vethf13729f (disabled), vetha8d28c5 (disabled), veth5d5f517 (disabled), vethb8f8e62 (disabled)

Recorder

oldest_recorder_run	5 mars 2025 à 21:41
current_recorder_run	11 mars 2025 à 16:00
estimated_db_size	137.29 MiB
database_engine	sqlite
database_version	3.48.0

___

mowgly13 · Mars 12, 2025, 2:08

J’ai fais un petit schéma pour illustrer un peu les comportements de sensor.power_pompe qui nous intéresse…

Normalement l’utilisation de la pompe sur une journée est totalement aléatoire comme on voit sur cette capture, aussi bien en temps d’utilisation qu’en intervalle entre chaque utilisation…

Et si on zoom un peu ça reste aléatoire

Sauf ce genre de schéma qui sont anormalement régulier aussi bien dans le temps d’utilisation que dans l’interval entre ces utilisations…

L’idée est donc de genre créer un log qui note les temps entre OFF et le prochain ON (intervalles de temps) et ainsi de suite de chaque utilisation, et que quand il remarque qu’il y a 3 intervalles identiques, (avec une tolérance de genre +15 secondes ou -15 secondes) c’est considéré comme anormal et donc notification car la seule façon d’avoir ce genre de schéma si régulier ne peut être qu’une fuite…

Ici dans l’exemple les temps d’ Intervals 294, 295 et 296 sont anormalement similaires (a + ou - 15 secondes prêt)

ATENTION: les intervalles peuvent être de + de 1h entre chaque OFF et ON mais peut importe, si il y a un schéma régulier même de par exemple 1h34 et 12 secondes entre 3 OFF et ON bah c’est un toute petite fuite mais ça reste une fuite…

hackdiy · Mars 12, 2025, 2:50

Salut,

Voici un code que tu peux essayer , et adapter a ta convenance .

Tu pourras facilement modifier :

Le nombre de déclenchements consécutifs** nécessaires avant d’envoyer la notif.
La tolérance en secondes pour accepter de légères différences entre les intervalles.
La puissance seuil pour considérer que la pompe s’est allumée.

Avant tout il te faut crée un helper input_text pour stocker l’historique :

pour le crée va dans Paramètres → Appareils & Services → Entités → Helpers (Aides)

Créer un helper
Choisis Texte
Nomme-le historique_pompe
Valide

alias: "Détection de fuite d'eau"
mode: restart
trigger:
  - platform: numeric_state
    entity_id: sensor.power_pompe
    above: 500  # Seuil de puissance indiquant que la pompe est active

action:
  - variables:
      nombre_declenchements: 3  # Nombre de cycles avant alerte
      tolerance_secondes: 30    # Tolérance sur la régularité des cycles
      historique: >-
        {{ state_attr('input_text.historique_pompe', 'history') | default('[]') | from_json }}

  - service: input_text.set_value
    data:
      entity_id: input_text.historique_pompe
      value: >-
        {{ (historique + [now().timestamp()])[-nombre_declenchements:] | to_json }}

  - condition: template
    value_template: >
      {% if historique | length < nombre_declenchements %}
        false
      {% else %}
        {% set deltas = historique | sort | batch(2, keep=True) | map('difference') | list %}
        {% set premiere_difference = deltas[0] %}
        {{ deltas | rejectattr('first', 'lessthan', premiere_difference - tolerance_secondes) 
                   | rejectattr('first', 'greaterthan', premiere_difference + tolerance_secondes) 
                   | list | length == deltas | length }}
      {% endif %}

  - service: notify.mobile_app_ton_telephone
    data:
      title: "🚨 Fuite d'eau détectée !"
      message: "La pompe s'est déclenchée {{ nombre_declenchements }} fois à intervalles réguliers."

Dis-moi si sa fonctionne .

mowgly13 · Mars 12, 2025, 3:10

Salut Hackdiy,

Super merci pour ta réponse mais je n’arrive pas a créer le fichier texte…

Je bute ici , je vais dans l’onglet Entités mais après je ne vois pas comment selectionner Helpers…

pour le crée va dans Paramètres → Appareils & Services → Entités → Helpers (Aides)

mowgly13 · Mars 12, 2025, 3:43

j’ai trouvé les helpers… C’était un onglais plus loin → Entrées

J’ai créé le fichier texte, nommé comme indiqué, j’ai ensuite été creer une automatisation, Modifier en YAML, copier/coller ton code en changeant mon téléphone mais malheureusement ca ne fonctionne pas, il ne se passe rien…

hackdiy · Mars 12, 2025, 11:16

1 - Vérifie l’historique pour voire si l’automatisation se déclenche, si non, essaye de la déclencher manuellement.

2 - Vérifie la valeur de input_text.historique_pompe

Dans Outils de développement → États, cherche l’entité input_text.historique_pompe.

Si sa valeur est vide ou « unknown », l’automatisation ne l’a peut-être pas encore mise à jour.
Si elle contient une liste JSON, tout va bien.

Tu peu aussi forcer un test manuel
Dans Outils de développement → Services, exécute en entrant ce code :

service: input_text.set_value
data:
  entity_id: input_text.historique_pompe
  value: "[{{ now().timestamp() }}, {{ now().timestamp() - 600 }}, {{ now().timestamp() - 1200 }}]"

Ça va simuler trois activations espacées de 10 minutes.
Si l’automatisation fonctionne, elle devrait envoyer une notification.

Dis-moi ce que tu observes !

mowgly13 · Mars 12, 2025, 12:37

Mes observations:

Quand j’ai été voir dans l’historique je n’ai rien vu…
J’ai lancé manuellement l’automatisation et je suis retourné voir et j’ai vu ça:

J’ai été vérifier la valeur de input_text.historique_pompe et elle était sur UNKNOW.

Par contre, après avoir forcé le test comme tu le suggère, je suis retourné voir la valeur et là j’ai vu ça:

J’ai effectivement forcé le test 4 fois…

Sauf que je n’ai jamais reçu de notification.

hackdiy · Mars 12, 2025, 1:02

L’erreur JSONDecodeError arrive souvent quand input_text.historique_pompe est unknown en manuel la valeur à charger, c’est une bonne chose.

il faut regarder l’historique de l’automatisation.

Va dans Paramètres → Automatisations et trouve ton automatisation « Détection de fuite d’eau ».

Clique dessus, puis dans l’onglet « Journal »,

mowgly13 · Mars 12, 2025, 1:13

c’est a ça que tu te réfère ?

Je suis pas certain de bien comprendre ce que je vois mais a chaque fois que la pompe s’allume, elle a l’air de se lancer mais par contre elle bloque a cette deuxième étape… Avec le code d’erreur :

Arrêté car une erreur s’est produite le 12 mars 2025 à 07:55:59 (durée d’exécution: 0.01 secondes)

JSONDecodeError: Input must be bytes, bytearray, memoryview, or str: line 1 column 1 (char 0)

(Ne serait-ce pas un truc tout bête du genre que j’ai pas activé python script pour les automatisations ? A un moment j’ai lu un truc du genre sur internet mais ne se suis pas certain de ce que ça veut dire)

hackdiy · Mars 12, 2025, 2:23

Vérifie comme précédemment la valeur de input_text.historique_pompe

Si elle contient une liste JSON, tout va bien. Essaye de l’effacer (input_text.set_value avec une valeur vide) et relance l’automatisation.

modifie la partie suivante pour éviter cette erreur

Initialisation de historique

Vérifie que la valeur de input_text.historique_pompe est bien un JSON valide avant d’utiliser from_json. :

historique: >-
  {% set valeur = states('input_text.historique_pompe') %}
  {% if valeur and valeur != 'unknown' and valeur != 'unavailable' %}
    {{ valeur | from_json }}
  {% else %}
    []
  {% endif %}

mowgly13 · Mars 12, 2025, 3:23

Alors depuis que j’ai modifié le code il se passe plus de chose dans l’Entrée historique_pompe… A chaque démarrage de la pompe, il y a une « entrée » qui apparaît, ça a l’air ok de ce côté là…
(peut-être est-ce d’ailleurs ca que tu appel la « Liste Json »)

J’ai essayé d’effacer avec l’action input_text.set_value mais il marque une erreur du fait que le champ Valeur est vide.

Quand je regarde l’historique de l’automatisation il n’y a plus de message d’erreur et il me semble que l’automatisation arrive au bout de son process… Ceci dit, je simule une fuite laissant couler un filet d’eau sur un robinet, la pompe s’enclenche 3 voir même 4 fois d’affilé avec ± le même temps entre chaque déclenchement mais l’automatisation se termine toujours en False, ce qui, si je déduis bien, n’envoie jamais la notification… Car en éffet je ne recois jamais la notification… Ca bloque quelque part.

Serais-ce un problème dans les critères de l’automatisation, qu’il ne détecte pas bien le temps entre chaque démarrage?
Ou bien lié au fait que la pompe fonctionne quelques secondes a plus de 500w avant de s’arrêter, et que du coup il ne sais pas si il doit prendre comme point de départ pour commencer a compter a partir de l’enclenchement ou a partir de l’arrêt de la pompe ?

En tout cas super merci pour ton aide et ta patience

Le code est maintenant celui ci avec la derniere modification demandée, j’ai juste pris l’initiative de diminuer la tolérence a 10 secondes, en m’assurant de ce point en faisant mes simulations de fuite d’eau

alias: Détection de fuite d'eau
description: ""
triggers:
  - entity_id: sensor.power_pompe
    above: 500
    trigger: numeric_state
actions:
  - variables:
      nombre_declenchements: 3
      tolerance_secondes: 10
      historique: >-
        {% set valeur = states('input_text.historique_pompe') %} {% if valeur
        and valeur != 'unknown' and valeur != 'unavailable' %}
          {{ valeur | from_json }}
        {% else %}
          []
        {% endif %}
  - data:
      entity_id: input_text.historique_pompe
      value: >-
        {{ (historique + [now().timestamp()])[-nombre_declenchements:] | to_json
        }}
    action: input_text.set_value
  - condition: template
    value_template: |
      {% if historique | length < nombre_declenchements %}
        false
      {% else %}
        {% set deltas = historique | sort | batch(2, keep=True) | map('difference') | list %}
        {% set premiere_difference = deltas[0] %}
        {{ deltas | rejectattr('first', 'lessthan', premiere_difference - tolerance_secondes) 
                   | rejectattr('first', 'greaterthan', premiere_difference + tolerance_secondes) 
                   | list | length == deltas | length }}
      {% endif %}
  - data:
      title: 🚨 Fuite d'eau détectée !
      message: >-
        La pompe s'est déclenchée {{ nombre_declenchements }} fois à intervalles
        réguliers.
    action: notify.mobile_app_pixel_9_pro
mode: restart

hackdiy · Mars 12, 2025, 5:13

C’est une très bonne question ! Oui, le problème pourrait effectivement être lié à la façon dont l’automatisation interprète les démarrages et arrêts de la pompe.

Problème possible :

Si la pompe monte rapidement à plus de 500W puis redescend sous 50W en moins de 5s, l’automatisation ne se déclenche pas.
Il pourrait aussi considérer un seul long démarrage au lieu de plusieurs petits cycles si la puissance oscille.

Actuellement, l’automatisation déclenche quand la puissance dépasse 50W, mais ne surveille pas quand elle retombe sous ce seuil.
Si ta pompe a des cycles très courts et s’arrête complètement après chaque démarrage, on peut ajouter un deuxième déclencheur pour détecter les arrêts :
à vérifier
modifie cette partie

trigger:
  - platform: numeric_state
    entity_id: sensor.power_pompe
    above: 50
    for:
      seconds: 2
    id: "start"

  - platform: numeric_state
    entity_id: sensor.power_pompe
    below: 10
    for:
      seconds: 2
    id: "stop"

ont va faire mise à jour du comptage des démarrages, au lieu de prendre uniquement le moment où la pompe démarre, on peut aussi attendre qu’elle s’arrête avant d’enregistrer un cycle.

A ajout dans variables :

  nouvel_evenement: >-
    {% if trigger.id == "stop" %}
      {{ now().timestamp() }}
    {% else %}
      none
    {% endif %}

mise à jour égalent dans l’historique (démarrage + arrêt) :

  historique_filtre: >-
    {{ historique_maj | select('float', '>', (now().timestamp() - tolerance_secondes)

mowgly13 · Mars 12, 2025, 5:53

Je ne suis pas certain de te suivre pour la dernière partie :

mise à jour égalent dans l’historique (démarrage + arrêt) :

historique_filtre: >-
    {{ historique_maj | select('float', '>', (now().timestamp() - tolerance_secondes)

Je ne suis pas certain de comprendre où je dois remplacer ou ajouter cette partie…

hackdiy · Mars 12, 2025, 6:24

La variable historique_filtre doit être définie dans la section variables de ton automatisation. Elle doit être placée après que historique_maj soit mis à jour pour s’assurer qu’on filtre les données les plus récentes.

mowgly13 · Mars 12, 2025, 6:38

Alors désolé, je fais de mon mieux mais je comprends toujours pas où je dois copier/coller… Serais-ce abusé de te demander si tu veux bien l’intégrer pour moi dans le code stp…

Rappel toi, je ne connais rien en codage, je ne fais que du copier/coller depuis tout à l’heure et là definitivement, tu m’as eu…

Ou si tu veux bien me dire précisément si je dois remplacer une partie par ca, ou si je dois l’ajouter après un mot ou après une ligne specifique, je crois avoir identifié la zone des variables parceque à un moment je vois le mot « variables » mais je dois remplacer tout ce qu’il y a dedans? de où a où? Ca ne correspond pas…

Super Merci

Je crois avoir réussi a faire ca… Mais à chaque fois que j’éssaye de copier/coller les lignes historique_filtre… dans le code, a tous les endroits éssayé j’ai un message d’erreur.

Bref j’en suis là:

alias: Détection de fuite d'eau
description: ""
triggers:
  - entity_id: sensor.power_pompe
    above: 50
    for:
      seconds: 2
    id: start
    trigger: numeric_state
  - entity_id: sensor.power_pompe
    below: 10
    for:
      seconds: 2
    id: stop
    trigger: numeric_state
actions:
  - variables:
      nombre_declenchements: 3
      tolerance_secondes: 10
      historique: >-
        {% set valeur = states('input_text.historique_pompe') %} {% if valeur
        and valeur != 'unknown' and valeur != 'unavailable' %}
          {{ valeur | from_json }}
        {% else %}
          []
        {% endif %}  
      nouvel_evenement: |-
        {% if trigger.id == "stop" %}
          {{ now().timestamp() }}
        {% else %}
          none
        {% endif %}
  - data:
      entity_id: input_text.historique_pompe
      value: >-
        {{ (historique + [now().timestamp()])[-nombre_declenchements:] | to_json
        }}
    action: input_text.set_value
  - condition: template
    value_template: |
      {% if historique | length < nombre_declenchements %}
        false
      {% else %}
        {% set deltas = historique | sort | batch(2, keep=True) | map('difference') | list %}
        {% set premiere_difference = deltas[0] %}
        {{ deltas | rejectattr('first', 'lessthan', premiere_difference - tolerance_secondes) 
                   | rejectattr('first', 'greaterthan', premiere_difference + tolerance_secondes) 
                   | list | length == deltas | length }}
      {% endif %}
  - data:
      title: 🚨 Fuite d'eau détectée !
      message: >-
        La pompe s'est déclenchée {{ nombre_declenchements }} fois à intervalles
        réguliers.
    action: notify.mobile_app_pixel_9_pro
mode: restart

hackdiy · Mars 12, 2025, 7:55

tu ne peut trouver la solution que par toi même avec les details fourni .

a toi de trouver la bonne formule pour le fonctionnement .

voilà une version complète et corrigée et optimisée pour bien gérer l’historique des déclenchements et éviter les erreurs JSON.

j’espère que ça sera la bonne

alias: "Détection de fuite d'eau"
description: "Détecte une fuite d'eau en analysant les déclenchements répétés de la pompe."
mode: restart

trigger:
  - platform: numeric_state
    entity_id: sensor.power_pompe
    above: 500
    id: "start"
  
  - platform: numeric_state
    entity_id: sensor.power_pompe
    below: 500
    id: "stop"

variables:
  nombre_declenchements: 3
  tolerance_secondes: 30

  # Récupération de l'historique depuis input_text
  historique: >-
    {% set raw = states('input_text.historique_pompe') %}
    {% if raw in ['unknown', 'unavailable', 'none'] %}
      []
    {% else %}
      {{ raw | from_json }}
    {% endif %}

  # Détecte un nouvel arrêt de la pompe
  nouvel_evenement: >-
    {% if trigger.id == "stop" %}
      {{ now().timestamp() }}
    {% else %}
      none
    {% endif %}

  # Ajoute le nouvel événement à l'historique
  historique_maj: >-
    {% if nouvel_evenement != 'none' %}
      {{ historique + [nouvel_evenement] }}
    {% else %}
      {{ historique }}
    {% endif %}

  # Filtrage des événements récents
  historique_filtre: >-
    {{ historique_maj | select('float', '>', (now().timestamp() - tolerance_secondes)) | list }}

condition:
  - condition: template
    value_template: "{{ historique_filtre | length >= nombre_declenchements }}"

action:
  - alias: "Mettre à jour l'historique"
    service: input_text.set_value
    target:
      entity_id: input_text.historique_pompe
    data:
      value: "{{ historique_filtre | to_json }}"

  - alias: "Envoyer une notification"
    service: notify.mobile_app_pixel_9_pro
    data:
      title: "Alerte Fuite d'Eau 🚨"
      message: "La pompe s'est déclenchée {{ historique_filtre | length }} fois en {{ tolerance_secondes }} secondes !"

tu peut aussi augmenter la longueur maximal de ton input_text crée précédemment a 255.

hackdiy · Mars 13, 2025, 1:25

On va crée deux automatisation pour séparé le processus en deux pour plus de clarté et de fiabilité :
Explication des corrections

Première automation : Enregistrement des démarrages

Stocke les 4 derniers démarrages de la pompe.
Garde un historique pour calculer les intervalles.

Deuxième automation : Détection d’une fuite

Compare les 3 derniers intervalles de démarrage.
Vérifie si les intervalles sont similaires avec une tolérance.
Envoie une notification si une fuite est détectée.

Essaie comme ça

Stockage des démarrages

On garde les 4 derniers démarrages et compare les 3 intervalles successifs avec la tolérance

alias: "Log démarrages pompe"
trigger:
  - platform: numeric_state
    entity_id: sensor.power_pompe
    above: 10  # Seuil de puissance pour considérer que la pompe est en marche (à ajuster)
action:
  - variables:
      historique: >-
        {% set h = states('input_text.historique_pompe') %}
        {% if h in ['unknown', 'unavailable', 'none', ''] %}
          []
        {% else %}
          {{ h | from_json }}
        {% endif %}
  - service: input_text.set_value
    target:
      entity_id: input_text.historique_pompe
    data:
      value: "{{ (historique + [now().timestamp()])[-4:] | to_json }}"

Détection de fuite basée sur 3 intervalles

Envoie de notification si une fuite est détectée avec les intervalles mesurés.

alias: "Détection de fuite d'eau"
trigger:
  - platform: state
    entity_id: input_text.historique_pompe
action:
  - variables:
      tolerance: 10  # Tolérance en secondes
      h: >-
        {% set h = states('input_text.historique_pompe') %}
        {% if h in ['unknown', 'unavailable', 'none', ''] %}
          []
        {% else %}
          {{ h | from_json }}
        {% endif %}
  - condition: template
    value_template: "{{ h | length == 4 }}"
  - variables:
      interval_1: "{{ (h[1] - h[0]) | int }}"
      interval_2: "{{ (h[2] - h[1]) | int }}"
      interval_3: "{{ (h[3] - h[2]) | int }}"
  - condition: template
    value_template: >-
      {{ (interval_1 - interval_2) | abs <= tolerance and
         (interval_2 - interval_3) | abs <= tolerance }}
  - service: notify.notify
    data:
      title: "🚨 Fuite détectée !"
      message: "Démarrages réguliers détectés (~{{ interval_1 }}s, ~{{ interval_2 }}s, ~{{ interval_3 }}s)"

Avec ça, la logique est respectée ! Dis-moi si ça fonctionne comme attendu !

mowgly13 · Mars 14, 2025, 4:41

CA MARRRCHHHEEEE !!!

J’ai vérifié que l’entrée input_text.historique_pompe soit toujours là.

J’ai créé deux nouvelles automatisations en copiant/collant tes deux codes.

J’ai simulé une fuite en laissant couler un mince filet d’eau d’un robinet.

Et les notifications sont bien arrivées !

Un énorme merci @hackdiy pour ta patience… T’as littéralement tout fait !

hackdiy · Mars 14, 2025, 5:04

Avec grand plaisir ! J’adore relever ce genre de défis techniques et voir tout fonctionner comme prévu.

Amuse-toi bien avec Home Assistant et profite de ton système ultra-intelligent maintenant !

mowgly13 · Mars 14, 2025, 5:20

Comme tu dit, ultra-intelligent et c’est ca la beauté de la chose… Grace à toi et tes deux codes, je dois plus me préocuper de cette pompe… Si il y a une fuite, je serais automatiquement tenu au courant.

C’est fou quand on y pense, maintenant c’est littéralement comme si un gars était 24h/24h devant cette pompe avec un papier, un crayon et un chrono prêt a m’appeler… C’est peut-être pas grand chose pour toi mais pour moi c’est facinant !

Quelle tranquilité d’esprit pour moi…

Encore merci