125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
import asyncio
|
|
from bleak import BleakClient, BleakScanner
|
|
import warnings, logging
|
|
warnings.filterwarnings("ignore")
|
|
logging.getLogger('apscheduler').setLevel(logging.ERROR)
|
|
|
|
# Mi sensor UUIDs
|
|
MI_SERVICE_UUID = "ebe0ccb0-7a0a-4b0c-8a1a-6ff2997da3a6"
|
|
TEMP_CHAR_UUID = "ebe0ccc1-7a0a-4b0c-8a1a-6ff2997da3a6"
|
|
|
|
# scanner constants
|
|
full_timeout = 60
|
|
attempts = 6
|
|
|
|
async def read_data(mac_address):
|
|
|
|
# Scan for the device
|
|
ratio = full_timeout / attempts
|
|
fail_count = full_timeout / ratio
|
|
device = None
|
|
print(f"Beginning read_data for {mac_address} - {full_timeout} timeout - {ratio} ratio - please stand by...")
|
|
while fail_count > 0 and not device:
|
|
# print(f"Starting discover loop {fail_count}")
|
|
devices = await BleakScanner.discover(timeout=ratio)
|
|
# print("BleakScanner discover timeout pass")
|
|
device = next((d for d in devices if d.address.lower() == mac_address.lower()), None)
|
|
if device:
|
|
# print(f"BleakScanner device variable loaded - {device.name}")
|
|
print(f".. Found device: {device.name} ({device.address}) - connecting...")
|
|
fail_count = 0
|
|
# else:
|
|
# print("BleakScanner device not loaded yet")
|
|
fail_count = fail_count - 1
|
|
|
|
if not device:
|
|
print(f"!! Device with MAC {mac_address} not found")
|
|
return None
|
|
|
|
# Connect to the device
|
|
mi_service = None
|
|
async with BleakClient(device) as client:
|
|
|
|
success = False
|
|
# Check if Mi service is available
|
|
|
|
#print(".. Waiting on client to connect...")
|
|
while not client.is_connected:
|
|
await asyncio.sleep(0.1)
|
|
while not success:
|
|
try:
|
|
#print(".. Client connected, getting services")
|
|
services = client.services
|
|
#print(".. Services acquired, loading uuid matches")
|
|
mi_service = next((s for s in services if s.uuid.lower() == MI_SERVICE_UUID.lower()), None)
|
|
success = True
|
|
#print("uuid matches loaded")
|
|
except Exception as e:
|
|
print(f"Mi service unavailable, infinite loop looping RIP - {e}")
|
|
|
|
# Read temperature data
|
|
sensor_data = await client.read_gatt_char(TEMP_CHAR_UUID)
|
|
|
|
# Parse the temperature value (same logic as in JavaScript)
|
|
sign = sensor_data[1] & (1 << 7)
|
|
temp = ((sensor_data[1] & 0x7F) << 8 | sensor_data[0])
|
|
if sign:
|
|
temp = temp - 32767
|
|
temp = temp / 100
|
|
hum = sensor_data[2]
|
|
|
|
print(f"-- Temperature: {temp:.1f}°C, Humidity: {hum}%")
|
|
return temp, hum
|
|
|
|
def sensor_data_query(mac_list):
|
|
result = []
|
|
for mac in mac_list:
|
|
try:
|
|
this_result = asyncio.run(read_data(mac))
|
|
except Exception as e:
|
|
print(f"Error reading BLE data - {e}")
|
|
result.append({"name": mac, "temp": this_result[0], "humidity": this_result[1]})
|
|
print(f"Data query complete - returning result")
|
|
return result
|
|
|
|
def test():
|
|
test_result = False
|
|
sensors = []
|
|
mac_list = [
|
|
"A4:C1:38:11:F4:2A",
|
|
"A4:C1:38:24:15:B0",
|
|
"A4:C1:38:D7:8D:26"
|
|
]
|
|
|
|
print("Starting BLE test...")
|
|
print(" Scanning MACs:")
|
|
for mac in mac_list:
|
|
print(f" {mac}")
|
|
test_list = mac_list
|
|
while not test_result:
|
|
print(" Test loop start. Initializing run...")
|
|
test_run_results = sensor_data_query(test_list)
|
|
for test_run_result in test_run_results:
|
|
sensors.append(test_run_result)
|
|
passed_list = [device['name'] for device in sensors]
|
|
print(f" Passed list - {passed_list}")
|
|
if len(mac_list) == len(passed_list):
|
|
print(" Test run passed, yee-haw")
|
|
test_result = True
|
|
else:
|
|
print(f" Test run incomplete/failed, re-running, updating test list")
|
|
mac_list_new = []
|
|
for mac in mac_list:
|
|
if mac not in passed_list:
|
|
mac_list_new.append(mac)
|
|
test_list = mac_list_new
|
|
print(f" Re-do list - {test_list}")
|
|
|
|
if __name__ == "__main__":
|
|
#print("Welp, this will keep going forever. Have fun :)")
|
|
#while True:
|
|
#try:
|
|
test()
|
|
#except Exception as e:
|
|
# print(f"Test failed brah - {e}")
|