now made for micropython on rp2040 / pi pico

python-test/copy.sh

+#!/bin/bash
+
+if [ $# -ne 0 ] ; then
+cat << EOF | rshell
+cp flow.py /pyboard
+cp poll.py /pyboard
+cp scan.py /pyboard
+cp $1 /pyboard/main.py
+EOF
+else
+cat << EOF | rshell
+cp flow.py /pyboard
+cp poll.py /pyboard
+cp scan.py /pyboard
+EOF
+fi

python-test/flow.py

 #!/usr/bin/env python
 
+import uasyncio as asyncio
 import sys
 import time
-import os
 
 from poll import (
     ble_conn,
     get_state, set_state
 )
 
-terminal_width = os.get_terminal_size().columns - 15
-
 def print_bar(value, start, end, unit):
-    width = terminal_width
-    s = "\r"
-    s += "#" * int((value - start) / (end - start) * width)
-    s += "-" * (width - int((value - start) / (end - start) * width))
-    s += " {}{}".format(value, unit)
-    print(s, end="", flush=True)
+    print("{}{} -> {}{} -> {}{}".format(start, unit, value, unit, end, unit))
 
 def sleep(t):
     w = terminal_width
         print_bar(i + 1, 0, w, "s")
     print()
 
-def wait_for_temp(client, temp):
+async def wait_for_temp(client, temp):
     print("Setting temperature {}".format(temp))
-    set_target_temp(client, temp)
+    await set_target_temp(client, temp)
 
     print("Waiting for temperature to rise...")
-    start = get_current_temp(client)
+    start = await get_current_temp(client)
     curr = start
     print_bar(curr, start, temp, " degC")
     while curr < temp:
         time.sleep(1.0)
-        curr = get_current_temp(client)
+        curr = await get_current_temp(client)
         print_bar(curr, start, temp, " degC")
     print()
 
     print("Reached temperature {}".format(temp))
 
-def flow_step(client, temp, t_wait, t_pump):
-    wait_for_temp(client, temp)
+async def flow_step(client, temp, t_wait, t_pump):
+    await wait_for_temp(client, temp)
 
     print("Waiting {}s for heat to settle...".format(t_wait))
     sleep(t_wait)
 
     print("Pumping for {}s".format(t_pump))
-    set_state(client, (True, True)) # turn on pump
+    await set_state(client, (True, True)) # turn on pump
     sleep(t_pump)
-    set_state(client, (True, False)) # turn off pump
+    await set_state(client, (True, False)) # turn off pump
 
-def flow(client):
+async def flow(client):
     print("Turning on heater")
-    set_state(client, (True, False))
+    await set_state(client, (True, False))
 
-    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)
+    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)
 
     print("Notification by pumping three times...")
     for i in range(0, 3):
         time.sleep(1.0)
-        set_state(client, (True, True)) # turn on pump
+        await set_state(client, (True, True)) # turn on pump
         time.sleep(1.0)
-        set_state(client, (True, False)) # turn off pump
+        await set_state(client, (True, False)) # turn off pump
 
     print("Turning heater off")
-    set_state(client, (False, False)) # turn off heater and pump
+    await set_state(client, (False, False)) # turn off heater and pump
 
 if __name__ == "__main__":
-    def main(address, adapter):
-        client = ble_conn(address, adapter)
+    async def main(address):
+        client = await ble_conn(address)
 
         try:
-            if get_unit_is_fahrenheit(client):
+            if await get_unit_is_fahrenheit(client):
                 raise RuntimeError("Imperial American scum is currently not supported :P")
 
             print("Starting Workflow")
-            flow(client)
+            await flow(client)
         except:
             print("\nTurning heater off")
-            set_state(client, (False, False)) # turn off heater and pump
-
-            print("Disconnecting")
-            client.disconnect()
-
+            await set_state(client, (False, False)) # turn off heater and pump
             raise
 
-        print("Disconnecting")
-        client.disconnect()
-
-    adapter = None
-    mac = None
-    if len(sys.argv) > 1:
-        adapter = int(sys.argv[1])
-    if len(sys.argv) > 2:
-        mac = sys.argv[2]
-
-    main(mac, adapter)
+    asyncio.run(main(None))

python-test/poll.py

 #!/usr/bin/env python
 
-import simplepyble
-from scan import ble_scan
+import uasyncio as asyncio
+import bluetooth
 import time
 
