Commit 77ee86e4 authored by Rahix's avatar Rahix

Merge 'Linux script for BLE file-transfer'

See merge request card10/firmware!378
parents 861d604a 12fa0cf0
Pipeline #4553 passed with stages
in 1 minute and 37 seconds
......@@ -98,8 +98,9 @@ static const appUpdateCfg_t bleUpdateCfg =
{
6000, /*! Connection idle period in ms before attempting
connection parameter update; set to zero to disable */
800/1.25, /*! Minimum connection interval in 1.25ms units */
1000/1.25, /*! Maximum connection interval in 1.25ms units */
30/1.25, /*! Minimum connection interval in 1.25ms units.
Values < 8 didn't work with my Tinkpad T470 */
40/1.25, /*! Maximum connection interval in 1.25ms units */
0, /*! Connection latency */
9000/10, /*! Supervision timeout in 10ms units */
5 /*! Number of update attempts before giving up */
......
#!/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()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment