Anycubic ACE 2 Pro — OTA Firmware Updater

ace2-ota-update.py

#!/usr/bin/env python3
"""
Anycubic ACE 2 Pro — OTA Firmware Updater

Implements the IAP (In-Application Programming) sequence reverse-engineered from the
Kobra S1 gklib binary (v2.7.0.9):

  1. IAP_UPGRADE (cmd 2)        — announce: total size + CRC-16/Kermit + version string
  2. IAP_FIRMWARE (cmd 3)       — send firmware in 64-byte chunks, each with its flash address
  3. IAP_UPGRADE_FINISH (cmd 4) — commit: ACE reboots and runs new firmware

Flash base address for ACE 2: 0x08024000 (IAP task, confirmed from ACE2 MCU disassembly)
Chunk size: 64 bytes (confirmed from gklib: `total = ceil(size / 64)`)

About the firmware file
-----------------------
gklib reads the firmware binary DIRECTLY with os.OpenFile + io.ReadAll — there is no
ota_local_info header parsing inside gklib. The file is expected to be a raw ARM image
for the OTA staging area (0x08024000). CRC-16/Kermit and size are computed over the
entire binary, and the version string is supplied via --version.

The .swu file from an Anycubic update package is a CPIO archive (SVR4 "newc" format).
This script auto-detects CPIO magic and extracts the ACE firmware binary from within.
gkapi (the Kobra S1 API layer) performs this extraction before handing the path to gklib.

Obtaining ACE 2 firmware:
  Option A — from a Kobra S1 OTA package (.swu file):
    Pass the .swu directly; the script extracts the ACE binary automatically.
    The relevant file inside the archive has "ace" or "filament_hub" in its name.

  Option B — capture from the printer over the network:
    Intercept HTTP(S) traffic while the Kobra S1 downloads an auto-update.
    The ACE firmware URL will appear in the printer's requests.

  Option C — SSH into the printer:
    Look in the OTA staging directory for the file gklib reads before flashing
    (path set in the printer config as fw_update_path for the filament hub).

Dependencies: pip install pyserial

Usage examples:
  # Flash a raw .bin binary:
  python ace2-ota-update.py COM3 ACE2_V1.1.31.bin --version 1.1.31

  # Extract from .swu ZIP archive and flash (with MD5 and password):
  python ace2-ota-update.py COM3 update.swu --version 1.1.31 --md5 f7968b51... --swu-password "U2FsdGVk..."

  # Dry-run: query version and parse firmware, then exit without flashing:
  python ace2-ota-update.py COM3 firmware.bin --version 1.1.31 --dry-run

  # Force flash even if version already matches:
  python ace2-ota-update.py COM3 firmware.bin --version 1.1.31 --force

  # Linux example:
  python ace2-ota-update.py /dev/ttyCH343USB0 firmware.bin --version 1.1.31
"""

import serial
import struct
import time
import sys
import argparse
import hashlib
import zipfile
import tarfile
import io

# ═══════════════════════════════════════════════════════════════════
# PROTOCOL CONSTANTS
# ═══════════════════════════════════════════════════════════════════

BAUD              = 230400
PREAMBLE          = b'\xff\xaa'
END_MARKER        = 0xFE
FLAG_REQUEST      = 0x00
FLAG_RESPONSE     = 0x80

CMD_DISCOVER      = 0
CMD_ASSIGN_ID     = 1
CMD_IAP_UPGRADE   = 2
CMD_IAP_FIRMWARE  = 3
CMD_IAP_FINISH    = 4
CMD_IAP_VERSION   = 5
CMD_GET_INFO      = 7

# ACE 2 OTA staging area — confirmed from IAP task disasm (0x8013FB8): MOVT #0x802 MOVW #0x4000
# CMD3 accepts first chunk only when address == 0x08024000; CRC verified over this region.
ACE2_FLASH_BASE   = 0x08024000

# Max protobuf payload per UART frame — enforced by BuildRequest in gklib (0x60e19c)
MAX_FRAME_PAYLOAD = 100