-serviceUuidVolcano3 = "10100000-5354-4f52-5a26-4249434b454c"
-serviceUuidVolcano4 = "10110000-5354-4f52-5a26-4249434b454c"
+from scan import ble_scan
+
+serviceUuidVolcano3 = bluetooth.UUID("10100000-5354-4f52-5a26-4249434b454c")
+serviceUuidVolcano4 = bluetooth.UUID("10110000-5354-4f52-5a26-4249434b454c")
 
-def ble_conn(address, adapter):
-    dev = ble_scan(address, adapter)
+async def ble_conn(address):
+    dev = await ble_scan(address)
 
-    if dev != None:
-        address = dev.address()
+    if dev:
+        address = dev.addr_hex()
         print("Connecting to '{}'...".format(address))
-        dev.connect()
+        connection = await dev.connect()
+        return connection
 
-    return dev
+    return None
 
-def get_current_temp(device):
-    val = device.read(serviceUuidVolcano4, "10110001-5354-4f52-5a26-4249434b454c")
-    num = int.from_bytes(val, byteorder="little")
+async def get_current_temp(device):
+    service = await device.service(serviceUuidVolcano4)
+    uuid = bluetooth.UUID("10110001-5354-4f52-5a26-4249434b454c")
+    characteristic = await service.characteristic(uuid)
+    val = await characteristic.read()
+    num = int.from_bytes(val, "little")
     return num / 10.0
 
-def get_target_temp(device):
-    val = device.read(serviceUuidVolcano4, "10110003-5354-4f52-5a26-4249434b454c")
-    num = int.from_bytes(val, byteorder="little")
+async def get_target_temp(device):
+    service = await device.service(serviceUuidVolcano4)
+    uuid = bluetooth.UUID("10110003-5354-4f52-5a26-4249434b454c")
+    characteristic = await service.characteristic(uuid)
+    val = await characteristic.read()
+    num = int.from_bytes(val, "little")
     return num / 10.0
 
-def set_target_temp(device, temp):
+async def set_target_temp(device, temp):
     val = int(temp * 10.0)
-    d = val.to_bytes(4, byteorder="little")
-    device.write_request(serviceUuidVolcano4, "10110003-5354-4f52-5a26-4249434b454c", d)
-
-def get_unit_is_fahrenheit(device):
-    val = device.read(serviceUuidVolcano3, "1010000d-5354-4f52-5a26-4249434b454c")
-    num = int.from_bytes(val, byteorder="little")
+    d = val.to_bytes(4, "little")
+    service = await device.service(serviceUuidVolcano4)
+    uuid = bluetooth.UUID("10110003-5354-4f52-5a26-4249434b454c")
+    characteristic = await service.characteristic(uuid)
+    await characteristic.write(d)
+
+async def get_unit_is_fahrenheit(device):
+    service = await device.service(serviceUuidVolcano3)
+    uuid = bluetooth.UUID("1010000d-5354-4f52-5a26-4249434b454c")
+    characteristic = await service.characteristic(uuid)
+    val = await characteristic.read()
+    num = int.from_bytes(val, "little")
     return (num & 0x200) != 0
 
-def get_state(device):
-    val = device.read(serviceUuidVolcano3, "1010000c-5354-4f52-5a26-4249434b454c")
-    num = int.from_bytes(val, byteorder="little")
+async def get_state(device):
+    service = await device.service(serviceUuidVolcano3)
+    uuid = bluetooth.UUID("1010000c-5354-4f52-5a26-4249434b454c")
+    characteristic = await service.characteristic(uuid)
+    val = await characteristic.read()
+    num = int.from_bytes(val, "little")
     heater = (num & 0x0020) != 0
     pump = (num & 0x2000) != 0
     return (heater, pump)
 
-def set_state(device, state):
+async def set_state(device, state):
     heater, pump = state
     if heater:
-        device.write_request(serviceUuidVolcano4, "1011000f-5354-4f52-5a26-4249434b454c", 0)
+        service = await device.service(serviceUuidVolcano4)
+        uuid = bluetooth.UUID("1011000f-5354-4f52-5a26-4249434b454c")
+        characteristic = await service.characteristic(uuid)
+        await characteristic.write(int(0).to_bytes(1, "little"))
     else:
-        device.write_request(serviceUuidVolcano4, "10110010-5354-4f52-5a26-4249434b454c", 0)
+        service = await device.service(serviceUuidVolcano4)
+        uuid = bluetooth.UUID("10110010-5354-4f52-5a26-4249434b454c")
+        characteristic = await service.characteristic(uuid)
+        await characteristic.write(int(0).to_bytes(1, "little"))
     if pump:
