Surveillance de batterie domestique avec MQTT : état de charge, tensions de cellules et alertes de batterie faible

La surveillance de batterie domestique vous indique combien d'énergie est stockée dans la batterie de votre maison connectée, à quel point la batterie est sollicitée et si une cellule commence à dériver, bien avant que cela ne devienne une panne. Dans ce tutoriel, vous allez publier les données du système de gestion de batterie (BMS), comme l'état de charge, la tension du pack, le courant, les tensions de cellules et la température, vers un broker MQTT, avec un système Victron Venus OS comme exemple. Vous ajouterez aussi des alertes de batterie faible et de déséquilibre des cellules qui ne submergent pas votre téléphone.

Que surveiller sur une batterie domestique

Un BMS expose des dizaines de valeurs. Voici celles qui comptent pour l'exploitation au quotidien et la détection précoce des défauts :

Pour les packs lithium fer phosphate (LiFePO4), l'écart de tension entre cellules est souvent le signal le plus parlant à long terme. Le SoC est estimé par le BMS et peut dériver ; les tensions de cellules sont mesurées. Un écart qui grandit de mois en mois signale un problème de cellule ou d'équilibrage.

D'où viennent les données

La façon d'obtenir les données du BMS dépend du système :

1. Systèmes Victron avec un appareil GX (Venus OS). L'appareil GX fait tourner un broker MQTT local qui expose toutes les valeurs du D-Bus interne. Activez-le dans Settings > Services (appelé Integrations sur les firmwares récents) avec l'option « MQTT on LAN ».
2. Les onduleurs hybrides exposent souvent le SoC, la tension et le courant de la batterie via des registres Modbus, de la même manière que les données PV ; consultez le suivi d'onduleur solaire pour l'approche Modbus.
3. Les packs DIY avec cartes BMS JK, JBD ou Daly proposent des interfaces Bluetooth ou UART. Des firmwares ESP32 (dont des composants ESPHome) les lisent et publient en MQTT.
4. Les batteries en rack serveur avec ports CAN ou RS485 remontent généralement leurs données à l'onduleur, qui devient alors votre source de données.

La suite de ce guide utilise Venus OS, car son interface MQTT est bien documentée, mais la logique de publication et d'alerte fonctionne avec n'importe quelle source.

Fonctionnement de MQTT sous Venus OS

L'appareil GX publie les valeurs sur des topics de la forme N/<portal-id>/<service>/<instance>/<path>, avec un payload JSON tel que {"value": 87.5}. Les valeurs de la batterie se trouvent sous le service battery :

N/<portal-id>/battery/512/Soc
N/<portal-id>/battery/512/Dc/0/Voltage
N/<portal-id>/battery/512/Dc/0/Current
N/<portal-id>/battery/512/Dc/0/Power
N/<portal-id>/battery/512/Dc/0/Temperature
N/<portal-id>/battery/512/System/MinCellVoltage
N/<portal-id>/battery/512/System/MaxCellVoltage

Le numéro d'instance (512 ici) varie selon l'installation, et les chemins des tensions de cellules ne sont disponibles que si le BMS les remonte. Deux détails piègent souvent. D'abord, l'appareil GX ne continue à publier que si un client envoie régulièrement une requête keepalive sur R/<portal-id>/keepalive ; sans elle, les mises à jour s'arrêtent au bout d'une minute environ. Ensuite, l'identifiant du portail apparaît dans le topic de N/<portal-id>/system/0/Serial, ce qui permet à un script de le découvrir.

Relayer les données de la batterie vers votre broker

Le script ci-dessous se connecte à l'appareil GX sur le réseau local, envoie les keepalives, collecte les valeurs de la batterie et publie un résumé JSON compact toutes les 10 secondes vers votre propre broker. Il se teste sur le serveur MQTT public gratuit broker.connectmq.com, qui ne nécessite aucun compte mais est lisible par tous et réservé aux tests ; il utilise donc un préfixe aléatoire.

pip install "paho-mqtt>=2.0"

import json
import threading
import time
import uuid
import paho.mqtt.client as mqtt

GX_HOST = "192.168.1.70"            # appareil GX Venus OS
PREFIX = f"connectmq/demo/{uuid.uuid4().hex[:8]}/battery/house"
PUBLISH_EVERY = 10

PATHS = {
    "Soc": "soc_pct",
    "Dc/0/Voltage": "voltage_v",
    "Dc/0/Current": "current_a",
    "Dc/0/Power": "power_w",
    "Dc/0/Temperature": "temp_c",
    "System/MinCellVoltage": "cell_min_v",
    "System/MaxCellVoltage": "cell_max_v",
}

values = {}
portal = {"id": None}

