Bonjour,
Bon a priori, il existe un script qui fonctionne pour V1 et V2, j'ai bossé dessus ce long week-end mais je n'ai pas réussi à les intégrer, pourtant, il semblerait que ca fonctionne a lire les commentaires.
domoticz_mijia_v2.py
Code : Tout sélectionner
import urllib.request
import base64
import time
import btlewrap
from btlewrap.base import BluetoothBackendException
from mijia.mitemp2_bt_poller import MiTemp2BtPoller, \
MI_HUMIDITY, MI_TEMPERATURE, MI_BATTERY
# Settings for the domoticz server
# Forum see: http://domoticz.com/forum/viewtopic.php?f=56&t=13306&hilit=mi+flora&start=20#p105255
domoticzserver = "127.0.0.1:8000"
domoticzusername = ""
domoticzpassword = ""
# So id devices use: sudo hcitool lescan
# Sensor IDs
# Create virtual sensors in dummy hardware
# type temperature & humidity
try:
import bluepy.btle # noqa: F401 pylint: disable=unused-import
BACKEND = btlewrap.BluepyBackend
except ImportError:
BACKEND = btlewrap.GatttoolBackend
base64string = base64.encodestring(('%s:%s' % (domoticzusername, domoticzpassword)).encode()).decode().replace('\n', '')
def domoticzrequest (url):
print(url)
request = urllib.request.Request(url)
request.add_header("Authorization", "Basic %s" % base64string)
response = urllib.request.urlopen(request)
return response.read()
def update(address,idx_temp):
poller = MiTemp2BtPoller(address, BACKEND)
loop = 0
try:
temp = poller.parameter_value(MI_TEMPERATURE)
except:
temp = "Not set"
while loop < 2 and temp == "Not set":
print("Error reading value retry after 5 seconds...\n")
time.sleep(5)
poller = MiTemp2BtPoller(address)
loop += 1
try:
temp = poller.parameter_value(MI_TEMPERATURE)
except:
temp = "Not set"
if temp == "Not set":
print("Error reading value\n")
return
global domoticzserver
print("Mi Sensor: " + address)
print("Firmware: {}".format(poller.firmware_version()))
print("Name: {}".format(poller.name()))
print("Temperature: {}°C".format(poller.parameter_value(MI_TEMPERATURE)))
print("Humidity: {}%".format(poller.parameter_value(MI_HUMIDITY)))
print("Battery: {}%".format(poller.parameter_value(MI_BATTERY)))
val_bat = "{}".format(poller.parameter_value(MI_BATTERY))
# Update temp
#val_temp = "{}".format(poller.parameter_value(MI_TEMPERATURE))
#domoticzrequest("http://" + domoticzserver + "/json.htm?type=command¶m=udevice&idx=" + idx_temp + "&nvalue=0&svalue=" + val_temp + "&battery=" + val_bat)
# Update humidity
#val_hum = "{}".format(poller.parameter_value(MI_HUMIDITY))
#domoticzrequest("http://" + domoticzserver + "/json.htm?type=command¶m=udevice&idx=" + idx_hum + "&svalue=" + val_hum + "&battery=" + val_bat)
#/json.htm?type=command¶m=udevice&idx=IDX&nvalue=0&svalue=TEMP;HUM;HUM_STAT
val_temp = "{}".format(poller.parameter_value(MI_TEMPERATURE))
val_hum = "{}".format(poller.parameter_value(MI_HUMIDITY))
val_comfort = "0"
if float(val_hum) < 40:
val_comfort = "2"
elif float(val_hum) <= 70:
val_comfort = "1"
elif float(val_hum) > 70:
val_comfort = "3"
#domoticzrequest("http://" + domoticzserver + "/json.htm?type=command¶m=udevice&idx=" + idx_temp + "&nvalue=0&svalue=" + val_temp + ";" + val_hum + ";"+ val_comfort + "&battery=" + val_bat)
#print("\n1: updating")
update("A4:C1:38:A1:D5:92","752")
#update("4C:65:A8:D0:26:D2","753")
#update("4C:65:A8:D0:57:2A","754")
mijia/mitemp2_bt_poller.py
Code : Tout sélectionner
""""
Read data from Mi Temp environmental (Temp and humidity) sensor.
"""
from datetime import datetime, timedelta
import logging
from threading import Lock
from btlewrap.base import BluetoothInterface, BluetoothBackendException
_HANDLE_READ_BATTERY_LEVEL = 0x001B
_HANDLE_READ_FIRMWARE_VERSION = 0x0012
_HANDLE_READ_NAME = 0x03
_HANDLE_READ_WRITE_SENSOR_DATA = 0x001C
MI_TEMPERATURE = "temperature"
MI_HUMIDITY = "humidity"
MI_BATTERY = "battery"
_LOGGER = logging.getLogger(__name__)
class MiTemp2BtPoller:
""""
A class to read data from Mi Temp plant sensors.
"""
def __init__(self, mac, backend, cache_timeout=600, retries=3, adapter='hci0'):
"""
Initialize a Mi Temp Poller for the given MAC address.
"""
self._mac = mac
self._bt_interface = BluetoothInterface(backend, adapter=adapter)
self._cache = None
self._cache_timeout = timedelta(seconds=cache_timeout)
self._last_read = None
self._fw_last_read = None
self.retries = retries
self.ble_timeout = 10
self.lock = Lock()
self._firmware_version = None
self.battery = None
def name(self):
"""Return the name of the sensor."""
with self._bt_interface.connect(self._mac) as connection:
name = connection.read_handle(_HANDLE_READ_NAME) # pylint: disable=no-member
if not name:
raise BluetoothBackendException("Could not read NAME using handle %s"
" from Mi Temp sensor %s" % (hex(_HANDLE_READ_NAME), self._mac))
return ''.join(chr(n) for n in name)
def fill_cache(self):
"""Fill the cache with new data from the sensor."""
_LOGGER.debug('Filling cache with new sensor data.')
try:
self.firmware_version()
except BluetoothBackendException:
# If a sensor doesn't work, wait 5 minutes before retrying
self._last_read = datetime.now() - self._cache_timeout + \
timedelta(seconds=300)
raise
with self._bt_interface.connect(self._mac) as connection:
try:
connection.wait_for_notification(_HANDLE_READ_WRITE_SENSOR_DATA, self,
self.ble_timeout) # pylint: disable=no-member
# If a sensor doesn't work, wait 5 minutes before retrying
except BluetoothBackendException:
self._last_read = datetime.now() - self._cache_timeout + \
timedelta(seconds=300)
return
def battery_level(self):
"""Return the battery level.
The battery level is updated when reading the firmware version. This
is done only once every 24h
"""
self.firmware_version()
return self.battery
def firmware_version(self):
"""Return the firmware version."""
if (self._firmware_version is None) or \
(datetime.now() - timedelta(hours=24) > self._fw_last_read):
self._fw_last_read = datetime.now()
with self._bt_interface.connect(self._mac) as connection:
res_firmware = connection.read_handle(_HANDLE_READ_FIRMWARE_VERSION) # pylint: disable=no-member
_LOGGER.debug('Received result for handle %s: %s',
_HANDLE_READ_FIRMWARE_VERSION, res_firmware)
res_battery = connection.read_handle(_HANDLE_READ_BATTERY_LEVEL) # pylint: disable=no-member
_LOGGER.debug('Received result for handle %s: %d',
_HANDLE_READ_BATTERY_LEVEL, res_battery)
if res_firmware is None:
self._firmware_version = None
else:
self._firmware_version = res_firmware.decode("utf-8")
if res_battery is None:
self.battery = 0
else:
self.battery = int(ord(res_battery))
return self._firmware_version
def parameter_value(self, parameter, read_cached=True):
"""Return a value of one of the monitored paramaters.
This method will try to retrieve the data from cache and only
request it by bluetooth if no cached value is stored or the cache is
expired.
This behaviour can be overwritten by the "read_cached" parameter.
"""
# Special handling for battery attribute
if parameter == MI_BATTERY:
return self.battery_level()
# Use the lock to make sure the cache isn't updated multiple times
with self.lock:
if (read_cached is False) or \
(self._last_read is None) or \
(datetime.now() - self._cache_timeout > self._last_read):
self.fill_cache()
else:
_LOGGER.debug("Using cache (%s < %s)",
datetime.now() - self._last_read,
self._cache_timeout)
if self.cache_available():
return self._parse_data()[parameter]
raise BluetoothBackendException("Could not read data from Mi Temp sensor %s" % self._mac)
def _check_data(self):
"""Ensure that the data in the cache is valid.
If it's invalid, the cache is wiped.
"""
if not self.cache_available():
return
parsed = self._parse_data()
_LOGGER.debug('Received new data from sensor: Temp=%.1f, Humidity=%.1f',
parsed[MI_TEMPERATURE], parsed[MI_HUMIDITY])
if parsed[MI_HUMIDITY] > 100: # humidity over 100 procent
self.clear_cache()
return
def clear_cache(self):
"""Manually force the cache to be cleared."""
self._cache = None
self._last_read = None
def cache_available(self):
"""Check if there is data in the cache."""
return self._cache is not None
def _parse_data(self):
data = self._cache
res = dict()
res[MI_TEMPERATURE] = round(int.from_bytes([data[0], data[1]], "little")/100.0, 1)
res[MI_HUMIDITY] = int.from_bytes([data[2]], "little")
return res
@staticmethod
def _format_bytes(raw_data):
"""Prettyprint a byte array."""
if raw_data is None:
return 'None'
return ' '.join([format(c, "02x") for c in raw_data]).upper()
def handleNotification(self, handle, raw_data): # pylint: disable=unused-argument,invalid-name
""" gets called by the bluepy backend when using wait_for_notification
"""
if raw_data is None:
return
self._cache = raw_data
self._check_data()
if self.cache_available():
self._last_read = datetime.now()
else:
# If a sensor doesn't work, wait 5 minutes before retrying
self._last_read = datetime.now() - self._cache_timeout + \
timedelta(seconds=300)
Si un expert peu y jeter un oeil, on ne doit pas être loin.
Il existe aussi une version pour V1 + V2 sur la même page
https://github.com/pFenners/mijia-senso ... cz/pull/10
je ne désespère pas de pouvoir les intégrer
Bonne journée
TG76