# Firmware data bytes per chunk.
# Confirmed from gklib disasm: total_chunks = ceil(file_size / 64), each chunk is 64 bytes.
# FirmwareRequest protobuf overhead: ~8 bytes → total payload ~72 bytes, well within 100.
CHUNK_SIZE        = 64

# Timeouts (seconds) matching gklib values
T_START   = 2.0   # OtaStart timeout
T_CHUNK   = 2.0   # OtaSendChunk timeout (now an upper bound; send_recv returns on the ack)
CHUNK_DELAY = 0.015  # deliberate pause after each acked chunk; see the chunk loop
T_FINISH  = 5.0   # OtaFinish timeout

# ═══════════════════════════════════════════════════════════════════
# CRC
# ═══════════════════════════════════════════════════════════════════

def crc16_kermit(data: bytes) -> int:
    """CRC-16/KERMIT — poly 0x8408, init 0xFFFF."""
    crc = 0xFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1
    return crc & 0xFFFF

# ═══════════════════════════════════════════════════════════════════
# MINIMAL PROTOBUF ENCODER
# ═══════════════════════════════════════════════════════════════════

def _varint(v: int) -> bytes:
    r = bytearray()
    while v > 0x7F:
        r.append((v & 0x7F) | 0x80)
        v >>= 7
    r.append(v & 0x7F)
    return bytes(r)

def _field_uint32(field: int, value: int) -> bytes:
    return _varint((field << 3) | 0) + _varint(value)

def _field_bytes(field: int, data: bytes) -> bytes:
    return _varint((field << 3) | 2) + _varint(len(data)) + data

def _field_string(field: int, text: str) -> bytes:
    return _field_bytes(field, text.encode())

def encode_upgrade_request(size: int, image_crc: int, version: str) -> bytes:
    """UpgradeRequest { size=1, crc=2, version=3 } — crc is CRC-16/Kermit, stored as uint32."""
    return _field_uint32(1, size) + _field_uint32(2, image_crc) + _field_string(3, version)

def encode_firmware_request(address: int, chunk: bytes) -> bytes:
    """FirmwareRequest { address=1, firmware=2 }"""
    return _field_uint32(1, address) + _field_bytes(2, chunk)

# ═══════════════════════════════════════════════════════════════════
# PACKET FRAMING
# ═══════════════════════════════════════════════════════════════════

def build_packet(cmd: int, payload: bytes, seq: int) -> bytes:
    plen = len(payload)
    if plen > MAX_FRAME_PAYLOAD:
        raise ValueError(f"Payload {plen} bytes exceeds frame limit {MAX_FRAME_PAYLOAD}")
    inner = bytearray([FLAG_REQUEST, seq & 0xFF, (seq >> 8) & 0xFF, cmd & 0xFF, plen & 0xFF])
    inner.extend(payload)
    crc = crc16_kermit(bytes(inner))
    return bytes(PREAMBLE + inner + bytes([crc & 0xFF, (crc >> 8) & 0xFF, END_MARKER]))

def parse_response(buf: bytearray):
    """Returns (packet_dict, bytes_consumed) or (None, 0/advance)."""
    while len(buf) >= 2:
        idx = buf.find(PREAMBLE)
        if idx < 0:
            return None, max(0, len(buf) - 1)
        if idx > 0:
            return None, idx
        if len(buf) < 10:
            return None, 0
        for end in range(9, min(len(buf), 270)):
            if buf[end] != END_MARKER:
                continue
            flags, seq = buf[2], buf[3] | (buf[4] << 8)
            cmd, plen = buf[5], buf[6]
            exp = 7 + plen + 2
            if end != exp:
                continue
            inner = bytes(buf[2:7 + plen])
            crc_recv = buf[7 + plen] | (buf[8 + plen] << 8)
            if crc_recv != crc16_kermit(inner):
                return None, end + 1
            return {
                "cmd": cmd,
                "is_resp": bool(flags & 0x80),
                "seq": seq,
                "payload": bytes(buf[7:7 + plen]),
            }, end + 1
        return None, 2 if len(buf) > 270 else 0
    return None, 0

