add simplepyble implementation to master branch

test_bleak/.gitignore

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

test_simplepyble/.gitignore

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

test_simplepyble/flow.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python
+
+import sys
+import time
+import os
+
+from poll import (
+    ble_conn,
+    get_current_temp,
+    get_target_temp, set_target_temp,
+    get_unit_is_fahrenheit,
+    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)
+
+def sleep(t):
+    w = terminal_width
+    if t < w:
+        w = int(t)
+    print_bar(0, 0, w, "s")
+    for i in range(0, w):
+        time.sleep(t / w)
+        print_bar(i + 1, 0, w, "s")
+    print()
+
+def wait_for_temp(client, temp):
+    print("Setting temperature {}".format(temp))
+    set_target_temp(client, temp)
+
+    print("Waiting for temperature to rise...")
+    start = get_current_temp(client)
+    curr = start
+    print_bar(curr, start, temp, " degC")
+    while curr < temp:
+        time.sleep(1.0)
+        curr = 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)
+
+    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
+    sleep(t_pump)
+    set_state(client, (True, False)) # turn off pump
+
+def flow(client):
+    print("Turning on heater")
+    set_state(client, (True, False))
+
+    flow_step(client, 185.0, 10.0, 5.0)
+    flow_step(client, 195.0, 5.0, 20.0)
+    flow_step(client, 205.0, 5.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
+        time.sleep(1.0)
+        set_state(client, (True, False)) # turn off pump
+
+    print("Turning heater off")
+    set_state(client, (False, False)) # turn off heater and pump
+
+    print("Resetting temperature")
+    set_target_temp(client, 190.0)
+
+if __name__ == "__main__":
+    def main(address, adapter):
+        client = ble_conn(address, adapter)
+
+        try:
+            if get_unit_is_fahrenheit(client):
+                raise RuntimeError("Imperial American scum is currently not supported :P")
+
+            print("Starting Workflow")
+            flow(client)
+        except:
+            print("\nTurning heater and pump off")
+            set_state(client, (False, False)) # turn off heater and pump
+
+            print("Disconnecting")
+            client.disconnect()
+
+            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)

test_simplepyble/poll.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python
+
+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, adapter):
+    dev = ble_scan(address, adapter)
+
+    if dev != None:
+        address = dev.address()
+        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
+
+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
+
+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")
+    return (num & 0x200) != 0
+
+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)
+
+def set_state(device, state):
+    heater, pump = state
+    if heater:
+        device.write_request(serviceUuidVolcano4, "1011000f-5354-4f52-5a26-4249434b454c", 0)
+    else:
+        device.write_request(serviceUuidVolcano4, "10110010-5354-4f52-5a26-4249434b454c", 0)
+    if pump:
+        device.write_request(serviceUuidVolcano4, "10110013-5354-4f52-5a26-4249434b454c", 0)
+    else:
+        device.write_request(serviceUuidVolcano4, "10110014-5354-4f52-5a26-4249434b454c", 0)
+
+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, adapter):
+        device = ble_conn(address, adapter)
+        if device == None:
+            return
+
+        try:
+            print("Writing...")
+            set_target_temp(device, 190.0)
+
+            print("Reading...")
+            for i in range(0, 5):
+                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)

test_simplepyble/scan.py

@@ -0,0 +1,41 @@
+import simplepyble
+
+def ble_scan(addr = None, adapterIndex = 0, timeout = 1):
+    adapters = simplepyble.Adapter.get_adapters()
+
+    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?
+    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
+
+    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)
+    if dev != None:
+        print("{} {}".format(dev.identifier(), dev.address()))