[Python] Capteur de température – humidité LCD Xiaomi BLUETOOTH
Publié : 22 nov. 2020, 12:14
Bonjour à tous 
Lors d'une promotion, j'ai compulsivement acheté 2 exemplaires de ce capteur de température & humidité XIAOMI, celui qui ressemble à ça :
Je pensais benoitement pouvoir les intégrer à mon installation domoticz, mais j'ai rapidement déchanté et (finalement) compris que le protocole utilisé est le Bluetooth... Mon serveur refusant obstinément de les reconnaitre, j'ai rapidement compris qu'il me fallait pour la nième fois réinventer l'eau chaude ^^
Après avoir découvert le protocole BLE, j'ai testé les différents framework qu'on peut trouver, et après bien des déconvenues mon choix s'est arrêté sur bleak : https://bleak.readthedocs.io/en/latest/index.html
Ma solution est, sans surprise, de créer un capteur virtuel de température & humidité et qu'un script se charge de le mettre à jour avec les valeurs qu'il lit sur le périphérique.
Voici le fruit de mon travail :
NB :
Lors d'une promotion, j'ai compulsivement acheté 2 exemplaires de ce capteur de température & humidité XIAOMI, celui qui ressemble à ça :
Je pensais benoitement pouvoir les intégrer à mon installation domoticz, mais j'ai rapidement déchanté et (finalement) compris que le protocole utilisé est le Bluetooth... Mon serveur refusant obstinément de les reconnaitre, j'ai rapidement compris qu'il me fallait pour la nième fois réinventer l'eau chaude ^^
Après avoir découvert le protocole BLE, j'ai testé les différents framework qu'on peut trouver, et après bien des déconvenues mon choix s'est arrêté sur bleak : https://bleak.readthedocs.io/en/latest/index.html
Ma solution est, sans surprise, de créer un capteur virtuel de température & humidité et qu'un script se charge de le mettre à jour avec les valeurs qu'il lit sur le périphérique.
Voici le fruit de mon travail :
Code : Tout sélectionner
"""
Domoticz dummy device updater for XIAOMI bluetooth Temp / Hum LCD sensor
----------------
A script getting data from this model of sensor and updating a related domoticz dummy device.
Created on 2020-11-22 by PBC43 <oratoire43+dev@gmail.com>
Based on https://github.com/hbldh/bleak/blob/master/examples/service_explorer.py
"""
import platform
import asyncio
import logging
import urllib.request
import base64
import time
from bleak import BleakClient
DEBUG = False
MAXRETRY = 15
RETRYSLEEP = 10
UPDATEOK = "success"
UPDATEFAILED = "FAILED"
DOMSERVER = "http://localhost:8080"
DOMUSER = "putyourdomoticzusernamehere" # !!!
DOMPASS = "putyourdomoticzuserpasswordhere" # !!!
BASE64STR = base64.encodebytes(('%s:%s' % (DOMUSER, DOMPASS)).encode()).decode().replace('\n', '')
TARGETSERVICE = "226c0000-6476-4566-7562-66734470666d"
TARGETCHARACTERISTIC = "226caa55-6476-4566-7562-66734470666d"
ITEMS = [
{"alias": "Salon", "domid": "putyourdomoticzdeviceidhere", "device": BleakClient("putyourbledeviceaddresshere")}, # !!!
#{"alias": "Chambre", "domid": "234", "device": BleakClient("58:2D:34:3A:1B:3A")}, # for example
]
FAIL = "\033[91m"
ENDC = "\033[0m"
def domrequest (url):
request = urllib.request.Request(url)
request.add_header("Authorization", "Basic %s" % BASE64STR)
response = urllib.request.urlopen(request)
return response.read()
class receivedvalues:
def _decode(self, val):
try:
return(val.decode())
except:
return(val)
def __init__(self):
self.temperature = None
self.humidity = None
self.comfort = None
def put(self, data):
data = self._decode(data)
if "=" in data:
_values = data.split(" ")
self.temperature = float(_values[0].split("=")[1])
self.humidity = float(_values[1].rstrip("\x00").split("=")[1])
if self.humidity < 40:
self.comfort = "2"
elif self.humidity <= 60:
self.comfort = "1"
elif self.humidity > 60:
self.comfort = "3"
def gettempval(sender, data, receiver):
receiver.put(data)
async def run(item):
global UPDATEOK, UPDATEFAILED
alias = item["alias"]
domid = item["domid"]
device = item["device"]
async with device:
x = await device.is_connected()
print(alias, "connected")
service = list(filter(lambda service: service.uuid == TARGETSERVICE, device.services))[0]
characteristic = list(filter(lambda characteristic: characteristic.uuid == TARGETCHARACTERISTIC, service.characteristics))[0]
try:
values = receivedvalues()
await device.start_notify(characteristic.uuid, lambda *args: gettempval(*args, values))
await asyncio.sleep(5.0)
print(alias, "=>", "temperature =", values.temperature, "humidity =", values.humidity, "comfort =", values.comfort)
if(float(values.temperature) and float(values.humidity) and float(values.comfort)):
req = domrequest(
DOMSERVER + "/json.htm?type=command¶m=udevice&idx=" + domid + "&nvalue=0&svalue="
+ str(values.temperature) + ";"
+ str(values.humidity) + ";"
+ str(values.comfort)
)
"""
if "OK" in str(req):
res = "OK"
else:
res = "FAILED"
"""
print(alias, ": domoticz update =>", (UPDATEOK if "OK" in str(req) else UPDATEFAILED))
await device.disconnect()
except Exception as e:
print(e)
index = 0
async def connect_to_device(item, loop):
global index
global MAXRETRY
global RETRYSLEEP
alias = item["alias"]
initialsleep = index * 5
time.sleep(initialsleep)
print("starting", alias, "loop")
ok = False
retry = 0
index = index + 1
while retry < MAXRETRY and not ok:
print("> connect to", alias, retry + 1)
try:
await run(item)
ok = True
print("<", alias, "disconnected")
except Exception as e:
print(alias, "conn. failed", FAIL, (str(e) if DEBUG else ""), ENDC)
if(retry + 1 < MAXRETRY):
time.sleep(RETRYSLEEP)
retry = retry + 1
def lookup(items):
loop = asyncio.get_event_loop()
i = 0
tasks = asyncio.gather(*(connect_to_device(item, loop) for item in ITEMS))
loop.run_until_complete(tasks)
if __name__ == "__main__":
lookup(ITEMS)
- Je débute en python, et à ce titre je n'utilise pas encore les solutions les plus simples
- L'établissement de la liaison est très erratique, soit que le périphérique n'est à cet instant pas visible, soit que la connexion est interrompue pour une raison inconnue. C'est pourquoi j'ai fixé un nombre important d'essais et que je ne lance personnellement mon script que toutes les 15 minutes via crontab
- A réinventer l'eau chaude, quelqu'un finit toujours par vous annoncer qu'un autre l'a fait avant vous, et que votre solution n'est pas propre, pour x raisons. Je ne suis pas revanchard, donc n'hésitez pas à commenter !