init commit

ble.py

@@ -0,0 +1,121 @@
+import asyncio
+from bleak import BleakClient, BleakScanner
+import warnings
+warnings.filterwarnings("ignore")
+
+# 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})")
+            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 = await client.get_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}%")
+            print("")
+            return temp, hum
+
+def sensor_data_query(mac_list):
+    result = []
+    for mac in mac_list:
+        this_result = asyncio.run(read_data(mac))
+        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}")