135 lines
4.9 KiB
Python
135 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from bleak import BleakClient, BleakScanner
|
|
|
|
|
|
DEFAULT_DEVICE_NAME = "RLCD-Metrics"
|
|
DEFAULT_STATUS_CHAR_UUID = "9f4af548-1a14-4fc1-8f5a-4b9f00000003"
|
|
DEFAULT_OTA_CONTROL_CHAR_UUID = "9f4af548-1a14-4fc1-8f5a-4b9f00000004"
|
|
DEFAULT_OTA_DATA_CHAR_UUID = "9f4af548-1a14-4fc1-8f5a-4b9f00000005"
|
|
|
|
|
|
async def find_device(name: str, address: str | None, timeout: float):
|
|
devices = await BleakScanner.discover(timeout=timeout)
|
|
for device in devices:
|
|
if address and device.address.lower() == address.lower():
|
|
return device
|
|
if device.name == name:
|
|
return device
|
|
return None
|
|
|
|
|
|
def status_callback(_sender, data: bytearray):
|
|
text = bytes(data).decode("ascii", errors="replace").strip()
|
|
if text:
|
|
print(text, flush=True)
|
|
|
|
|
|
async def send_ota(args) -> int:
|
|
firmware = Path(args.firmware).resolve()
|
|
if not firmware.is_file():
|
|
print(f"Firmware not found: {firmware}", file=sys.stderr)
|
|
return 2
|
|
|
|
size = firmware.stat().st_size
|
|
if size <= 0:
|
|
print(f"Firmware is empty: {firmware}", file=sys.stderr)
|
|
return 2
|
|
|
|
print(f"Scanning for {args.name}...", flush=True)
|
|
device = await find_device(args.name, args.address, args.scan_timeout)
|
|
if device is None:
|
|
print(f"Device not found: {args.name}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Connecting to {device.name} ({device.address})...", flush=True)
|
|
async with BleakClient(device, timeout=args.connect_timeout) as client:
|
|
await client.start_notify(args.status_char_uuid, status_callback)
|
|
print(f"Sending OTA image: {firmware} ({size} bytes)", flush=True)
|
|
|
|
try:
|
|
await asyncio.wait_for(
|
|
client.write_gatt_char(
|
|
args.ota_control_char_uuid,
|
|
f"BEGIN,{size}".encode("ascii"),
|
|
response=True,
|
|
),
|
|
timeout=args.write_timeout,
|
|
)
|
|
await asyncio.sleep(0.5)
|
|
|
|
sent = 0
|
|
last_percent = -1
|
|
with firmware.open("rb") as handle:
|
|
while True:
|
|
chunk = handle.read(args.chunk_size)
|
|
if not chunk:
|
|
break
|
|
await asyncio.wait_for(
|
|
client.write_gatt_char(
|
|
args.ota_data_char_uuid,
|
|
chunk,
|
|
response=args.write_with_response,
|
|
),
|
|
timeout=args.write_timeout,
|
|
)
|
|
sent += len(chunk)
|
|
percent = int(sent * 100 / size)
|
|
if percent != last_percent and (percent % 5 == 0 or percent == 100):
|
|
print(f"OTA progress: {percent}% ({sent}/{size})", flush=True)
|
|
last_percent = percent
|
|
if args.chunk_delay > 0:
|
|
await asyncio.sleep(args.chunk_delay)
|
|
|
|
await asyncio.wait_for(
|
|
client.write_gatt_char(args.ota_control_char_uuid, b"END", response=True),
|
|
timeout=args.write_timeout,
|
|
)
|
|
except Exception:
|
|
try:
|
|
await client.write_gatt_char(args.ota_control_char_uuid, b"ABORT", response=True)
|
|
except Exception:
|
|
pass
|
|
raise
|
|
|
|
print("OTA image sent. Waiting for board reboot notice...", flush=True)
|
|
await asyncio.sleep(args.finish_wait)
|
|
|
|
return 0
|
|
|
|
|
|
def parse_args():
|
|
default_firmware = Path(__file__).resolve().parents[2] / "build" / "rlcd_clock.bin"
|
|
parser = argparse.ArgumentParser(description="Send RLCD firmware over BLE OTA.")
|
|
parser.add_argument("--firmware", default=str(default_firmware))
|
|
parser.add_argument("--name", default=DEFAULT_DEVICE_NAME)
|
|
parser.add_argument("--address", default=None)
|
|
parser.add_argument("--status-char-uuid", default=DEFAULT_STATUS_CHAR_UUID)
|
|
parser.add_argument("--ota-control-char-uuid", default=DEFAULT_OTA_CONTROL_CHAR_UUID)
|
|
parser.add_argument("--ota-data-char-uuid", default=DEFAULT_OTA_DATA_CHAR_UUID)
|
|
parser.add_argument("--chunk-size", type=int, default=180)
|
|
parser.add_argument("--chunk-delay", type=float, default=0.01)
|
|
parser.add_argument("--write-timeout", type=float, default=10.0)
|
|
parser.add_argument("--write-with-response", action="store_true")
|
|
parser.add_argument("--scan-timeout", type=float, default=20.0)
|
|
parser.add_argument("--connect-timeout", type=float, default=20.0)
|
|
parser.add_argument("--finish-wait", type=float, default=8.0)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
if os.name != "nt":
|
|
print("This helper is tuned for the current Windows BLE setup.", file=sys.stderr)
|
|
args = parse_args()
|
|
return asyncio.run(send_ota(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|