-        device.write_request(serviceUuidVolcano4, "10110013-5354-4f52-5a26-4249434b454c", 0)
+        service = await device.service(serviceUuidVolcano4)
+        uuid = bluetooth.UUID("10110013-5354-4f52-5a26-4249434b454c")
+        characteristic = await service.characteristic(uuid)
+        await characteristic.write(int(0).to_bytes(1, "little"))
     else:
-        device.write_request(serviceUuidVolcano4, "10110014-5354-4f52-5a26-4249434b454c", 0)
+        service = await device.service(serviceUuidVolcano4)
+        uuid = bluetooth.UUID("10110014-5354-4f52-5a26-4249434b454c")
+        characteristic = await service.characteristic(uuid)
+        await characteristic.write(int(0).to_bytes(1, "little"))
 
 if __name__ == "__main__":
-    def test_poll(device):
-        temp = get_current_temp(device)
+    async def test_poll(device):
+        temp = await get_current_temp(device)
         print("Current Temperature: {}".format(temp))
 
-        target = get_target_temp(device)
+        target = await get_target_temp(device)
         print("Target Temperature: {}".format(target))
 
-        fahrenheit = get_unit_is_fahrenheit(device)
+        fahrenheit = await get_unit_is_fahrenheit(device)
         if fahrenheit:
             print("Unit is Fahrenheit")
         else:
             print("Unit is Celsius")
 
-        heater, pump = get_state(device)
+        heater, pump = await get_state(device)
         if heater:
             print("Heater is On")
         else:
         else:
             print("Pump is Off")
 
-    def test(address, adapter):
-        device = ble_conn(address, adapter)
+    async def test(address):
+        device = await ble_conn(address)
         if device == None:
             return
 
-        try:
+        async with device:
             print("Writing...")
-            set_target_temp(device, 190.0)
+            await set_target_temp(device, 190.0)
 
             print("Reading...")
             for i in range(0, 5):
-                test_poll(device)
+                await test_poll(device)
                 print()
                 time.sleep(2.0)
-        except:
-            print("Disconnecting")
-            client.disconnect()
-
-            raise
-
-        print("Disconnecting")
-        client.disconnect()
-
-    import sys
-
-    adapter = None
-    mac = None
-    if len(sys.argv) > 1:
-        adapter = int(sys.argv[1])
-    if len(sys.argv) > 2:
-        mac = sys.argv[2]
 
-    test(mac, adapter)
+    asyncio.run(test(None))

python-test/scan.py

-import simplepyble
+# https://github.com/micropython/micropython-lib/blob/master/micropython/bluetooth/aioble/examples/temp_client.py
 
-def ble_scan(addr = None, adapterIndex = 0, timeout = 1):
-    adapters = simplepyble.Adapter.get_adapters()
+import uasyncio as asyncio
+import aioble
+import bluetooth
+import sys
 
-    if len(adapters) == 0:
-        print("No adapters found")
-        return None
-
-    adapter = adapters[adapterIndex]
-    print("Selected adapter: {} [{}]".format(adapter.identifier(), adapter.address()))
-
-    # TODO abort scan when found?
+async def ble_scan(addr = None, timeout = 0.1):
     print("Scanning for '{}' for {}s...".format(addr, timeout))
-    adapter.scan_for(timeout * 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
+    scanner = aioble.scan(int(timeout * 1000.0), interval_us=30000, window_us=30000, active=True)
+    async with scanner as s:
+        async for d in s:
+            print("Scan: '{}' [{}]".format(d.name(), d.device.addr_hex()))
+            if addr != None:
+                if addr == d.device.addr_hex():
+                    return d.device
+            else:
+                if d.name() == "S&B VOLCANO H":
+                    return d.device
 
     print("No device found")
     return None
 
 if __name__ == "__main__":
-    import sys
-
-    adapter = None
-    mac = None
-    if len(sys.argv) > 1:
-        adapter = int(sys.argv[1])
-    if len(sys.argv) > 2:
-        mac = sys.argv[2]
-
-    dev = ble_scan(mac, adapter)
+    dev = asyncio.run(ble_scan())
     if dev != None:
-        print("{} {}".format(dev.identifier(), dev.address()))
+        print("{}".format(dev.addr_hex()))