Pont BLE vers MQTT sur Raspberry Pi avec Python et Bleak

Un pont Raspberry Pi BLE vers MQTT transforme des capteurs Bluetooth bon marché sur pile en appareils connectés que vous pouvez tracer, surveiller et automatiser. Dans ce tutoriel, vous utiliserez la radio Bluetooth intégrée du Pi, la bibliothèque bleak et paho-mqtt pour écouter les annonces des capteurs, décoder la température, l'humidité et le niveau de batterie, puis les publier vers un broker MQTT à un rythme raisonnable. Vous testerez d'abord avec broker.connectmq.com, puis passerez à des identifiants privés.

Pourquoi relier le BLE à MQTT

Les capteurs Bluetooth Low Energy sont petits, peu coûteux et fonctionnent un an ou plus sur une pile bouton, mais leur portée se limite à une ou deux pièces et ils ne peuvent pas atteindre Internet seuls. Un Raspberry Pi doté d'une radio intégrée (Pi 3B et suivants, Zero W et Zero 2 W) peut être placé au centre d'une maison, d'une serre ou d'une chambre froide, entendre tous les capteurs à proximité et transmettre les données en Wi-Fi ou en Ethernet. C'est une brique très pratique pour la maison connectée, et Jeedom ou Domoticz savent eux aussi consommer ces données en MQTT.

La plupart des capteurs diffusent leurs mesures dans des paquets d'annonce (advertisements). C'est pratique : le pont n'a jamais besoin de se connecter au capteur, si bien que des dizaines d'appareils peuvent être lus passivement sans vider leurs piles. La lecture par connexion (GATT) reste possible avec bleak, mais elle est plus lente, consomme davantage d'énergie côté capteur et limite le nombre d'appareils gérables simultanément. Ce guide se concentre sur les annonces, avec une note sur le GATT à la fin.

Ce dont vous avez besoin

La logique de décodage dépend de l'appareil, mais pas la structure du pont. Si vous avez un autre capteur, vous ne modifiez qu'une seule fonction.

Installer bleak et paho-mqtt

Sous Linux, bleak utilise BlueZ via D-Bus : aucun accès root ni paquet système supplémentaire n'est nécessaire sur une installation standard. Vérifiez que la radio est active, puis créez un environnement virtuel :

bluetoothctl show | grep -E "Powered|Name"
sudo rfkill unblock bluetooth
python3 -m venv ~/ble-bridge/.venv
source ~/ble-bridge/.venv/bin/activate
pip install bleak "paho-mqtt>=2.0"

Un scan rapide confirme que bleak voit les appareils :

python -c "import asyncio; from bleak import BleakScanner; print(*asyncio.run(BleakScanner.discover(timeout=8)), sep='\n')"

Décoder les annonces

Le firmware personnalisé des capteurs LYWSD03MMC diffuse les mesures sous forme de service data avec l'UUID Environmental Sensing 0x181A. Deux formats sont courants, et leur longueur permet de les distinguer :

• Format pvvx, 15 octets, little-endian : MAC (6 octets, inversés), température en int16 par pas de 0,01 °C, humidité en uint16 par pas de 0,01 %, tension de batterie en millivolts (uint16), pourcentage de batterie, un compteur de trames et des flags.
• Format ATC1441, 13 octets, big-endian : MAC (6 octets), température en int16 par pas de 0,1 °C, pourcentage d'humidité, pourcentage de batterie, tension de batterie en millivolts (uint16) et un compteur.

Les décoder avec struct est court et facile à tester. Pour d'autres appareils, examinez le dictionnaire manufacturer_data (indexé par identifiant de fabricant) ou service_data (indexé par UUID sous forme de chaîne) de l'annonce et appliquez le format documenté par le fabricant.

Le script du pont Raspberry Pi BLE vers MQTT

Le script exécute un scanner continu avec une fonction de rappel de détection. Chaque annonce provenant d'un capteur connu est décodée et stockée ; une tâche asynchrone distincte publie la dernière mesure de chaque capteur au maximum une fois toutes les 30 secondes. paho-mqtt fait tourner son propre thread réseau via loop_start(), ce qui lui permet de cohabiter proprement avec asyncio.

import asyncio
import json
import os
import struct
import time
import uuid

import paho.mqtt.client as mqtt
from bleak import BleakScanner

BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
USERNAME = os.getenv("MQTT_USERNAME")
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID", f"ble-bridge-{uuid.uuid4().hex[:8]}")
PREFIX = os.getenv("TOPIC_PREFIX", "connectmq/demo/b4n8r2/ble")
PUBLISH_EVERY = 30  # secondes par capteur

ESS_UUID = "0000181a-0000-1000-8000-00805f9b34fb"
latest = {}        # mac -> mesure
last_sent = {}     # mac -> horodatage


def decode_181a(data: bytes):
    if len(data) == 15:  # format personnalisé pvvx
        temp, hum, mv, batt = struct.unpack_from("<hHHB", data, 6)
        return {"temperature": temp / 100, "humidity": hum / 100,
                "battery_mv": mv, "battery": batt}
    if len(data) == 13:  # format ATC1441
        temp, hum, batt, mv = struct.unpack_from(">hBBH", data, 6)
        return {"temperature": temp / 10, "humidity": hum,
                "battery_mv": mv, "battery": batt}
    return None


def on_advertisement(device, adv):
    data = adv.service_data.get(ESS_UUID)
    if not data:
        return
    reading = decode_181a(data)
    if reading:
        reading["rssi"] = adv.rssi
        reading["ts"] = int(time.time())
        latest[device.address.replace(":", "").lower()] = reading


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.publish(f"{PREFIX}/bridge/status", "online", qos=1, retain=True)


def make_client():
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        c.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        c.tls_set()
    c.will_set(f"{PREFIX}/bridge/status", "offline", qos=1, retain=True)
    c.reconnect_delay_set(1, 60)
    c.on_connect = on_connect
    c.connect_async(BROKER, PORT, keepalive=60)
    c.loop_start()
    return c


async def publisher(client):
    while True:
        now = time.time()
        for mac, reading in list(latest.items()):
            if now - last_sent.get(mac, 0) >= PUBLISH_EVERY and client.is_connected():
                client.publish(f"{PREFIX}/{mac}", json.dumps(reading), qos=1, retain=True)
                last_sent[mac] = now
        await asyncio.sleep(1)


async def main():
    client = make_client()
    scanner = BleakScanner(detection_callback=on_advertisement)
    await scanner.start()
    print("scanning...")
    try:
        await publisher(client)
    finally:
        await scanner.stop()
        client.publish(f"{PREFIX}/bridge/status", "offline", qos=1, retain=True)
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    asyncio.run(main())

Abonnez-vous depuis votre ordinateur portable pour voir arriver les mesures :

mosquitto_sub -h broker.connectmq.com -p 1883 -i ble-viewer-b4n8r2 -t "connectmq/demo/b4n8r2/ble/#" -v

Chaque capteur obtient son propre topic nommé d'après son adresse MAC, et la mesure est retenue : un tableau de bord qui se connecte plus tard voit immédiatement la valeur la plus récente. Si vous préférez des noms parlants, gardez un petit dictionnaire qui associe les adresses MAC aux noms des pièces et utilisez-les dans le topic.

Les choix de conception qui comptent

Limiter la fréquence par appareil. Les capteurs peuvent émettre une annonce toutes les quelques secondes. Les publier toutes gaspille de la bande passante et, sur le serveur MQTT public gratuit, atteint vite la limite de 10 messages par seconde et par client dès que vous avez beaucoup de capteurs. Une fenêtre de 30 secondes par capteur suffit largement pour le climat d'une pièce.

Retenir le dernier état, pas l'historique. Les messages retenus donnent instantanément la valeur courante aux nouveaux abonnés. L'historique a sa place dans une base de données : un abonné qui écrit dans SQLite ou InfluxDB, comme celui du guide Grafana et InfluxDB sur Raspberry Pi, est le bon endroit pour cela.

Inclure le RSSI. La puissance du signal vous indique quand un capteur est sur le point de sortir de la zone de portée ou quand le pont doit être déplacé. C'est aussi un moyen simple de détecter un capteur défaillant avant que sa mesure de batterie ne chute.

Surveiller le silence. Un capteur qui cesse d'émettre devient invisible pour le pont. Ajoutez le champ ts aux tableaux de bord, ou déclenchez une alerte lorsqu'il devient trop ancien. Les règles d'alerte de seuil de ConnectMQ fonctionnent sur des valeurs numériques : un abonné qui publie « secondes depuis la dernière mesure » pour chaque capteur vous offre une alarme d'obsolescence simple.