dest = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"batt-pub-{uuid.uuid4().hex[:6]}")
dest.will_set(f"{PREFIX}/status", "offline", qos=1, retain=True)
dest.connect("broker.connectmq.com", 1883, keepalive=60)
dest.loop_start()
dest.publish(f"{PREFIX}/status", "online", qos=1, retain=True)


def gx_connect(client, userdata, flags, reason_code, properties):
    print("Connected to GX:", reason_code)
    client.subscribe("N/+/system/0/Serial")
    client.subscribe("N/+/battery/+/#")


def gx_message(client, userdata, msg):
    parts = msg.topic.split("/")
    if parts[2] == "system" and parts[-1] == "Serial" and portal["id"] is None:
        portal["id"] = parts[1]
        print("Portal ID:", portal["id"])
        client.publish(f"R/{portal['id']}/keepalive", "")
        return
    if parts[2] != "battery":
        return
    path = "/".join(parts[4:])
    if path in PATHS:
        try:
            values[PATHS[path]] = json.loads(msg.payload).get("value")
        except (ValueError, AttributeError):
            pass


def keepalive_loop(client):
    while True:
        if portal["id"]:
            client.publish(f"R/{portal['id']}/keepalive", "")
        time.sleep(30)


gx = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"batt-gx-{uuid.uuid4().hex[:6]}")
gx.on_connect = gx_connect
gx.on_message = gx_message
gx.connect(GX_HOST, 1883, keepalive=60)
gx.loop_start()
threading.Thread(target=keepalive_loop, args=(gx,), daemon=True).start()

print("Publishing to", PREFIX)
while True:
    time.sleep(PUBLISH_EVERY)
    if not values:
        continue
    out = dict(values)
    if out.get("cell_min_v") is not None and out.get("cell_max_v") is not None:
        out["cell_spread_mv"] = round((out["cell_max_v"] - out["cell_min_v"]) * 1000)
    out["ts"] = int(time.time())
    dest.publish(f"{PREFIX}/telemetry", json.dumps(out), qos=1, retain=True)
    print(out)

Observez le résultat :

mosquitto_sub -h broker.connectmq.com -p 1883 -i batt-view-3318 -t "connectmq/demo/+/battery/#" -v

{"soc_pct": 63.4, "voltage_v": 52.71, "current_a": -18.2, "power_w": -959,
 "temp_c": 21.0, "cell_min_v": 3.291, "cell_max_v": 3.304, "cell_spread_mv": 13, "ts": 1760611200}

Un courant et une puissance négatifs signifient que la batterie se décharge. Gardez cette convention de signe cohérente entre les sources pour que les tableaux de bord et les automatisations n'aient pas besoin de cas particuliers.

Des alertes de surveillance de batterie domestique sans bruit

Une alerte de batterie faible qui se déclenche toutes les dix secondes pendant que le SoC oscille autour du seuil apprend aux gens à l'ignorer. Utilisez une hystérésis (des seuils différents pour lever et pour effacer l'alerte) et publiez les transitions d'alerte, pas chaque relevé :

import json
import paho.mqtt.client as mqtt

TELEMETRY = "connectmq/demo/<your-id>/battery/house/telemetry"
ALERTS = "connectmq/demo/<your-id>/battery/house/alerts"

RULES = {
    # nom : (champ, lever_si, effacer_si)
    "low_soc": ("soc_pct", lambda v: v < 20, lambda v: v > 25),
    "cell_imbalance": ("cell_spread_mv", lambda v: v > 50, lambda v: v < 30),
    "cold_battery": ("temp_c", lambda v: v < 5, lambda v: v > 8),
    "hot_battery": ("temp_c", lambda v: v > 45, lambda v: v < 40),
}
active = set()


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TELEMETRY, qos=1)


def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    for name, (field, raise_if, clear_if) in RULES.items():
        v = data.get(field)
        if v is None:
            continue
        if name not in active and raise_if(v):
            active.add(name)
            client.publish(ALERTS, json.dumps({"alert": name, "state": "raised", field: v}), qos=1)
        elif name in active and clear_if(v):
            active.discard(name)
            client.publish(ALERTS, json.dumps({"alert": name, "state": "cleared", field: v}), qos=1)


c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="batt-alerts-9902")
c.on_connect = on_connect
c.on_message = on_message
c.connect("broker.connectmq.com", 1883, keepalive=60)
c.loop_forever()

Tout ce qui peut envoyer des notifications, comme Node-RED ou Home Assistant, peut s'abonner au topic d'alertes et les transmettre à votre téléphone ; Jeedom et Domoticz savent aussi communiquer en MQTT. Pour de simples seuils sur une seule valeur, le portail ConnectMQ propose aussi des règles d'alerte sur seuil avec un journal des alertes, ce qui couvre « SoC sous 20 » sans faire tourner votre propre script ; la logique d'hystérésis ci-dessus est ce que vous ajoutez lorsque vous en avez besoin.

