lauchmelder/xiaomi_mi_scale
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

0.1.1

Browse source
This commit is contained in:
lolouk44 2020-05-27 23:11:52 +01:00
parent 22dbb9e6d1
commit c47614f196
1 changed files with 141 additions and 105 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

246
src/Xiaomi_Scale.py
View file

@ -22,7 +22,9 @@ import binascii
import time import time
import os import os
import sys import sys
import subprocess
from bluepy import btle from bluepy import btle
from bluepy.btle import Scanner, BTLEDisconnectError, BTLEManagementError, DefaultDelegate
import paho.mqtt.client as mqtt import paho.mqtt.client as mqtt
from datetime import datetime from datetime import datetime
@ -38,6 +40,7 @@ MQTT_TIMEOUT = int(os.getenv('MQTT_TIMEOUT', 60))
MQTT_PREFIX = os.getenv('MQTT_PREFIX', 'miscale') MQTT_PREFIX = os.getenv('MQTT_PREFIX', 'miscale')
TIME_INTERVAL = int(os.getenv('TIME_INTERVAL', 30)) TIME_INTERVAL = int(os.getenv('TIME_INTERVAL', 30))
OLD_MEASURE = '' OLD_MEASURE = ''
MQTT_CONNECTED = False
# User Variables... # User Variables...
USER1_GT = int(os.getenv('USER1_GT', '70')) # If the weight is greater than this number, we'll assume that we're weighing User #1 USER1_GT = int(os.getenv('USER1_GT', '70')) # If the weight is greater than this number, we'll assume that we're weighing User #1
@ -59,125 +62,158 @@ USER3_DOB = os.getenv('USER3_DOB', '1988-01-01') # DOB (in yyyy-mm-dd format)
class ScanProcessor(): class ScanProcessor():
def GetAge(self, d1): def GetAge(self, d1):
d1 = datetime.strptime(d1, "%Y-%m-%d") d1 = datetime.strptime(d1, "%Y-%m-%d")
d2 = datetime.strptime(datetime.today().strftime('%Y-%m-%d'),'%Y-%m-%d') d2 = datetime.strptime(datetime.today().strftime('%Y-%m-%d'),'%Y-%m-%d')
return abs((d2 - d1).days)/365 return abs((d2 - d1).days)/365
def __init__(self): def __init__(self):
self.mqtt_client = None global MQTT_CONNECTED
self.connected = False DefaultDelegate.__init__(self)
self._start_client() if not MQTT_CONNECTED:
self.mqtt_client = None
self._start_client()
def handleDiscovery(self, dev, isNewDev, isNewData): def handleDiscovery(self, dev, isNewDev, isNewData):
global OLD_MEASURE global OLD_MEASURE
if dev.addr == MISCALE_MAC.lower() and isNewDev: if dev.addr == MISCALE_MAC.lower() and isNewDev:
for (sdid, desc, data) in dev.getScanData(): for (sdid, desc, data) in dev.getScanData():
### Xiaomi V1 Scale ### ### Xiaomi V1 Scale ###
if data.startswith('1d18') and sdid == 22: if data.startswith('1d18') and sdid == 22:
measunit = data[4:6] measunit = data[4:6]
measured = int((data[8:10] + data[6:8]), 16) * 0.01 measured = int((data[8:10] + data[6:8]), 16) * 0.01
unit = '' unit = ''
if measunit.startswith(('03', 'b3')): unit = 'lbs' if measunit.startswith(('03', 'b3')): unit = 'lbs'
if measunit.startswith(('12', 'b2')): unit = 'jin' if measunit.startswith(('12', 'b2')): unit = 'jin'
if measunit.startswith(('22', 'a2')): unit = 'kg' ; measured = measured / 2 if measunit.startswith(('22', 'a2')): unit = 'kg' ; measured = measured / 2
if unit: if unit:
if OLD_MEASURE != round(measured, 2): if OLD_MEASURE != round(measured, 2):
print('') print('')
self._publish(round(measured, 2), unit, str(datetime.today().strftime('%Y-%m-%d-%H:%M:%S')), "", "") self._publish(round(measured, 2), unit, str(datetime.today().strftime('%Y-%m-%d-%H:%M:%S')), "", "")
OLD_MEASURE = round(measured, 2) OLD_MEASURE = round(measured, 2)
### Xiaomi V2 Scale ### ### Xiaomi V2 Scale ###
if data.startswith('1b18') and sdid == 22: if data.startswith('1b18') and sdid == 22:
data2 = bytes.fromhex(data[4:]) data2 = bytes.fromhex(data[4:])
ctrlByte1 = data2[1] ctrlByte1 = data2[1]
isStabilized = ctrlByte1 & (1<<5) isStabilized = ctrlByte1 & (1<<5)
hasImpedance = ctrlByte1 & (1<<1) hasImpedance = ctrlByte1 & (1<<1)
measunit = data[4:6] measunit = data[4:6]
measured = int((data[28:30] + data[26:28]), 16) * 0.01 measured = int((data[28:30] + data[26:28]), 16) * 0.01
unit = '' unit = ''
if measunit == "03": unit = 'lbs' if measunit == "03": unit = 'lbs'
if measunit == "02": unit = 'kg' ; measured = measured / 2 if measunit == "02": unit = 'kg' ; measured = measured / 2
#mitdatetime = datetime.strptime(str(int((data[10:12] + data[8:10]), 16)) + " " + str(int((data[12:14]), 16)) +" "+ str(int((data[14:16]), 16)) +" "+ str(int((data[16:18]), 16)) +" "+ str(int((data[18:20]), 16)) +" "+ str(int((data[20:22]), 16)), "%Y %m %d %H %M %S") #mitdatetime = datetime.strptime(str(int((data[10:12] + data[8:10]), 16)) + " " + str(int((data[12:14]), 16)) +" "+ str(int((data[14:16]), 16)) +" "+ str(int((data[16:18]), 16)) +" "+ str(int((data[18:20]), 16)) +" "+ str(int((data[20:22]), 16)), "%Y %m %d %H %M %S")
miimpedance = str(int((data[24:26] + data[22:24]), 16)) miimpedance = str(int((data[24:26] + data[22:24]), 16))
if unit and isStabilized: if unit and isStabilized:
if OLD_MEASURE != round(measured, 2) + int(miimpedance): if OLD_MEASURE != round(measured, 2) + int(miimpedance):
print('') print('')
self._publish(round(measured, 2), unit, str(datetime.today().strftime('%Y-%m-%d-%H:%M:%S')), hasImpedance, miimpedance) self._publish(round(measured, 2), unit, str(datetime.today().strftime('%Y-%m-%d-%H:%M:%S')), hasImpedance, miimpedance)
OLD_MEASURE = round(measured, 2) + int(miimpedance) OLD_MEASURE = round(measured, 2) + int(miimpedance)
if not dev.scanData: if not dev.scanData:
print ('\t(no data)') print ('\t(no data)')
def _start_client(self): def _start_client(self):
self.mqtt_client = mqtt.Client() global MQTT_CONNECTED
self.mqtt_client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD) self.mqtt_client = mqtt.Client()
self.mqtt_client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
def _on_connect(client, _, flags, return_code): def _on_connect(client, _, flags, return_code):
self.connected = True global MQTT_CONNECTED
sys.stdout.write("MQTT connection: %s\n" % mqtt.connack_string(return_code)) MQTT_CONNECTED = True
loop_flag = 0
sys.stdout.write("MQTT connection: %s\n" % mqtt.connack_string(return_code))
self.mqtt_client.on_connect = _on_connect self.mqtt_client.on_connect = _on_connect
self.mqtt_client.connect(MQTT_HOST, MQTT_PORT, MQTT_TIMEOUT) self.mqtt_client.connect(MQTT_HOST, MQTT_PORT, MQTT_TIMEOUT)
self.mqtt_client.loop_start() self.mqtt_client.loop_start()
def _publish(self, weight, unit, mitdatetime, hasImpedance, miimpedance): while not MQTT_CONNECTED: # wait for MQTT connecting Ack
if not self.connected: time.sleep(.01)
sys.stderr.write('Not connected to MQTT server\n')
exit()
if int(weight) > USER1_GT:
user = USER1_NAME
height = USER1_HEIGHT
age = self.GetAge(USER1_DOB)
sex = USER1_SEX
elif int(weight) < USER2_LT:
user = USER2_NAME
height = USER2_HEIGHT
age = self.GetAge(USER2_DOB)
sex = USER2_SEX
else:
user = USER3_NAME
height = USER3_HEIGHT
age = self.GetAge(USER3_DOB)
sex = USER3_SEX
lib = Xiaomi_Scale_Body_Metrics.bodyMetrics(weight, height, age, sex, 0)
message = '{'
message += '"Weight":"' + "{:.2f}".format(weight) + '"'
message += ',"BMI":"' + "{:.2f}".format(lib.getBMI()) + '"'
message += ',"Basal Metabolism":"' + "{:.2f}".format(lib.getBMR()) + '"'
message += ',"Visceral Fat":"' + "{:.2f}".format(lib.getVisceralFat()) + '"'
if hasImpedance: def _publish(self, weight, unit, mitdatetime, hasImpedance, miimpedance):
lib = Xiaomi_Scale_Body_Metrics.bodyMetrics(weight, height, age, sex, int(miimpedance)) global MQTT_CONNECTED
message += ',"Lean Body Mass":"' + "{:.2f}".format(lib.getLBMCoefficient()) + '"' if not MQTT_CONNECTED:
message += ',"Body Fat":"' + "{:.2f}".format(lib.getFatPercentage()) + '"' sys.stderr.write('Not connected to MQTT server\n')
message += ',"Water":"' + "{:.2f}".format(lib.getWaterPercentage()) + '"' exit()
message += ',"Bone Mass":"' + "{:.2f}".format(lib.getBoneMass()) + '"' if int(weight) > USER1_GT:
message += ',"Muscle Mass":"' + "{:.2f}".format(lib.getMuscleMass()) + '"' user = USER1_NAME
message += ',"Protein":"' + "{:.2f}".format(lib.getProteinPercentage()) + '"' height = USER1_HEIGHT
#message += ',"Body Type":"' + str(lib.getBodyTypeScale(getBodyType())) + '"' age = self.GetAge(USER1_DOB)
#message += ',"Metabolic Age":"' + str(lib.getMetabolicAge()) + '"' sex = USER1_SEX
elif int(weight) < USER2_LT:
user = USER2_NAME
height = USER2_HEIGHT
age = self.GetAge(USER2_DOB)
sex = USER2_SEX
else:
user = USER3_NAME
height = USER3_HEIGHT
age = self.GetAge(USER3_DOB)
sex = USER3_SEX
lib = Xiaomi_Scale_Body_Metrics.bodyMetrics(weight, height, age, sex, 0)
message = '{'
message += '"Weight":"' + "{:.2f}".format(weight) + '"'
message += ',"BMI":"' + "{:.2f}".format(lib.getBMI()) + '"'
message += ',"Basal Metabolism":"' + "{:.2f}".format(lib.getBMR()) + '"'
message += ',"Visceral Fat":"' + "{:.2f}".format(lib.getVisceralFat()) + '"'
message += ',"TimeStamp":"' + mitdatetime + '"' if hasImpedance:
message += '}' lib = Xiaomi_Scale_Body_Metrics.bodyMetrics(weight, height, age, sex, int(miimpedance))
self.mqtt_client.publish(MQTT_PREFIX + '/' + user + '/weight', message, qos=1, retain=True) bodyscale = ['Obese', 'Overweight', 'Thick-set', 'Lack-exerscise', 'Balanced', 'Balanced-muscular', 'Skinny', 'Balanced-skinny', 'Skinny-muscular']
sys.stdout.write('Sent data to topic %s: %s' % (MQTT_PREFIX + '/' + user + '/weight', message + '\n')) message += ',"Lean Body Mass":"' + "{:.2f}".format(lib.getLBMCoefficient()) + '"'
message += ',"Body Fat":"' + "{:.2f}".format(lib.getFatPercentage()) + '"'
message += ',"Water":"' + "{:.2f}".format(lib.getWaterPercentage()) + '"'
message += ',"Bone Mass":"' + "{:.2f}".format(lib.getBoneMass()) + '"'
message += ',"Muscle Mass":"' + "{:.2f}".format(lib.getMuscleMass()) + '"'
message += ',"Protein":"' + "{:.2f}".format(lib.getProteinPercentage()) + '"'
message += ',"Body Type":"' + str(bodyscale[lib.getBodyType()]) + '"'
message += ',"Metabolic Age":"' + "{:.0f}".format(lib.getMetabolicAge()) + '"'
message += ',"TimeStamp":"' + mitdatetime + '"'
message += '}'
try:
self.mqtt_client.publish(MQTT_PREFIX + '/' + user + '/weight', message, qos=1, retain=True)
sys.stdout.write('Sent data to topic %s: %s' % (MQTT_PREFIX + '/' + user + '/weight', message + '\n'))
except:
sys.stdout.write('Could not publish to MQTT, Disconnecting...\n')
MQTT_CONNECTED = False
self.mqtt_client.disconnect()
pass
def main(): def main():
sys.stdout.write(' \n') sys.stdout.write(' \n')
sys.stdout.write('-------------------------------------\n') sys.stdout.write('-------------------------------------\n')
sys.stdout.write('Starting Xiaomi mi Scale...\n') sys.stdout.write('Starting Xiaomi mi Scale...\n')
scanner = btle.Scanner().withDelegate(ScanProcessor()) BluetoothFailCounter = 0
while True: while True:
try: try:
scanner.scan(5, passive=True) # Adding passive=True to try and fix issues on RPi devices scanner = btle.Scanner().withDelegate(ScanProcessor())
except: scanner.scan(5) # Adding passive=True to try and fix issues on RPi devices
sys.stderr.write("Error while running the script, continuing. If you see this message too often/constantly there is probably a real issue...\n") except BTLEDisconnectError as error:
pass sys.stderr.write(f"btle disconnected: {error}\n")
time.sleep(TIME_INTERVAL) pass
except BTLEManagementError as error:
sys.stderr.write(f"Bluetooth connection error: {error}\n")
if BluetoothFailCounter >= 4:
sys.stderr.write(f"5+ Bluetooth connection errors. Resetting Bluetooth...\n")
cmd = 'hciconfig hci0 reset'
ps = subprocess.Popen(cmd, shell=True)
time.sleep(30)
BluetoothFailCounter = 0
else:
BluetoothFailCounter+=1
pass
except:
sys.stderr.write("Error while running the script, continuing...\n")
pass
else:
BluetoothFailCounter = 0
time.sleep(TIME_INTERVAL)
if __name__ == "__main__": if __name__ == "__main__":
main() main()