def decode_varint(data: bytes, pos: int):
    result, shift = 0, 0
    while pos < len(data):
        b = data[pos]; pos += 1
        result |= (b & 0x7F) << shift
        if not (b & 0x80):
            return result, pos
        shift += 7
    return result, pos

def pb_decode(data: bytes) -> dict:
    fields, pos = {}, 0
    while pos < len(data):
        tag, pos = decode_varint(data, pos)
        fnum, wtype = tag >> 3, tag & 7
        if wtype == 0:
            val, pos = decode_varint(data, pos)
        elif wtype == 2:
            ln, pos = decode_varint(data, pos)
            val = data[pos:pos + ln]; pos += ln
        elif wtype == 5:
            val = struct.unpack_from('<f', data, pos)[0]; pos += 4
        elif wtype == 1:
            val = struct.unpack_from('<d', data, pos)[0]; pos += 8
        else:
            break
        fields.setdefault(fnum, []).append((wtype, val))
    return fields

def get_field(fields: dict, num: int, default=0):
    return fields.get(num, [(0, default)])[0][1]

# ═══════════════════════════════════════════════════════════════════
# TRANSPORT
# ═══════════════════════════════════════════════════════════════════

class ACE2Transport:
    def __init__(self, port: str):
        self.ser = serial.Serial(port, BAUD, timeout=0.1)
        self._seq = 0

    def close(self):
        self.ser.close()

    def _next_seq(self) -> int:
        self._seq = (self._seq % 0xFFFF) + 1
        return self._seq

    def send_recv(self, cmd: int, payload: bytes, timeout: float) -> list:
        seq = self._next_seq()
        pkt = build_packet(cmd, payload, seq)
        self.ser.reset_input_buffer()
        self.ser.write(pkt)
        self.ser.flush()

        buf = bytearray()
        results = []
        deadline = time.time() + timeout
        while time.time() < deadline:
            if self.ser.in_waiting:
                buf.extend(self.ser.read(self.ser.in_waiting))
                while len(buf) > 4:
                    p, n = parse_response(buf)
                    if n > 0:
                        buf = buf[n:]
                    else:
                        break
                    # The BOOTLOADER answers without the 0x80 response bit (observed
                    # 2026-08-28: GET_INFO from IAP mode returns flags=1, while the
                    # application returns flags=0x81). Filtering on is_resp made the
                    # flasher deaf to the bootloader -- i.e. deaf exactly when it is
                    # needed for recovery. Match on the command id and accept either.
                    if p and p["cmd"] == cmd and (p["is_resp"] or p["payload"] != payload):
                        results.append(p)
                # Return as soon as the device has answered. The original loop always ran
                # to the full timeout, so every one of the ~1119 firmware chunks cost the
                # whole T_CHUNK (2.0s) even though the ACE acks in milliseconds -- that is
                # the entire 37-minute flash time, almost all of it idle.
                if results:
                    return results
            else:
                time.sleep(0.002)
        return results

# ═══════════════════════════════════════════════════════════════════
# VERSION QUERY
# ═══════════════════════════════════════════════════════════════════

def get_ace_version(transport: ACE2Transport) -> tuple[str, str] | None:
    """Returns (version, boot_version) or None."""
    results = transport.send_recv(CMD_GET_INFO, b'', timeout=2.0)
    for r in results:
        f = pb_decode(r["payload"])
        if 1 in f:
            version = get_field(f, 1, b'').decode(errors='replace')
            boot    = get_field(f, 2, b'').decode(errors='replace')
            return version, boot
    return None

# ═══════════════════════════════════════════════════════════════════
# FIRMWARE FILE PARSING
# ═══════════════════════════════════════════════════════════════════

# Keywords that identify the ACE/filament-hub firmware file inside archives
_ACE_KEYWORDS = ('ace', 'filament_hub', 'filament-hub')

# CPIO newc (SVR4) constants — older/alternative swu format
_CPIO_MAGIC   = (b'070701', b'070702')
_CPIO_HDR_LEN = 110