Mesurez l'écart entre cellules au repos ou sous faible charge. Pendant une charge ou une décharge importante, et près de la fin de charge, les cellules LiFePO4 divergent naturellement davantage ; une alerte de déséquilibre pendant ces périodes n'indique donc pas forcément une panne.

Suivre le comportement de la batterie dans le temps

Les valeurs en direct répondent à « à quel point la batterie est-elle pleine maintenant » ; l'historique répond à « la batterie est-elle toujours en bonne santé ». Comme le broker ne conserve que le dernier retained message, stockez la télémétrie avec un abonné comme Telegraf écrivant dans InfluxDB, l'enregistreur de Home Assistant ou un petit script écrivant dans SQLite. Trois graphiques sont rapidement rentables :

• Le SoC avec la puissance PV et réseau sur le même axe temporel, qui montre si la batterie est bien dimensionnée pour votre consommation du soir.
• Le SoC minimal quotidien, qui révèle les déficits saisonniers et indique si une réserve est réglée trop haut ou trop bas.
• L'écart entre cellules au repos sur plusieurs mois, l'indicateur de santé à long terme des packs LiFePO4. Une lente tendance à la hausse est le signal pour vérifier les réglages d'équilibrage ou contacter le fournisseur.

Liste de contrôle de dépannage

1. Aucun message de l'appareil GX ? Vérifiez que MQTT on LAN est activé et que vous vous êtes abonné après la connexion.
2. Les valeurs s'arrêtent au bout d'une minute ? Le keepalive n'atteint pas R/<portal-id>/keepalive ; vérifiez l'identifiant de portail découvert.
3. Tensions de cellules absentes ? Votre BMS ne les remonte peut-être pas à Venus OS ; consultez la page de l'appareil batterie sur le GX.
4. Le SoC saute après une charge complète ? Le BMS a resynchronisé son estimation ; c'est normal après des périodes sans charge complète.
5. Les valeurs semblent figées sur le tableau de bord ? Vérifiez le topic de statut retenu ; s'il indique offline, la passerelle s'est arrêtée.

Passer à un broker privé

Les données de batterie révèlent quand vous êtes chez vous et comment votre système est configuré, et le serveur MQTT public gratuit est partagé : utilisez donc un broker privé pour l'installation permanente. Un broker MQTT managé comme ConnectMQ permet à la passerelle sur votre Raspberry Pi d'établir une seule connexion TLS sortante, sans port à ouvrir sur votre box. Créez un broker dans le portail ConnectMQ et modifiez la connexion de destination :

dest = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
dest.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
dest.tls_set()
dest.connect("broker.connectmq.com", 8883, keepalive=60)

Dès lors, les données de la batterie peuvent guider les décisions aux côtés de votre suivi énergétique de la maison et de votre borne de recharge, par exemple pour ne recharger la voiture qu'une fois la batterie au-dessus d'un SoC défini. Pour les options d'hébergement du broker, consultez le guide du broker MQTT cloud, et pour une vue d'ensemble, le guide de l'automatisation IoT.

Surveillez votre batterie avec ConnectMQ

ConnectMQ fournit un broker MQTT sécurisé par TLS avec des identifiants générés, un explorateur de topics en direct et des règles d'alerte sur seuil pour des valeurs comme l'état de charge. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, et connectez votre passerelle batterie en quelques minutes.

Questions fréquentes

Quel est un bon seuil de batterie faible pour une batterie domestique ?

Beaucoup d'utilisateurs de LiFePO4 lèvent une alerte autour de 15 à 20 % de SoC et l'effacent quelques points plus haut. Suivez la recommandation de SoC minimal du fabricant de votre batterie pour l'usage quotidien.

Pourquoi surveiller les tensions de cellules plutôt que seulement l'état de charge ?

Le SoC est une estimation qui peut dériver, tandis que les tensions de cellules sont des mesures directes. Un écart croissant entre la cellule la plus haute et la plus basse est un signe précoce de problème d'équilibrage ou de cellule faible.

Venus OS a-t-il besoin d'Internet pour MQTT ?

Non. MQTT on LAN fonctionne entièrement en local. La passerelle de ce guide lit les données localement et n'établit qu'une connexion sortante vers le broker de votre choix.

Puis-je surveiller une batterie DIY sans matériel Victron ?

Oui. Les cartes BMS JK, JBD et Daly peuvent être lues en Bluetooth ou en UART par un firmware ESP32, comme les composants ESPHome, qui publient directement en MQTT avec la même conception de topics et d'alertes.