fiatlux/firmware/otaflash.py
jedi 24f5e4398d
All checks were successful
continuous-integration/drone/push Build is passing
Add error handling to OTA process
2021-09-12 22:00:59 +02:00

72 lines
2 KiB
Python
Executable file

#!/usr/bin/env python3
import time
import websocket
import argparse
import zlib
from websocket import WebSocketTimeoutException
parser = argparse.ArgumentParser(description='Update fiatlux firmware via websocket.')
parser.add_argument("binfile")
args = parser.parse_args()
bs = 1024
def parse_reply(bytes):
cmd = bytes[0:1].decode("utf-8")
ret = int.from_bytes(bytes[1:2], "big")
val = int.from_bytes(bytes[2:4], "big")
return {'cmd': cmd, 'ret': ret, 'val': val}
with open(args.binfile, "rb") as f:
try:
ws = websocket.WebSocket()
print("send {}".format(args.binfile))
ws.connect("ws://172.16.0.1")
i = 0
bytes = f.read()
rolling = zlib.crc32(bytes)
total = len(bytes)
while True:
chunk = bytes[bs * i:bs * i + bs]
msg = b'F\x00\x00\x00'
msg += i.to_bytes(2, 'big')
msg += len(chunk).to_bytes(2, 'big')
msg += (zlib.crc32(chunk) & 0xffffffff).to_bytes(4, 'big')
msg += chunk
ws.send(msg)
print("\r{:6.2f}%".format(100 * i * bs / total), end='')
reply = parse_reply(ws.recv())
if reply['cmd'] == 'F' and reply['ret'] == 0:
i += 1
elif reply['cmd'] == 'F' and reply['ret'] == 1:
print("Error: SEQUENCE_OUT_OF_ORDER")
i = reply['val']
elif reply['cmd'] == 'F' and reply['ret'] == 2:
print("Error: CHECKSUM_MISMATCH")
i = reply['val']
else:
print(reply)
time.sleep(0.05)
if len(chunk) != bs:
break
print("\rdone ")
msg = b'C\x00\x00\x00'
msg += total.to_bytes(4, 'big')
msg += (rolling).to_bytes(4, 'big')
ws.settimeout(5)
ws.send(msg)
reply = parse_reply(ws.recv())
print(reply)
ws.close()
except WebSocketTimeoutException:
pass
except ConnectionResetError:
pass
except KeyboardInterrupt:
pass