def _find_in_tar(tar_bytes: bytes, mode: str) -> tuple[bytes, str] | None:
    """Extract the first ACE firmware .bin from a tar archive."""
    with tarfile.open(fileobj=io.BytesIO(tar_bytes), mode=mode) as tf:
        for member in tf.getmembers():
            if member.isfile() and any(kw in member.name.lower() for kw in _ACE_KEYWORDS):
                f = tf.extractfile(member)
                if f:
                    return f.read(), member.name
    return None


def _extract_from_zip_swu(data: bytes, password: str | None) -> tuple[bytes, str] | None:
    """Extract ACE firmware from a ZIP-format .swu (Anycubic KS1/S1 packaging).

    Archive layout (confirmed from gkapi disassembly at 0x482360):
      swu.zip  (AES/ZipCrypto encrypted, password from cloud OTA notification)
      └── update_swu/
          └── setup.tar.gz   (or setup.tar)
              └── ACE2_*.bin
    """
    pwd_bytes = password.encode() if password else None
    with zipfile.ZipFile(io.BytesIO(data)) as zf:
        # Locate the inner tarball (gkapi tries setup.tar first, then setup.tar.gz)
        tar_name = next(
            (n for n in zf.namelist() if n.endswith(('setup.tar.gz', 'setup.tar'))),
            None,
        )
        if tar_name is None:
            return None
        tar_bytes = zf.read(tar_name, pwd=pwd_bytes)
        mode = 'r:gz' if tar_name.endswith('.gz') else 'r:'
        return _find_in_tar(tar_bytes, mode)


def _extract_from_cpio_swu(data: bytes) -> tuple[bytes, str] | None:
    """Walk a CPIO newc archive and return (content, filename) for the ACE firmware entry."""
    pos = 0
    while pos + _CPIO_HDR_LEN <= len(data):
        hdr = data[pos:pos + _CPIO_HDR_LEN]
        if hdr[:6] not in _CPIO_MAGIC:
            return None
        filesize = int(hdr[54:62], 16)
        namesize = int(hdr[94:102], 16)
        pos += _CPIO_HDR_LEN
        name = data[pos:pos + namesize - 1].decode(errors='replace')
        pos = (pos + namesize + 3) & ~3  # align to 4-byte boundary
        if name == 'TRAILER!!!':
            break
        file_data = data[pos:pos + filesize]
        pos = (pos + filesize + 3) & ~3
        if any(kw in name.lower() for kw in _ACE_KEYWORDS):
            return file_data, name
    return None


class FirmwareImage:
    def __init__(self, data: bytes, version: str, image_crc: int):
        self.data      = data
        self.version   = version
        self.image_crc = image_crc  # CRC-16/Kermit — verified by the MCU after flashing


def load_firmware(path: str, version: str,
                  expected_md5: str | None = None,
                  swu_password: str | None = None) -> FirmwareImage:
    """
    Load the ACE 2 firmware image.

    gklib opens the final ARM binary with os.OpenFile + io.ReadAll, computes CRC-16/Kermit
    over the full file, and sends it in 64-byte chunks. gkapi handles archive extraction
    before gklib is involved.

    Supported input formats (auto-detected by magic bytes):
      • Raw ARM binary (.bin)        — used directly
      • ZIP archive (.swu, KS1/S1)   — unzip with optional password → setup.tar.gz → .bin
      • CPIO newc archive (.swu)      — walk archive → .bin  (older format, fallback)

    MD5 is checked over the raw input file before any extraction (matching gkapi behaviour).
    """
    with open(path, 'rb') as f:
        raw_bytes = f.read()

    # MD5 check over the raw file — same point gkapi validates before calling gklib
    if expected_md5 is not None:
        actual_md5 = hashlib.md5(raw_bytes).hexdigest()
        if actual_md5.lower() != expected_md5.lower():
            raise ValueError(
                f"MD5 mismatch!\n"
                f"  expected: {expected_md5.lower()}\n"
                f"  actual:   {actual_md5}"
            )
        print(f"[firmware] MD5 OK: {actual_md5}")

    image_data = raw_bytes

    if raw_bytes[:2] == b'PK':
        # ZIP-format .swu (Anycubic KS1/S1): ZIP → setup.tar.gz → .bin
        result = _extract_from_zip_swu(raw_bytes, swu_password)
        if result is None:
            raise ValueError(
                "File is a ZIP archive but no ACE firmware entry found inside setup.tar.gz. "
                "Check that --swu-password is correct and the archive contains an ACE binary."
            )
        image_data, source = result
        print(f"[firmware] Extracted from ZIP .swu: '{source}'  ({len(image_data)} bytes)")

    elif raw_bytes[:6] in _CPIO_MAGIC:
        # CPIO-format .swu (fallback / older format)
        result = _extract_from_cpio_swu(raw_bytes)
        if result is None:
            raise ValueError(
                "File is a CPIO archive but no ACE firmware entry found. "
                "Expected a file with 'ace' or 'filament_hub' in its name."
            )
        image_data, source = result
        print(f"[firmware] Extracted from CPIO .swu: '{source}'  ({len(image_data)} bytes)")

    image_crc = crc16_kermit(image_data)
    print(f"[firmware] {len(image_data)} bytes  CRC16/Kermit=0x{image_crc:04X}  version={version}")
    return FirmwareImage(image_data, version, image_crc)

