feat(tools): Tool to upload a file via BLE

tools/card10-ble-transfer.py

+#!/usr/bin/env python3
+
+import bluepy
+import time
+import binascii
+import struct
+import sys
+import argparse
+import tqdm
+import os
+
+
+rx_done = False
+rx_data = None
+
+
+def main() -> None:
+    global rx_done, rx_data
+    parser = argparse.ArgumentParser(
+        description="""\
+Transfer a file to a card10 using Bluetooth Low Energy.
+"""
+    )
+
+    parser.add_argument(
+        "-d",
+        "--directory",
+        help="Target directory on the card10. Root directory by default",
+    )
+    parser.add_argument("-s", "--silent", help="Don't print status information")
+    parser.add_argument(
+        "-m", "--mtu", type=int, default=96, help="MTU to use. Default 96 bytes"
+    )
+    parser.add_argument(
+        "mac", help="BT MAC address of the card10. Format: CA:4D:10:XX:XX:XX"
+    )
+    parser.add_argument("file", help="file to transfer")
+
+    args = parser.parse_args()
+
+    t0 = time.time()
+    p = bluepy.btle.Peripheral(args.mac)
+
+    mtu = args.mtu
+    p.setMTU(mtu)
+    chunk_size = mtu - 8
+
+    s = p.getServiceByUUID("42230100-2342-2342-2342-234223422342")
+    tx = s.getCharacteristics("42230101-2342-2342-2342-234223422342")[0]
+    rx = s.getCharacteristics("42230102-2342-2342-2342-234223422342")[0]
+
+    print("Connection setup time:", int(time.time() - t0), "seconds")
+
+    class MyDelegate(bluepy.btle.DefaultDelegate):
+        def __init__(self):
+            bluepy.btle.DefaultDelegate.__init__(self)
+
+        def handleNotification(self, cHandle, data):
+            global rx_done, rx_data
+            rx_data = data
+            rx_done = True
+
+    def check_crc(data, crc):
+        our_crc = binascii.crc32(data)
+        return crc == struct.pack(">I", our_crc)
+
+    p.setDelegate(MyDelegate())
+
+    if args.directory is not None:
+        filename = args.directory + "/" + args.file
+    else:
+        filename = args.file
+    print("Target filename:", filename)
+
+    file_size = os.path.getsize(args.file)
+    t = tqdm.tqdm(total=file_size, unit="bytes")
+
+    start = b"s" + bytes(filename, "ASCII")
+    tx.write(start)
+    p.waitForNotifications(10.0)
+    if not rx_done:
+        print("No reply")
+        sys.exit(1)
+
+    if not (rx_data[0:1] == b"S" and check_crc(start, rx_data[1:])):
+        print("Filename not acknowledged")
+        return
+
+    with open(args.file) as f:
+        offset = 0
+        # t0 = time.time()
+        while True:
+            payload = bytes(f.read(chunk_size), "UTF-8")
+            if len(payload) == 0:
+                break
+
+            chunk = b"c" + struct.pack(">I", offset) + payload
+            # print((int)((time.time() - t0) * 1000))
+            # t0 = time.time()
+            tx.write(chunk)
+            p.waitForNotifications(10.0)
+            if not rx_done:
+                print("No reply")
+                # TODO: Handle retries
+                sys.exit(1)
+
+            if not (rx_data[0:1] == b"C" and check_crc(chunk, rx_data[1:])):
+                print("Chunk not acknowledged")
+                break
+
+            t.update(len(payload))
+            offset += len(payload)
+
+            if len(payload) < chunk_size:
+                break
+
+    finish = b"f"
+    tx.write(finish)
+    p.waitForNotifications(10.0)
+    if not rx_done:
+        print("No reply")
+        sys.exit(1)
+
+    if not rx_data[0:1] == b"F":
+        print("Finish not acknowledged")
+
+    t.update(0)
+    t.close()
+
+
+if __name__ == "__main__":
+    main()