Merge pull request #1259 from frickeo/enhancement-913_bleak-instead-b…

…luepy enhancement #913: Updated 'get_beacon_key.py' to use bleak instead of bluepy
custom-components · Nov 9, 2023 · c002ae9 · c002ae9
2 parents ad2e45c + 478d3f8
commit c002ae9
Show file tree

Hide file tree

Showing 2 changed files with 65 additions and 33 deletions.
diff --git a/custom_components/ble_monitor/ble_parser/get_beacon_key.py b/custom_components/ble_monitor/ble_parser/get_beacon_key.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 
 # Usage:
-#   pip3 install bluepy
+#   pip3 install bleak asyncio
 #   python3 get_beacon_key.py <MAC> <PRODUCT_ID>
 #
 # List of PRODUCT_ID:
@@ -14,20 +14,24 @@
 # Example:
 #   python3 get_beacon_key.py AB:CD:EF:12:34:56 950
 
+import asyncio
 import random
 import re
 import sys
 
-from bluepy.btle import UUID, DefaultDelegate, Peripheral
+from bleak import BleakClient
+from bleak.uuids import normalize_uuid_16
 
 MAC_PATTERN = r"^[0-9A-F]{2}:[0-9A-F]{2}:[0-9A-F]{2}:[0-9A-F]{2}:[0-9A-F]{2}:[0-9A-F]{2}$"
 
 UUID_SERVICE = "fe95"
 
-HANDLE_AUTH = 3
-HANDLE_FIRMWARE_VERSION = 10
-HANDLE_AUTH_INIT = 19
-HANDLE_BEACON_KEY = 25
+# The characteristics of the 'fe95' service have unique uuid values and thus can be addressed via their uuid
+# this can be checked by using the service explorer from https://github.com/hbldh/bleak/blob/master/examples/service_explorer.py
+HANDLE_AUTH = normalize_uuid_16(0x0001)
+HANDLE_FIRMWARE_VERSION = normalize_uuid_16(0x0004)
+HANDLE_AUTH_INIT = normalize_uuid_16(0x0010)
+HANDLE_BEACON_KEY = normalize_uuid_16(0x0014)
 
 MI_KEY1 = bytes([0x90, 0xCA, 0x85, 0xDE])
 MI_KEY2 = bytes([0x92, 0xAB, 0x54, 0xFA])
@@ -94,7 +98,7 @@ def generateRandomToken() -> bytes:
     return token
 
 
-def get_beacon_key(mac, product_id):
+async def get_beacon_key(mac, product_id):
     reversed_mac = reverseMac(mac)
     token = generateRandomToken()
 
@@ -103,29 +107,57 @@ def get_beacon_key(mac, product_id):
 
     # Connect
     print("Connection in progress...")