Dépannage

1. Aucun appareil trouvé : lancez bluetoothctl power on et rfkill list. Sur un Pi où le Wi-Fi et le Bluetooth partagent la même puce, un trafic Wi-Fi 2,4 GHz intense peut réduire la réception BLE ; l'Ethernet ou le Wi-Fi 5 GHz aident.
2. Appareils trouvés mais pas de service data : le capteur utilise le firmware d'origine, qui chiffre ses annonces, ou un autre format. Affichez adv dans la fonction de rappel pour inspecter les champs bruts.
3. org.bluez.Error.InProgress : un autre processus scanne déjà. Arrêtez les autres scanners, ou ne lancez qu'un seul pont par adaptateur.
4. Les mesures s'arrêtent au bout de quelques heures : BlueZ a parfois besoin d'être redémarré sur les noyaux anciens. Exécuter le pont comme service systemd avec Restart=always règle ce cas, comme expliqué dans lancer un script Python MQTT au démarrage.
5. Portée faible : une rallonge USB pour éloigner le dongle du boîtier métallique du Pi et des ports USB 3 peut faire une différence notable, car l'USB 3 est connu pour provoquer des interférences à 2,4 GHz.

Lire des caractéristiques GATT

Certains appareils n'exposent leurs données que via une connexion. bleak gère ce cas avec BleakClient :

from bleak import BleakClient

async def read_battery(address):
    async with BleakClient(address) as client:
        value = await client.read_gatt_char("00002a19-0000-1000-8000-00805f9b34fb")
        return value[0]

Les connexions sont coûteuses : lisez les valeurs GATT à un rythme lent et arrêtez le scan pendant la connexion si votre adaptateur peine à gérer les deux à la fois. La documentation de bleak décrit les notifications, qui sont la manière efficace de recevoir en continu les données d'un appareil connecté.

Passer en production

Une fois le pont fonctionnel, quittez le broker public, lisible par tout le monde. Créez un broker privé sur ConnectMQ, définissez MQTT_HOST, MQTT_PORT=8883, MQTT_USERNAME, MQTT_PASSWORD et MQTT_CLIENT_ID à partir du portail, et changez le préfixe de topic pour quelque chose comme home/ble. Le pont n'a besoin d'aucun port entrant, seulement d'une connexion TLS sortante : il fonctionne donc derrière n'importe quelle box ou modem cellulaire. Un broker MQTT managé signifie aussi qu'il n'y a aucun processus de broker à maintenir sur le Pi.

Si MQTT sur le Pi est nouveau pour vous, le tutoriel MQTT sur Raspberry Pi en présente les bases, et l'article sur la passerelle IoT Raspberry Pi montre comment combiner le BLE avec d'autres protocoles sur un seul appareil.

Connectez vos capteurs BLE à ConnectMQ

Inscrivez-vous sur app.connectmq.com à l'offre Serverless gratuite, sans carte bancaire, et vous disposerez d'identifiants de broker en quelques minutes. Pointez-y le pont et ouvrez l'explorateur de topics en direct pour voir chaque capteur apparaître sous son propre topic.

Questions fréquentes

Combien de capteurs BLE un Raspberry Pi peut-il gérer ?

Avec un scan passif des annonces, des dizaines de capteurs à portée radio sont courants, car le Pi se contente d'écouter. La portée et les murs vous limiteront généralement avant le processeur.

Faut-il appairer les capteurs ?

Non. Les capteurs qui fonctionnent par annonces diffusent à quiconque écoute : aucun appairage ni connexion n'est nécessaire. Les annonces chiffrées font exception et requièrent la clé de l'appareil.

Peut-on faire tourner ce pont sur un Raspberry Pi Zero 2 W ?

Oui. Le pont consomme peu de processeur et de mémoire, et le Zero 2 W intègre Bluetooth et Wi-Fi, ce qui en fait un bon pont économique par pièce ou par bâtiment.

Faut-il utiliser la QoS 0 ou la QoS 1 pour les mesures de capteurs ?

La QoS 1 est un bon choix par défaut pour les mesures que vous voulez voir livrées au moins une fois. Pour des valeurs très fréquentes où la mesure suivante remplace la précédente, la QoS 0 convient.