simplepyble, untested

python-test/.gitignore

@@ -0,0 +1 @@
+venv

python-test/flow.py

@@ -1,7 +1,9 @@
 #!/usr/bin/env python
 
 import sys
-import asyncio
+import time
+import os
+
 from poll import (
     ble_conn,
     get_current_temp,
@@ -10,7 +12,7 @@ from poll import (
     get_state, set_state
 )
 
-terminal_width = 80 + 20 # TODO detect?
+terminal_width = os.get_terminal_size().columns - 15
 
 def print_bar(value, start, end, unit):
     width = terminal_width
@@ -20,83 +22,79 @@ def print_bar(value, start, end, unit):
     s += " {}{}".format(value, unit)
     print(s, end="", flush=True)
 
-async def sleep(t):
+def sleep(t):
     w = terminal_width
     if t < w:
         w = int(t)
     print_bar(0, 0, w, "s")
     for i in range(0, w):
-        await asyncio.sleep(t / w)
+        time.sleep(t / w)
         print_bar(i + 1, 0, w, "s")
     print()
 
-async def wait_for_temp(client, temp):
+def wait_for_temp(client, temp):
     print("Setting temperature {}".format(temp))
-    await set_target_temp(client, temp)
+    set_target_temp(client, temp)
 
     print("Waiting for temperature to rise...")
-    start = await get_current_temp(client)
+    start = get_current_temp(client)
     curr = start
     print_bar(curr, start, temp, " degC")
     while curr < temp:
-        await asyncio.sleep(1.0)
-        curr = await get_current_temp(client)
+        time.sleep(1.0)
+        curr = get_current_temp(client)
         print_bar(curr, start, temp, " degC")
     print()
 
     print("Reached temperature {}".format(temp))
 
-async def flow_step(client, temp, t_wait, t_pump):
-    await wait_for_temp(client, temp)
+def flow_step(client, temp, t_wait, t_pump):
+    wait_for_temp(client, temp)
 
     print("Waiting {}s for heat to settle...".format(t_wait))
-    await sleep(t_wait)
+    sleep(t_wait)
 
     print("Pumping for {}s".format(t_pump))
-    await set_state(client, (True, True)) # turn on pump
-    await sleep(t_pump)
-    await set_state(client, (True, False)) # turn off pump
+    set_state(client, (True, True)) # turn on pump
+    sleep(t_pump)
+    set_state(client, (True, False)) # turn off pump
 
-async def flow(client):
+def flow(client):
     print("Turning on heater")
-    await set_state(client, (True, False))
+    set_state(client, (True, False))
 
-    await flow_step(client, 190.0, 20.0, 5.0)
-    await flow_step(client, 205.0, 10.0, 20.0)
-    await flow_step(client, 220.0, 10.0, 20.0)
+    flow_step(client, 190.0, 20.0, 5.0)
+    flow_step(client, 205.0, 10.0, 20.0)
+    flow_step(client, 220.0, 10.0, 20.0)
 
     print("Notification by pumping three times...")
     for i in range(0, 3):
-        await asyncio.sleep(1.0)
-        await set_state(client, (True, True)) # turn on pump
-        await asyncio.sleep(1.0)
-        await set_state(client, (True, False)) # turn off pump
+        time.sleep(1.0)
+        set_state(client, (True, True)) # turn on pump
+        time.sleep(1.0)
+        set_state(client, (True, False)) # turn off pump
 
     print("Turning heater off")
-    await set_state(client, (False, False)) # turn off heater and pump
+    set_state(client, (False, False)) # turn off heater and pump
 
-async def main(address):
-    device = await ble_conn(address)
+if __name__ == "__main__":
+    def main(address):
+        device = ble_conn(address)
 
-    print("Connecting...")
-    async with device as client:
         try:
-            if await get_unit_is_fahrenheit(client):
-                print("Imperial American scum is currently not supported :P")
-                sys.exit(42)
+            if get_unit_is_fahrenheit(client):
+                raise RuntimeError("Imperial American scum is currently not supported :P")
 
             print("Starting Workflow")
-            await flow(client)
+            flow(client)
         except:
             print("\nTurning heater off")
-            await set_state(client, (False, False)) # turn off heater and pump
+            set_state(client, (False, False)) # turn off heater and pump
             raise
 
         print("Disconnecting...")
 
-if __name__ == "__main__":
-    if len(sys.argv) <= 1:
-        print("Please pass MAC address of device")
-        sys.exit(1)
-
-    asyncio.run(main(sys.argv[1]))
+    arg = None
+    if len(sys.argv) > 1:
+        arg = sys.argv[1]
+    main(arg)

python-test/poll.py

@@ -1,111 +1,101 @@
 #!/usr/bin/env python
 
-import sys
-import asyncio
-from bleak import BleakClient, BleakScanner
-from bleak.exc import BleakDBusError
-
-async def ble_conn(address):
-    attempts = 10
-    print("Opening device..", end = "", flush=True)
-    while attempts > 0:
-        try:
-            if (attempts % 10) == 0:
-                print(".", end = "", flush=True)
-
-            device = await BleakScanner.find_device_by_address(address)
-            client = BleakClient(device)
-            print("", flush=True)
-            return client
-        except BleakDBusError as e:
-            attempts -= 1
-            if attempts == 0:
-                print("", flush=True)
-                print(e)
-            else:
-                await asyncio.sleep(0.1)
-
-    print("Could not connect to device")
-    return None
-
-async def get_current_temp(client):
-    val = await client.read_gatt_char("10110001-5354-4f52-5a26-4249434b454c")
+import simplepyble
+from scan import ble_scan
+import time
+
+serviceUuidVolcano3 = "10100000-5354-4f52-5a26-4249434b454c"
+serviceUuidVolcano4 = "10110000-5354-4f52-5a26-4249434b454c"
+
+def ble_conn(address):
+    dev = ble_scan(address)
+
+    if dev != None:
+        print("Connecting to {}...".format(address))
+        dev.connect()
+
+    return dev
+
+def get_current_temp(device):
+    val = device.read(serviceUuidVolcano4, "10110001-5354-4f52-5a26-4249434b454c")
     num = int.from_bytes(val, byteorder="little")
     return num / 10.0
 
-async def get_target_temp(client):
-    val = await client.read_gatt_char("10110003-5354-4f52-5a26-4249434b454c")
+def get_target_temp(device):
+    val = device.read(serviceUuidVolcano4, "10110003-5354-4f52-5a26-4249434b454c")
     num = int.from_bytes(val, byteorder="little")
     return num / 10.0
 
-async def set_target_temp(client, temp):
+def set_target_temp(device, temp):
     val = int(temp * 10.0)
     d = val.to_bytes(4, byteorder="little")
-    await client.write_gatt_char("10110003-5354-4f52-5a26-4249434b454c", d)
+    device.write_request(serviceUuidVolcano4, "10110003-5354-4f52-5a26-4249434b454c", d)
 
-async def get_unit_is_fahrenheit(client):
-    val = await client.read_gatt_char("1010000d-5354-4f52-5a26-4249434b454c")
+def get_unit_is_fahrenheit(device):
+    val = device.read(serviceUuidVolcano3, "1010000d-5354-4f52-5a26-4249434b454c")
     num = int.from_bytes(val, byteorder="little")
     return (num & 0x200) != 0
 
-async def get_state(client):
-    val = await client.read_gatt_char("1010000c-5354-4f52-5a26-4249434b454c")
+def get_state(device):
+    val = device.read(serviceUuidVolcano3, "1010000c-5354-4f52-5a26-4249434b454c")
     num = int.from_bytes(val, byteorder="little")
     heater = (num & 0x0020) != 0
     pump = (num & 0x2000) != 0
     return (heater, pump)
 
-async def set_state(client, state):
+def set_state(device, state):
     heater, pump = state
     if heater:
-        await client.write_gatt_char("1011000f-5354-4f52-5a26-4249434b454c", 0)
+        device.write_request(serviceUuidVolcano4, "1011000f-5354-4f52-5a26-4249434b454c", 0)
     else:
-        await client.write_gatt_char("10110010-5354-4f52-5a26-4249434b454c", 0)
+        device.write_request(serviceUuidVolcano4, "10110010-5354-4f52-5a26-4249434b454c", 0)
     if pump:
-        await client.write_gatt_char("10110013-5354-4f52-5a26-4249434b454c", 0)
-    else:
-        await client.write_gatt_char("10110014-5354-4f52-5a26-4249434b454c", 0)
-
-async def test_poll(client):
-    temp = await get_current_temp(client)
-    print("Current Temperature: {}".format(temp))
-
-    target = await get_target_temp(client)
-    print("Target Temperature: {}".format(target))
-
-    fahrenheit = await get_unit_is_fahrenheit(client)
-    if fahrenheit:
-        print("Unit is Fahrenheit")
+        device.write_request(serviceUuidVolcano4, "10110013-5354-4f52-5a26-4249434b454c", 0)
     else:
-        print("Unit is Celsius")
+        device.write_request(serviceUuidVolcano4, "10110014-5354-4f52-5a26-4249434b454c", 0)
 
-    heater, pump = await get_state(client)
-    if heater:
-        print("Heater is On")
-    else:
-        print("Heater is Off")
-    if pump:
-        print("Pump is On")
-    else:
-        print("Pump is Off")
-
-async def test(address):
-    device = await ble_conn(address)
+if __name__ == "__main__":
+    def test_poll(device):
+        temp = get_current_temp(device)
+        print("Current Temperature: {}".format(temp))
+
+        target = get_target_temp(device)
+        print("Target Temperature: {}".format(target))
+
+        fahrenheit = get_unit_is_fahrenheit(device)
+        if fahrenheit:
+            print("Unit is Fahrenheit")
+        else:
+            print("Unit is Celsius")
+
+        heater, pump = get_state(device)
+        if heater:
+            print("Heater is On")
+        else:
+            print("Heater is Off")
+        if pump:
+            print("Pump is On")
+        else:
+            print("Pump is Off")
+
+    def test(address):
+        device = ble_conn(address)
+        if device == None:
+            return
 
-    print("Connecting...")
-    async with device as client:
         print("Writing...")
-        await set_target_temp(client, 190.0)
+        set_target_temp(device, 190.0)
 
         print("Reading...")
         for i in range(0, 5):
-            await test_poll(client)
+            test_poll(device)
             print()
-            await asyncio.sleep(2.0)
+            time.sleep(2.0)
 
-if __name__ == "__main__":
-    if len(sys.argv) <= 1:
-        print("Please pass MAC address of device")
-        sys.exit(1)
+    import sys
+
+    arg = None
+    if len(sys.argv) > 1:
+        arg = sys.argv[1]
 
-    asyncio.run(test(sys.argv[1]))
+    test(arg)

python-test/scan.py

@@ -1,16 +1,39 @@
-#!/usr/bin/env python
+import simplepyble
 
-import asyncio
-from bleak import BleakScanner
+def ble_scan(addr):
+    adapters = simplepyble.Adapter.get_adapters()
 
-async def scan():
-    print("scanning for 5 seconds, please wait...")
-    devices = await BleakScanner.discover(return_adv=True)
-    for d, a in devices.values():
-        print()
-        print(d)
-        print("-" * len(str(d)))
-        print(a)
+    if len(adapters) == 0:
+        print("No adapters found")
+        return None
+
+    # TODO allow selection of bluetooth adapter
+    adapter = adapters[0]
+    print("Selected adapter: {} [{}]".format(adapter.identifier(), adapter.address()))
+
+    # TODO support longer scans?
+    print("Scanning for '{}' for 1s...".format(addr))
+    adapter.scan_for(1000)
+
+    peripherals = adapter.scan_get_results()
+    for peripheral in peripherals:
+        if addr != None:
+            if addr == peripheral.address():
+                return peripheral
+        else:
+            if peripheral.identifier() == "S&B VOLCANO H":
+                return peripheral
+
+    print("No device found")
+    return None
 
 if __name__ == "__main__":
-    asyncio.run(scan())
+    import sys
+
+    arg = None
+    if len(sys.argv) > 1:
+        arg = sys.argv[1]
+
+    dev = ble_scan(arg)
+    if dev != None:
+        print("{} {}".format(dev.identifier(), dev.address()))