-    peripheral = Peripheral(deviceAddr=mac)
-    print("Successful connection!")
-
-    # Auth (More information: https://github.com/archaron/docs/blob/master/BLE/ylkg08y.md)
-    print("Authentication in progress...")
-    auth_service = peripheral.getServiceByUUID(UUID_SERVICE)
-    auth_descriptors = auth_service.getDescriptors()
-    peripheral.writeCharacteristic(HANDLE_AUTH_INIT, MI_KEY1, "true")
-    auth_descriptors[1].write(SUBSCRIBE_TRUE, "true")
-    peripheral.writeCharacteristic(HANDLE_AUTH, cipher(mixA(reversed_mac, product_id), token), "true")
-    peripheral.waitForNotifications(10.0)
-    peripheral.writeCharacteristic(3, cipher(token, MI_KEY2), "true")
-    print("Successful authentication!")
-
-    # Read
-    beacon_key = cipher(token, peripheral.readCharacteristic(HANDLE_BEACON_KEY)).hex()
-    firmware_version = cipher(token, peripheral.readCharacteristic(HANDLE_FIRMWARE_VERSION)).decode()
-
-    print(f"beaconKey: '{beacon_key}'")
-    print(f"firmware_version: '{firmware_version}'")
-
-
-def main(argv):
+    client = BleakClient(mac)
+    try:
+        await client.connect()
+        print("Successful connection!")
+
+        # An asyncio future object is needed for callback handling
+        future = asyncio.get_event_loop().create_future()
+
+        # Auth (More information: https://github.com/archaron/docs/blob/master/BLE/ylkg08y.md)
+        print("Authentication in progress...")
+
+        # Send 0x90, 0xCA, 0x85, 0xDE bytes to authInitCharacteristic.
+        await client.write_gatt_char(HANDLE_AUTH_INIT, MI_KEY1, True)
+        # Subscribe authCharacteristic.
+        # (a lambda callback is used to set the futures result on the notification event)
+        await client.start_notify(HANDLE_AUTH, lambda _, data: future.set_result(data))
+        # Send cipher(mixA(reversedMac, productID), token) to authCharacteristic.
+        await client.write_gatt_char(HANDLE_AUTH, cipher(mixA(reversed_mac, product_id), token), True)
+        # Now you'll get a notification on authCharacteristic. You must wait for this notification before proceeding to next step
+        await asyncio.wait_for(future, 10.0)
+
+        # The notification data can be ignored or used to check an integrity, this is optional
+        print(f"notifyData:  '{future.result().hex()}'")
+        # If you want to perform a check, compare cipher(mixB(reversedMac, productID), cipher(mixA(reversedMac, productID), res))
+        # where res is received payload ...
+        print(f"cipheredRes: '{cipher(mixB(reversed_mac, product_id), cipher(mixA(reversed_mac, product_id), future.result())).hex()}'")
+        # ... with your token, they must equal.
+        print(f"randomToken: '{token.hex()}'")
+
+        # Send 0x92, 0xAB, 0x54, 0xFA to authCharacteristic.
+        await client.write_gatt_char(HANDLE_AUTH, cipher(token, MI_KEY2), True)
+        print("Successful authentication!")
+
+        # Read
+        beacon_key = cipher(token, await client.read_gatt_char(HANDLE_BEACON_KEY)).hex()
+        # Read from verCharacteristics. You can ignore the response data, you just have to perform a read to complete authentication process.
+        # If the data is used, it will show the firmware version
+        firmware_version = cipher(token, await client.read_gatt_char(HANDLE_FIRMWARE_VERSION)).decode()
+
+        print(f"beaconKey: '{beacon_key}'")
+        print(f"firmware_version: '{firmware_version}'")
+
+        print("Disconnection in progress...")
+    except Exception as e:
+        print(e)
+    finally:
+        await client.disconnect()
+        print("Disconnected!")
+
+
+async def main(argv):
     # ARGS
     if len(argv) <= 2:
         print("usage: get_beacon_key.py <MAC> <PRODUCT_ID>\n")
@@ -152,8 +184,8 @@ def main(argv):
         return
 
     # BEACON_KEY
-    get_beacon_key(mac, product_id)
+    await get_beacon_key(mac, product_id)
 
 
 if __name__ == '__main__':
-    main(sys.argv)
+    asyncio.run(main(sys.argv))
diff --git a/docs/faq.md b/docs/faq.md
@@ -336,7 +336,7 @@ We have created a python script that will get the beaconkey by connecting to the
 ```
 wget https://raw.githubusercontent.com/custom-components/ble_monitor/master/custom_components/ble_monitor/ble_parser/get_beacon_key.py
 apt-get install python3-pip libglib2.0-dev
-pip3 install bluepy
+pip3 install bleak asyncio
 python3 get_beacon_key.py <MAC> <PRODUCT_ID>
 ```
 Replace `<MAC>` with your MAC address of the remote/dimmer and replace `<PRODUCT_ID>` with one of the following numbers, corresponding to your remote/dimmer.