diff --git a/meshtastic/__main__.py b/meshtastic/__main__.py index ca1680a58..99d5d3dd7 100644 --- a/meshtastic/__main__.py +++ b/meshtastic/__main__.py @@ -487,36 +487,62 @@ def onConnected(interface): closeNow = True waitForAckNak = True - if not isinstance(interface, meshtastic.tcp_interface.TCPInterface): - meshtastic.util.our_exit( - "Error: OTA update currently requires a TCP connection to the node (use --host)." + if isinstance(interface, meshtastic.tcp_interface.TCPInterface): + ota = meshtastic.ota.ESP32WiFiOTA(args.ota_update, interface.hostname) + + print(f"Triggering OTA update on {interface.hostname}...") + interface.getNode(args.dest, False, **getNode_kwargs).startOTA( + ota_mode=admin_pb2.OTAMode.OTA_WIFI, + ota_file_hash=ota.hash_bytes() + ) + + print("Waiting for device to reboot into OTA mode...") + time.sleep(5) + + retries = 5 + while retries > 0: + try: + ota.update() + break + + except Exception as e: + retries -= 1 + if retries == 0: + meshtastic.util.our_exit(f"\nOTA update failed: {e}") + + time.sleep(2) + + print("\nOTA update completed successfully!") + elif isinstance(interface, meshtastic.ble_interface.BLEInterface): + ota = meshtastic.ota.ESP32BLEOTA(args.ota_update) + + print(f"Triggering OTA update on {interface.client.bleak_client.address}...") + interface.getNode(args.dest, False, **getNode_kwargs).startOTA( + ota_mode=admin_pb2.OTAMode.OTA_BLE, + ota_file_hash=ota.hash_bytes() ) - ota = meshtastic.ota.ESP32WiFiOTA(args.ota_update, interface.hostname) - - print(f"Triggering OTA update on {interface.hostname}...") - interface.getNode(args.dest, False, **getNode_kwargs).startOTA( - ota_mode=admin_pb2.OTAMode.OTA_WIFI, - ota_file_hash=ota.hash_bytes() - ) - - print("Waiting for device to reboot into OTA mode...") - time.sleep(5) - - retries = 5 - while retries > 0: - try: - ota.update() - break - - except Exception as e: - retries -= 1 - if retries == 0: - meshtastic.util.our_exit(f"\nOTA update failed: {e}") + interface.close() - time.sleep(2) + print("Waiting for device to reboot into OTA mode...") + time.sleep(5) - print("\nOTA update completed successfully!") + retries = 5 + while retries > 0: + try: + ota.update() + break + + except Exception as e: + retries -= 1 + if retries == 0: + meshtastic.util.our_exit(f"\nOTA update failed: {e}") + + time.sleep(2) + else: + meshtastic.util.our_exit( + "Error: OTA update currently requires a BLE or TCP connection to the node (use --host or --ble)." + ) if args.enter_dfu: closeNow = True @@ -2306,7 +2332,7 @@ def addRemoteAdminArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPars group.add_argument( "--ota-update", - help="Perform an OTA update on the local node (ESP32, firmware version >=2.7.18, WiFi/TCP only for now). " + help="Perform an OTA update on the local node (ESP32, firmware version >=2.7.18, WiFi/TCP or BLE only). " "Specify the path to the firmware file.", metavar="FIRMWARE_FILE", action="store", diff --git a/meshtastic/ota.py b/meshtastic/ota.py index af9feae42..0af6eb58d 100644 --- a/meshtastic/ota.py +++ b/meshtastic/ota.py @@ -5,7 +5,8 @@ import socket import logging from typing import Optional, Callable - +import meshtastic +import queue logger = logging.getLogger(__name__) @@ -126,3 +127,179 @@ def update(self, progress_callback: Optional[Callable[[int, int], None]] = None) if self._socket: self._socket.close() self._socket = None + +class ESP32BLEOTA: + """ESP32 BLE Unified OTA updates.""" + + SERVICE_UUID = "4fafc201-1fb5-459e-8fcc-c5c9c331914b" + DEVICE_NAME_UUID = "00002a00-0000-1000-8000-00805f9b34fb" + WRITE_UUID = "62ec0272-3ec5-11eb-b378-0242ac130005" + NOTIFY_UUID = "62ec0272-3ec5-11eb-b378-0242ac130003" + + class NotificationReceiver: + """Thread-safe queue receiver for BLE notifications.""" + def __init__(self) -> None: + self._queue: queue.Queue[str] = queue.Queue() + + def on_notify(self, _, data: bytearray) -> None: + """Callback passed to Bleak client.start_notify.""" + text = bytes(data).decode("utf-8", errors="replace").strip("\r\n") + self._queue.put(text) + + def wait_for_next(self, timeout: Optional[float] = 10.0) -> Optional[str]: + """Block until a notification arrives, or return None if timeout expires.""" + try: + return self._queue.get(timeout=timeout) + except queue.Empty: + return None + + def __init__(self, filename: str): + self._filename = filename + + if not os.path.exists(self._filename): + raise FileNotFoundError(f"File {self._filename} does not exist") + + self._file_hash = _file_sha256(self._filename) + + def hash_bytes(self) -> bytes: + """Return the hash as bytes.""" + return self._file_hash.digest() + + def hash_hex(self) -> str: + """Return the hash as a hex string.""" + return self._file_hash.hexdigest() + + def _read_device_name(self, address: str) -> Optional[str]: + """Read the device name""" + device_name = None + client = meshtastic.ble_interface.BLEClient(address) + try: + client.connect() + if client.has_characteristic(self.DEVICE_NAME_UUID): + raw_bytes = client.read_gatt_char(self.DEVICE_NAME_UUID) + device_name = raw_bytes.decode("utf-8").strip("\x00\r\n ") + except Exception: + pass + finally: + try: + client.disconnect() + except Exception: + pass + client.close() + return device_name + + def update(self, progress_callback: Optional[Callable[[int, int], None]] = None): + """Perform the OTA update.""" + with open(self._filename, "rb") as f: + data = f.read() + size = len(data) + + logger.info(f"Starting OTA update with {self._filename} ({size} bytes, hash {self.hash_hex()})") + + logger.info("Scanning for BLE devices in OTA mode (takes 10 seconds)...") + with meshtastic.ble_interface.BLEClient() as client: + response = client.discover( + timeout=10, return_adv=True, service_uuids=[self.SERVICE_UUID.lower()], scanning_mode="active" + ) + + devices = [d[0] for d in response.values() if self.SERVICE_UUID in (u.lower() for u in d[1].service_uuids)] + if not devices: + raise OTAError("Could not find any device in OTA mode") + print("Found devices in OTA mode:") + for idx, device in enumerate(devices): + device.name = self._read_device_name(device.address) + print(f"\t[{idx}] {device.name} ({device.address})") + + # We have to let the user choose because the Bluetooth address changes + # in OTA bootloader compared to meshtastic firmware + selected_device = None + while not selected_device: + user_selection = input("Please select number of device you want to update (q to quit): ") + if user_selection.isdigit(): + idx = int(user_selection) + if 0 <= idx < len(devices): + selected_device = devices[idx] + if user_selection == "q": + return + + # Establishing Bluetooth connection + client = meshtastic.ble_interface.BLEClient(selected_device.address) + try: + client.connect() + + logger.info(f"Connected to {selected_device.name or selected_device.address}") + + # Register callback for BLE notify (Messages from the device) + notification_receiver = self.NotificationReceiver() + client.start_notify(self.NOTIFY_UUID, callback=notification_receiver.on_notify) + + # Send start command + client.write_gatt_char(self.WRITE_UUID, f"OTA {size} {self.hash_hex()}\n".encode("utf-8"), response=True) + + # Wait for OK from the device + while True: + response = notification_receiver.wait_for_next(timeout=10.0) + if response: + if response == "OK": + break + elif response == "ERASING": + logger.info("Device is erasing flash...") + elif response.startswith("ERR "): + raise OTAError(f"Device reported error: {response}") + else: + logger.warning(f"Unexpected response: {response}") + else: + raise OTAError("Timeout waiting for response") + + # Streaming firmware data in chunks of 512 bytes (or MTU size) + sent_bytes = 0 + chunk_size = min(512, client.bleak_client.mtu_size - 3) + while sent_bytes < size: + chunk = data[sent_bytes : sent_bytes + chunk_size] + client.write_gatt_char(self.WRITE_UUID, chunk, response=True) + + sent_bytes += len(chunk) + + if sent_bytes < size: + while True: + response = notification_receiver.wait_for_next(timeout=10.0) + if response: + if response == "ACK": + break + elif response.startswith("ERR "): + raise OTAError(f"Device reported error: {response}") + else: + logger.warning(f"Unexpected response: {response}") + else: + raise OTAError("Timeout waiting for response") + + if progress_callback: + progress_callback(sent_bytes, size) + else: + print(f"[{sent_bytes / size * 100:5.1f}%] Sent {sent_bytes} of {size} bytes...", end="\r") + + if not progress_callback: + print() + + # Wait for OK from device + logger.info("Firmware sent, waiting for verification...") + while True: + response = notification_receiver.wait_for_next(timeout=10.0) + if response: + if response == "OK": + logger.info("OTA update completed successfully!") + break + if response.startswith("ERR "): + raise OTAError(f"OTA update failed: {response}") + else: + logger.warning(f"Unexpected final response: {response}") + else: + raise OTAError("Timeout waiting for response") + + finally: + # Terminating Bluetooth connection + try: + client.disconnect() + except Exception: + pass + client.close()