# ═══════════════════════════════════════════════════════════════════
# IAP FLASH SEQUENCE
# ═══════════════════════════════════════════════════════════════════

def iap_upgrade(transport: ACE2Transport, fw: FirmwareImage, verbose: bool) -> bool:
    total   = len(fw.data)
    n_chunks = (total + CHUNK_SIZE - 1) // CHUNK_SIZE

    # ── Step 1: IAP_UPGRADE ─────────────────────────────────────────
    print(f"\n[step 1/3] IAP_UPGRADE  size={total}  crc16=0x{fw.image_crc:04X}  version={fw.version}")
    payload = encode_upgrade_request(total, fw.image_crc, fw.version)
    if verbose:
        print(f"  payload: {payload.hex()}")
    results = transport.send_recv(CMD_IAP_UPGRADE, payload, timeout=T_START)
    if not results:
        print("[ERROR] No response to IAP_UPGRADE — is the ACE connected and powered?")
        return False
    resp = results[0]
    code = get_field(pb_decode(resp["payload"]), 1, 0)
    if code != 0:
        print(f"[ERROR] IAP_UPGRADE rejected: code={code}")
        return False
    print("  ACE accepted upgrade announcement.")

    # ── Step 2: IAP_FIRMWARE chunks ─────────────────────────────────
    print(f"\n[step 2/3] IAP_FIRMWARE  {n_chunks} chunks × {CHUNK_SIZE} bytes  (base=0x{ACE2_FLASH_BASE:08X})")
    for i in range(n_chunks):
        offset = i * CHUNK_SIZE
        chunk  = fw.data[offset:offset + CHUNK_SIZE]
        addr   = ACE2_FLASH_BASE + offset
        payload = encode_firmware_request(addr, chunk)

        if len(payload) > MAX_FRAME_PAYLOAD:
            print(f"[ERROR] Chunk {i}: protobuf payload {len(payload)} exceeds {MAX_FRAME_PAYLOAD} bytes. Reduce CHUNK_SIZE.")
            return False

        results = transport.send_recv(CMD_IAP_FIRMWARE, payload, timeout=T_CHUNK)
        if not results:
            print(f"\n[ERROR] No response for chunk {i} (addr=0x{addr:08X})")
            return False
        code = get_field(pb_decode(results[0]["payload"]), 1, 0)
        if code != 0:
            print(f"\n[ERROR] Chunk {i} rejected at addr=0x{addr:08X}: code={code}")
            return False

        # The MCU may ack a chunk before its flash write has settled. The original script hid
        # this behind a 2s always-wait loop; now that we return on the ack, keep a deliberate
        # margin instead of hammering the flash controller. 15ms x 1120 chunks = ~17s total.
        if CHUNK_DELAY:
            time.sleep(CHUNK_DELAY)

        # Progress bar
        pct = (i + 1) / n_chunks * 100
        bar = int(pct / 2)
        print(f"\r  [{('█' * bar):<50s}] {pct:5.1f}%  chunk {i+1}/{n_chunks}", end='', flush=True)

    print()  # newline after progress bar
    print("  All chunks transferred.")

    # ── Step 3: IAP_UPGRADE_FINISH ───────────────────────────────────
    print(f"\n[step 3/3] IAP_UPGRADE_FINISH")
    results = transport.send_recv(CMD_IAP_FINISH, b'', timeout=T_FINISH)
    if not results:
        # A non-response here may be normal — ACE reboots immediately.
        print("  No response (ACE may have rebooted — this is normal).")
    else:
        code = get_field(pb_decode(results[0]["payload"]), 1, 0)
        print(f"  Response code: {code}")
    return True

# ═══════════════════════════════════════════════════════════════════
# POST-UPDATE VERSION VERIFY
# ═══════════════════════════════════════════════════════════════════

def _norm_ver(v: str) -> str:
    """Strip leading 'V'/'v' so '1.1.31' and 'V1.1.31' compare equal."""
    return v.lstrip('Vv')


def wait_for_version(transport: ACE2Transport, expected: str) -> bool:
    print(f"\n[verify] Waiting for ACE to reboot and report version {expected}...")
    deadline = time.time() + 30.0
    while time.time() < deadline:
        time.sleep(2.0)
        result = get_ace_version(transport)
        if result:
            ver, boot = result
            print(f"  ACE reports: version={ver}  boot_version={boot}")
            if _norm_ver(ver) == _norm_ver(expected):
                print("  Version matches.")
                return True
            print(f"  (expected {expected}, still waiting...)")
        else:
            print("  (no response yet)")
    print("[WARNING] Timed out waiting for expected version. Check ACE manually.")
    return False

# ═══════════════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════════════

def main():
    global CHUNK_SIZE

    parser = argparse.ArgumentParser(
        description="ACE 2 Pro OTA firmware updater",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("port",     help="Serial port (e.g. COM3 or /dev/ttyCH343USB0)")
    parser.add_argument("firmware", help="Path to ACE 2 firmware (.bin) or update archive (.swu)")
    parser.add_argument("--version", metavar="X.Y.Z", required=True,
                        help="Version string to send in IAP_UPGRADE (e.g. 1.1.31)")
    parser.add_argument("--md5", metavar="HASH",
                        help="Expected MD5 hex digest of the firmware file; aborts if mismatch")
    parser.add_argument("--swu-password", metavar="PASS",
                        help="Password for encrypted ZIP .swu archive (from Anycubic OTA notification)")
    parser.add_argument("--force",  action="store_true",
                        help="Flash even if ACE already reports the target version")
    parser.add_argument("--dry-run", action="store_true",
                        help="Query version and parse firmware, then exit without flashing")
    parser.add_argument("--chunk-size", type=int, default=CHUNK_SIZE, metavar="N",
                        help=f"Firmware bytes per chunk (default: {CHUNK_SIZE}; max safe: ~90)")
    parser.add_argument("--verbose", action="store_true",
                        help="Print raw payload hex for each command")
    args = parser.parse_args()

    CHUNK_SIZE = args.chunk_size

    # ── Load firmware file ──────────────────────────────────────────
    print(f"[firmware] Loading: {args.firmware}")
    fw = load_firmware(args.firmware, args.version,
                       expected_md5=args.md5, swu_password=args.swu_password)
    n_chunks = (len(fw.data) + CHUNK_SIZE - 1) // CHUNK_SIZE
    print(f"[firmware] Will send {n_chunks} chunks of {CHUNK_SIZE} bytes each")

    # Validate chunk payload size
    test_payload = encode_firmware_request(ACE2_FLASH_BASE, b'\x00' * CHUNK_SIZE)
    if len(test_payload) > MAX_FRAME_PAYLOAD:
        print(f"[ERROR] chunk-size={CHUNK_SIZE} produces {len(test_payload)}-byte protobuf payload, "
              f"exceeding {MAX_FRAME_PAYLOAD}-byte frame limit. Use a smaller --chunk-size.")
        sys.exit(1)

    # ── Connect ─────────────────────────────────────────────────────
    print(f"\n[serial] Connecting to {args.port} at {BAUD} baud...")
    transport = ACE2Transport(args.port)
    print("[serial] Connected.")

    try:
        # ── Query current version ───────────────────────────────────
        print("\n[version] Querying current ACE version...")
        current = get_ace_version(transport)
        if current:
            print(f"  Current: version={current[0]}  boot_version={current[1]}")
        else:
            print("  No response. ACE may not be connected or initialised.")

        if args.dry_run:
            print("\n[dry-run] Exiting without flashing.")
            return

        # ── Version check ───────────────────────────────────────────
        if current and _norm_ver(current[0]) == _norm_ver(fw.version) and not args.force:
            print(f"\n[skip] ACE already reports version {current[0]}. Use --force to flash anyway.")
            return

        # ── Confirm ─────────────────────────────────────────────────
        current_ver = current[0] if current else "unknown"
        print(f"\n  About to flash: {current_ver}  →  {fw.version}")
        print(f"  Image: {len(fw.data)} bytes  CRC16=0x{fw.image_crc:04X}")
        try:
            confirm = input("  Proceed? [y/N] ").strip().lower()
        except EOFError:
            confirm = 'y'
        if confirm not in ('y', 'yes'):
            print("Aborted.")
            return

        # ── Flash ───────────────────────────────────────────────────
        print()
        started = time.time()
        ok = iap_upgrade(transport, fw, args.verbose)
        elapsed = time.time() - started

        if ok:
            print(f"\n[done] Flash complete in {elapsed:.1f}s")
            wait_for_version(transport, fw.version)
        else:
            print("\n[FAILED] Firmware update did not complete successfully.")

    except KeyboardInterrupt:
        print("\n[interrupted]")
    finally:
        transport.close()
        print("[serial] Disconnected.")


if __name__ == "__main__":
    main()

Thank you for this — I have flashed an ACE 2 Pro many times with it while building custom firmware
for the unit, and it did the job every time. Two small bugs turned up along the way; a patch is at
https://github.com/Simon-CR/ace2-pro-firmware-research (tools/ota_updater.patch), deliberately
as a patch rather than a fork so credit and updates stay here.

1. Flashes take ~37 minutes when they could take ~27 seconds

send_recv loops until its full timeout even after the device has answered:

deadline = time.time() + timeout
while time.time() < deadline:
    if self.ser.in_waiting:
        ...
        if p and p["is_resp"] and p["cmd"] == cmd:
            results.append(p)          # collected, but the loop keeps running
    else:
        time.sleep(0.005)
return results

With T_CHUNK = 2.0 and ~1119 chunks that is exactly the observed flash time, essentially all of
it idle waiting. Returning as soon as a matching reply arrives gives a measured 26.5 s flash
for the same 71 592-byte image.

I kept a deliberate 15 ms pause after each acked chunk, in case the MCU acks before its flash
write has settled — 1120 × 15 ms ≈ 17 s of margin and still two orders of magnitude better.

2. The tool cannot hear the bootloader — which matters most during recovery

Replies are filtered on the 0x80 response bit:

"is_resp": bool(flags & 0x80),
...
if p and p["is_resp"] and p["cmd"] == cmd:

But the bootloader answers with flags = 1, not 0x81. The application uses 0x81, so this
is invisible in normal use — and then, when a device is sitting in IAP mode and you need to
re-flash it, the tool reports:

[version] Querying current ACE version...
  No response. ACE may not be connected or initialised.
[ERROR] No response to IAP_UPGRADE — is the ACE connected and powered?

The device is perfectly healthy and answering; the tool is just discarding its replies. I hit this
twice and briefly believed I had destroyed the unit. (For reference, in IAP mode GET_INFO
returns boot_version = V1.0.2, status = upgrading, and version = whatever string the host
announced in IAP_UPGRADE.)

Matching on the command id instead of the response bit fixes it.

Also worth knowing, though not a bug: the post-flash verification compares against the string
passed to --version, so flashing an image whose internal version string differs always ends with
a spurious "Timed out waiting for expected version" after a completely successful flash.

And one device-side gotcha that looks like a tool bug but is not: IAP_UPGRADE on the MCU
(0x0800D4E4) begins ldrb r2,[r5,#24]; cbnz r2, skip — if an OTA is already pending, the entire
handler is skipped, and because the result code was zeroed at entry the host still sees SUCCESS.
IAP_UPGRADE_FINISH likewise only arms the commit when that state byte is exactly 2, returning 0
either way. So one failed attempt silently poisons every later one until the ACE is power-cycled,
and the tool reports three green steps throughout. Power-cycling before each flash attempt makes
it reliable.

Thanks again — without your protocol work and firmware analysis none of what I have been doing
would have been possible.

@Simon-CR thank you. I have applied your patch to the Gist code. Anyway, feel free to use the code in those Gists as you like - no need for attribution and no need to create patches for the code here. You can just use full code.

BTW, @DnG-Crafts made a GUI utility for flashing ACE2 firmware which doesn't have the bug in my original code: DnG-Crafts/U1-Ace#21 (comment)

---

I saw that you made some changes to the ACE 2 firmware to make it compatible with Bambu and Snapmaker tags. Did you also create Klipper ACE 2 overlay which utilizes those firmware changes?

Just for reference, here is the firmware I am currently using: https://github.com/hakimio/SnapmakerU1-Extended-Firmware

Changes compared to the original: paxx12-snapmaker-u1/SnapmakerU1-Extended-Firmware@develop...hakimio:SnapmakerU1-Extended-Firmware:ace-2

Its not only compatible, it writes Anycubic and Openspool on the ACE2

@Simon-CR Thank you very much for your work!!

17:38:34 
$ ACE_TAG_READ ACE=0 SLOT=0
17:38:34 
// [multiACE] tag read started (ACE 1 slot 1)
17:38:34 
[multiACE] rc522: searching for the tag (rotating up to 600 mm)
17:38:59 
[multiACE] rc522: card present (SELECT ok), UID=535B2DF4620001
17:38:59 
[multiACE] [spool] tag '535B2DF4620001' on ACE 1 slot 1 matches no table entry
17:38:59 
[multiACE] rc522: UID 535B2DF4620001 not in the table (a Spoolman spool with this card_uid will adopt on the next sweep)
17:39:01 
[multiACE] rc522: lane restored (20 mm)
17:39:15 
$ ACE_TAG_WRITE ACE=0 SLOT=0 MATERIAL=PLA COLOR=1188FF DUMP=1
17:39:15 
// [multiACE] tag write started (ACE 1 slot 1, PLA #1188FF)
17:39:16 
[multiACE] rc522: searching for the tag to write (up to 600 mm)
17:39:42 
[multiACE] rc522: writing OpenSpool to UID 535B2DF4620001 (27 pages). backup=0368d210556170706c69636174696f6e2f6a736f6e7b2270726f746f636f6c223a226f70656e73706f6f6c222c2276657273696f6e223a22312e30222c2274797065223a22504c41222c22636f6c6f725f686578223a22393235423542222c226272616e64223a22222c226d696e5f74
17:40:31 
[multiACE] rc522: write VERIFIED - PLA #1188FF (UID 535B2DF4620001)
17:40:33 
[multiACE] rc522: lane restored (20 mm)
17:40:49 
$ ACE_TAG_READ ACE=0 SLOT=0
17:40:49 
// [multiACE] tag read started (ACE 1 slot 1)
17:40:50 
[multiACE] rc522: searching for the tag (rotating up to 600 mm)
17:41:15 
[multiACE] rc522: card present (SELECT ok), UID=535B2DF4620001
17:41:15 
[multiACE] rc522: OpenSpool tag: PLA - #1188FF (UID 535B2DF4620001)
17:41:15 
[multiACE] [spool] tag '535B2DF4620001' on ACE 1 slot 1 matches no table entry
17:41:17 
[multiACE] rc522: lane restored (20 mm)

@decay71 Hey, following up on your ACE2 work—I recently deployed multiACE on a single-extruder setup and put together some thoughts on offloading tasks to the ACE MCU to save U1 host CPU, plus a few zero-load web UI additions. Opened an issue with the details over on the repo: decay71/multiACE#125. Let me know what you think whenever you get a chance!