From 2cf910674e9554fc63a8462df58a0bef98e0bcc6 Mon Sep 17 00:00:00 2001 From: Daniel Schaefer Date: Sun, 12 Jul 2026 10:50:57 +0800 Subject: [PATCH 1/2] buspirate_ctrl: Parse fmap to flash single region Signed-off-by: Daniel Schaefer --- scripts/buspirate_ctrl.py | 153 ++++++++++++++++++++++++++++++++++---- 1 file changed, 137 insertions(+), 16 deletions(-) diff --git a/scripts/buspirate_ctrl.py b/scripts/buspirate_ctrl.py index 67db992..293408a 100755 --- a/scripts/buspirate_ctrl.py +++ b/scripts/buspirate_ctrl.py @@ -24,9 +24,12 @@ ./buspirate_ctrl.py --pty-bridge # PTY bridge (Ctrl+C to stop) ./buspirate_ctrl.py --pty-bridge --reset # Bridge, then reset (captures boot log) ./buspirate_ctrl.py --pty-bridge --enter-flash-mode # Bridge + enter flash mode - ./buspirate_ctrl.py --flash ./result/ # Full flash workflow + ./buspirate_ctrl.py --flash ./result/ # Full flash workflow (RO+RW) + ./buspirate_ctrl.py --flash ./result/ --section rw # Flash only the RW section + ./buspirate_ctrl.py --flash ./result/ --section ro # Flash only the RO section ./buspirate_ctrl.py --flash ./result/ --no-reset # Flash without reboot ./buspirate_ctrl.py --flash ./result/ --log # Flash, reset, print boot log + ./buspirate_ctrl.py --fmap ./result/ # Print ec.bin FMAP layout (no hardware) Signal control (while --pty-bridge is running): kill -USR1 # Toggle EC reset @@ -43,15 +46,13 @@ import struct import subprocess import sys +import tempfile import threading import time from pathlib import Path -import serial.tools.list_ports - -# Add upstream pybpio library to path -sys.path.insert(0, str(Path(__file__).parent / "BusPirate-BPIO2-flatbuffer-interface" / "python")) -from pybpio.bpio_client import BPIOClient +# NOTE: pyserial and pybpio are imported lazily inside the functions that +# talk to hardware, so offline commands (e.g. --fmap) work without them. # --------------------------------------------------------------------------- @@ -78,6 +79,8 @@ def find_bp5_binport(): interface string "Bus Pirate BIN". Falls back to sorted port list index 1 if interface strings are unavailable. """ + import serial.tools.list_ports + env_port = os.environ.get("BP5_BINPORT") if env_port: return env_port @@ -326,6 +329,11 @@ def gpio_release_all(bp): def setup_bp5(port, debug=False): """Open BPIO client, verify connection, enable PSU and UART.""" + import serial + sys.path.insert(0, str( + Path(__file__).parent / "BusPirate-BPIO2-flatbuffer-interface" / "python")) + from pybpio.bpio_client import BPIOClient + bp = BPIOClient(port, debug=debug) # Set write timeout so we don't block forever if BP5 isn't responding @@ -380,6 +388,80 @@ def cleanup_bp5(bp): pass +# --------------------------------------------------------------------------- +# Firmware image / flash section layout +# --------------------------------------------------------------------------- + +# FMAP header: signature + ver + base + size + name + nareas +_FMAP_HDR = struct.Struct("<8sBBQI32sH") +# FMAP area: offset + size + name + flags +_FMAP_AREA = struct.Struct("10} {'SIZE':>10} {'END':>10}") + for name, (off, size) in areas.items(): + print(f" {name:<22} {off:#010x} {size:#010x} {off + size:#010x}") + + +def section_bounds(data, section): + """Return (flash_offset, length) of the requested section within data. + + section is "all", "ro", or "rw". RO/RW bounds come from the image's + FMAP (WP_RO / EC_RW areas); if absent, fall back to a half-image split. + """ + if section == "all": + return 0, len(data) + + areas = parse_fmap(data) + if areas and "WP_RO" in areas and "EC_RW" in areas: + ro_off, ro_size = areas["WP_RO"] + rw_off, rw_size = areas["EC_RW"] + else: + half = len(data) // 2 + ro_off, ro_size = 0, half + rw_off, rw_size = half, len(data) - half + + if section == "ro": + return ro_off, ro_size + return rw_off, rw_size + + # --------------------------------------------------------------------------- # Command handlers # --------------------------------------------------------------------------- @@ -467,8 +549,13 @@ def cmd_log(bp, reset=False, debug=False): print("\nLog stopped.", file=sys.stderr) -def cmd_flash(bp, firmware_dir, no_reset=False, log=False, debug=False): - """Full flash workflow: enter flash mode, PTY bridge, uartupdatetool, reset.""" +def cmd_flash(bp, firmware_dir, section="all", no_reset=False, log=False, debug=False): + """Full flash workflow: enter flash mode, PTY bridge, uartupdatetool, reset. + + section selects which part of ec.bin to program: "all" (default), + "ro", or "rw". For "ro"/"rw" only that region is erased and written, + leaving the other region untouched. + """ fw_dir = Path(firmware_dir) ec_bin = fw_dir / "ec.bin" monitor_bin = fw_dir / "npcx_monitor.bin" @@ -476,7 +563,15 @@ def cmd_flash(bp, firmware_dir, no_reset=False, log=False, debug=False): if not ec_bin.exists() or not monitor_bin.exists(): sys.exit(f"Error: ec.bin and/or npcx_monitor.bin not found in {fw_dir}") + # Determine which slice of the image to program. + image = ec_bin.read_bytes() + flash_off, length = section_bounds(image, section) + payload = image[flash_off:flash_off + length] + print(f"Firmware: {ec_bin}") + print(f"Section: {section} " + f"(flash 0x{flash_off:06x}..0x{flash_off + len(payload):06x}, " + f"{len(payload)} bytes)") # Enter flash mode print("Entering EC flash mode...") @@ -499,12 +594,29 @@ def cmd_flash(bp, firmware_dir, no_reset=False, log=False, debug=False): check=True, ) - print("Flashing ec.bin...") - subprocess.run( - [tool, "--port", port_arg, "--opr", "wr", "--auto", - "--addr", "0x0000", "--file", str(ec_bin)], - check=True, - ) + if section == "all": + print("Flashing ec.bin...") + subprocess.run( + [tool, "--port", port_arg, "--opr", "wr", "--auto", + "--addr", "0x0000", "--file", str(ec_bin)], + check=True, + ) + else: + # Write only the selected region at its flash offset. uartupdatetool + # writes the whole --file, so hand it just the region's bytes. + with tempfile.NamedTemporaryFile( + suffix=f"_{section}.bin", delete=False) as tf: + tf.write(payload) + slice_path = tf.name + try: + print(f"Flashing {section} section...") + subprocess.run( + [tool, "--port", port_arg, "--opr", "wr", "--auto", + "--offset", f"0x{flash_off:x}", "--file", slice_path], + check=True, + ) + finally: + os.unlink(slice_path) print("Flash complete.") @@ -553,6 +665,8 @@ def main(): help="PTY bridge (blocks until Ctrl+C)") group.add_argument("--flash", metavar="DIR", help="Full flash workflow with uartupdatetool") + group.add_argument("--fmap", metavar="PATH", + help="Print the FMAP of an ec.bin (file or dir) and exit") # Combinable flags parser.add_argument("--reset", action="store_true", @@ -563,9 +677,16 @@ def main(): help="Enter EC flash mode before primary action") parser.add_argument("--no-reset", action="store_true", help="Skip reset after --flash") + parser.add_argument("--section", choices=["all", "ro", "rw"], default="all", + help="Which ec.bin region to flash (default: all)") args = parser.parse_args() + # Offline command: no BP5 hardware needed. + if args.fmap: + print_fmap(args.fmap) + return + if not (args.reset or args.reset_hold or args.pty_bridge or args.flash or args.log): parser.error("One of --reset, --reset-hold, --pty-bridge, --flash, or --log is required") @@ -601,8 +722,8 @@ def sigint_handler(signum, frame): elif args.reset_hold: cmd_reset_hold(bp) elif args.flash: - cmd_flash(bp, args.flash, no_reset=args.no_reset, - log=args.log, debug=args.debug) + cmd_flash(bp, args.flash, section=args.section, + no_reset=args.no_reset, log=args.log, debug=args.debug) elif args.log: cmd_log(bp, reset=args.reset, debug=args.debug) elif args.reset: From 49acf427087c6a2fff36f7c846ca8a82bb511b9e Mon Sep 17 00:00:00 2001 From: Daniel Schaefer Date: Mon, 13 Jul 2026 22:10:02 +0800 Subject: [PATCH 2/2] Dump flash NOT working. I think something is wrong in buspirate firmware. Signed-off-by: Daniel Schaefer --- scripts/buspirate_ctrl.py | 189 ++++++++++++++++++++++++++++++++++---- 1 file changed, 172 insertions(+), 17 deletions(-) diff --git a/scripts/buspirate_ctrl.py b/scripts/buspirate_ctrl.py index 293408a..70ad857 100755 --- a/scripts/buspirate_ctrl.py +++ b/scripts/buspirate_ctrl.py @@ -29,6 +29,7 @@ ./buspirate_ctrl.py --flash ./result/ --section ro # Flash only the RO section ./buspirate_ctrl.py --flash ./result/ --no-reset # Flash without reboot ./buspirate_ctrl.py --flash ./result/ --log # Flash, reset, print boot log + ./buspirate_ctrl.py --dump flash.bin # Dump EC's current flash to a file ./buspirate_ctrl.py --fmap ./result/ # Print ec.bin FMAP layout (no hardware) Signal control (while --pty-bridge is running): @@ -67,6 +68,16 @@ PTY_OVERFLOW_MAXLEN = 1024 * 1024 # 1MB max buffered data +SCRIPT_DIR = Path(__file__).parent.resolve() +UARTUPDATETOOL = str(SCRIPT_DIR / "uartupdatetool") +# npcx_monitor.bin is the flash-service stub loaded into EC SRAM. It is +# chip-level (npcx9), identical across boards, so a copy is bundled here +# and used when a firmware dir doesn't provide one. +BUNDLED_MONITOR = SCRIPT_DIR / "npcx_monitor.bin" + +# The NPCX bootrom loads the monitor to this SRAM address and executes it. +MONITOR_LOAD_ADDR = "0x200c3020" + # --------------------------------------------------------------------------- # Port detection @@ -121,13 +132,29 @@ class PtyBridge: we flush, preventing data loss. """ - def __init__(self, bp, debug=False): + def __init__(self, bp, debug=False, stats=False): self.bp = bp self.debug = debug + self.stats = stats self._shutdown = threading.Event() self._slave_ready = threading.Event() self._buf = collections.deque(maxlen=PTY_OVERFLOW_MAXLEN) + # Lightweight throughput counters (see print_stats). Maintained + # cheaply on the hot path; _buf_bytes mirrors bytes queued in _buf. + self._buf_bytes = 0 + self._s = { + 'bpio_bytes': 0, # bytes received from BPIO async UART + 'bpio_chunks': 0, # async DataResponse chunks received + 'pty_bytes': 0, # bytes written out to the PTY master + 'eagain': 0, # PTY-full events (consumer too slow) + 'max_buf_bytes': 0, # peak backlog queued toward the PTY + 'max_gap_ms': 0.0, # longest gap between async chunks + 'max_qdepth': 0, # peak depth of pybpio's async_queue + } + self._last_rx = None + self._last_report = None + # Create PTY pair self._master_fd, self._slave_fd = os.openpty() os.set_blocking(self._master_fd, False) @@ -146,6 +173,13 @@ def __init__(self, bp, debug=False): self._reader_thread.start() self._async_thread.start() + def _async_qsize(self): + """Depth of pybpio's async queue (0 if the internal API is absent).""" + try: + return self.bp._async_queue.qsize() + except Exception: + return 0 + def _flush_buf(self): """Try to flush Python buffer to PTY master.""" while self._buf: @@ -153,8 +187,11 @@ def _flush_buf(self): try: os.write(self._master_fd, chunk) self._buf.popleft() + self._buf_bytes -= len(chunk) + self._s['pty_bytes'] += len(chunk) except OSError as e: if e.errno == errno.EAGAIN: + self._s['eagain'] += 1 return # Kernel buffer full, retry later raise @@ -163,6 +200,9 @@ def _write_to_master(self, data): if not data: return self._buf.append(bytes(data)) + self._buf_bytes += len(data) + if self._buf_bytes > self._s['max_buf_bytes']: + self._s['max_buf_bytes'] = self._buf_bytes if self._slave_ready.is_set(): self._flush_buf() @@ -176,6 +216,26 @@ def _bpio_async_loop(self): pkt = self.bp.check_async_data(timeout=0.05) if pkt and pkt.get('data_read'): data = bytes(pkt['data_read']) + if self.stats: + now = time.monotonic() + self._s['bpio_bytes'] += len(data) + self._s['bpio_chunks'] += 1 + if self._last_rx is not None: + gap = (now - self._last_rx) * 1000.0 + if gap > self._s['max_gap_ms']: + self._s['max_gap_ms'] = gap + self._last_rx = now + qd = self._async_qsize() + if qd > self._s['max_qdepth']: + self._s['max_qdepth'] = qd + if self._last_report is None or now - self._last_report > 2.0: + self._last_report = now + print(f"[dump] rx {self._s['bpio_bytes']}B " + f"in {self._s['bpio_chunks']} chunks " + f"({self._s['bpio_bytes'] / max(1, self._s['bpio_chunks']):.1f} B/chunk), " + f"buf {self._buf_bytes}B, " + f"eagain {self._s['eagain']}, " + f"asyncq {qd}", file=sys.stderr) if self.debug: printable = ''.join( chr(b) if 0x20 <= b < 0x7f else '.' @@ -223,9 +283,8 @@ def _pty_reader_loop(self): # Wait for tio init (tcflush) to finish time.sleep(0.2) self._slave_ready.set() - buflen = sum(len(c) for c in self._buf) print(f"Reader connected, flushing " - f"{buflen} buffered bytes") + f"{self._buf_bytes} buffered bytes") if event & (select.POLLHUP | select.POLLERR): time.sleep(0.1) except OSError as e: @@ -236,6 +295,37 @@ def _pty_reader_loop(self): print(f"pty-reader error: {e}", file=sys.stderr) return + def print_stats(self, expected=None): + """Print throughput counters gathered during the session. + + expected is the payload size we hoped to receive from the EC; if + bpio_bytes falls short of it, bytes were lost at/before the BP5 + (device or USB) rather than in our host-side bridge. + """ + s = self._s + print("--- bridge stats ---", file=sys.stderr) + print(f" bytes from BPIO (UART RX): {s['bpio_bytes']} " + f"in {s['bpio_chunks']} chunks", file=sys.stderr) + print(f" bytes written to PTY: {s['pty_bytes']}", file=sys.stderr) + print(f" still queued in bridge: {self._buf_bytes}", file=sys.stderr) + print(f" peak bridge backlog: {s['max_buf_bytes']} bytes", + file=sys.stderr) + print(f" PTY-full (EAGAIN) events: {s['eagain']}", file=sys.stderr) + print(f" avg chunk size: " + f"{s['bpio_bytes'] / max(1, s['bpio_chunks']):.1f} bytes", + file=sys.stderr) + print(f" max gap between chunks: {s['max_gap_ms']:.1f} ms", + file=sys.stderr) + print(f" peak async-queue depth: {s['max_qdepth']}", file=sys.stderr) + if expected is not None: + delta = s['bpio_bytes'] - expected + note = ("device/USB-side loss (BP5 delivered fewer bytes than the " + "EC sent)" if delta < 0 else + "BP5 delivered the full payload; any corruption is " + "host-side (bridge/uartupdatetool timing)") + print(f" vs expected {expected}: {delta:+d} bytes -> {note}", + file=sys.stderr) + def stop(self): """Shut down the bridge.""" self._shutdown.set() @@ -366,7 +456,10 @@ def setup_bp5(port, debug=False): fw_maj = st.get('version_firmware_major', 0) fw_min = st.get('version_firmware_minor', 0) - print(f"Connected: FW v{fw_maj}.{fw_min}") + fw_hash = st.get('version_firmware_git_hash') + fw_date = st.get('version_firmware_date') + extra = f" ({fw_hash} {fw_date})" if (fw_hash or fw_date) else "" + print(f"Connected: FW v{fw_maj}.{fw_min}{extra}") print("Enabling PSU (3.3V for IO buffers)...") bp.configuration_request(psu_enable=True, psu_set_mv=3300) @@ -558,10 +651,14 @@ def cmd_flash(bp, firmware_dir, section="all", no_reset=False, log=False, debug= """ fw_dir = Path(firmware_dir) ec_bin = fw_dir / "ec.bin" - monitor_bin = fw_dir / "npcx_monitor.bin" - if not ec_bin.exists() or not monitor_bin.exists(): - sys.exit(f"Error: ec.bin and/or npcx_monitor.bin not found in {fw_dir}") + if not ec_bin.exists(): + sys.exit(f"Error: ec.bin not found in {fw_dir}") + + # Prefer the firmware dir's monitor; fall back to the bundled copy. + monitor_bin = fw_dir / "npcx_monitor.bin" + if not monitor_bin.exists(): + monitor_bin = BUNDLED_MONITOR # Determine which slice of the image to program. image = ec_bin.read_bytes() @@ -584,20 +681,17 @@ def cmd_flash(bp, firmware_dir, section="all", no_reset=False, log=False, debug= port_arg = pty_name.removeprefix("/dev/") print(f"PTY bridge: {pty_name}") - script_dir = os.path.dirname(os.path.abspath(__file__)) - tool = os.path.join(script_dir, "uartupdatetool") - print("Flashing monitor...") subprocess.run( - [tool, "--port", port_arg, "--opr", "wr", - "--addr", "0x200c3020", "--file", str(monitor_bin)], + [UARTUPDATETOOL, "--port", port_arg, "--opr", "wr", + "--addr", MONITOR_LOAD_ADDR, "--file", str(monitor_bin)], check=True, ) if section == "all": print("Flashing ec.bin...") subprocess.run( - [tool, "--port", port_arg, "--opr", "wr", "--auto", + [UARTUPDATETOOL, "--port", port_arg, "--opr", "wr", "--auto", "--addr", "0x0000", "--file", str(ec_bin)], check=True, ) @@ -611,7 +705,7 @@ def cmd_flash(bp, firmware_dir, section="all", no_reset=False, log=False, debug= try: print(f"Flashing {section} section...") subprocess.run( - [tool, "--port", port_arg, "--opr", "wr", "--auto", + [UARTUPDATETOOL, "--port", port_arg, "--opr", "wr", "--auto", "--offset", f"0x{flash_off:x}", "--file", slice_path], check=True, ) @@ -644,6 +738,58 @@ def cmd_flash(bp, firmware_dir, section="all", no_reset=False, log=False, debug= print("Done.") +def cmd_dump(bp, out_file, monitor=None, no_reset=False, debug=False): + """Dump the EC's current flash contents to a file. + + Reading flash needs npcx_monitor.bin loaded into SRAM first, same as + writing. Uses the bundled monitor unless one is given. + """ + monitor_bin = Path(monitor) if monitor else BUNDLED_MONITOR + if monitor_bin.is_dir(): + monitor_bin = monitor_bin / "npcx_monitor.bin" + if not monitor_bin.exists(): + sys.exit(f"Error: npcx_monitor.bin not found at {monitor_bin}") + out = Path(out_file) + + print(f"Monitor: {monitor_bin}") + print(f"Output: {out}") + + # Enter flash mode + print("Entering EC flash mode...") + gpio_enter_flash_mode(bp, debug=debug) + time.sleep(0.5) + + with PtyBridge(bp, debug=debug, stats=True) as bridge: + port_arg = bridge.pty_path.removeprefix("/dev/") + print(f"PTY bridge: {bridge.pty_path}") + + print("Flashing monitor...") + subprocess.run( + [UARTUPDATETOOL, "--port", port_arg, "--opr", "wr", + "--addr", MONITOR_LOAD_ADDR, "--file", str(monitor_bin)], + check=True, + ) + + print("Reading flash (this can take a while at 115200)...") + rc = subprocess.run( + [UARTUPDATETOOL, "--port", port_arg, "--read-flash", + "--file", str(out)], + ).returncode + + out_size = out.stat().st_size if out.exists() else None + bridge.print_stats(expected=out_size) + + if rc != 0: + print(f"WARNING: uartupdatetool exited {rc}; dump may be incomplete.", + file=sys.stderr) + print(f"Flash dumped to {out} ({out.stat().st_size} bytes)") + + if not no_reset: + print("Rebooting EC...") + gpio_reset(bp) + print("Done.") + + # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- @@ -665,6 +811,8 @@ def main(): help="PTY bridge (blocks until Ctrl+C)") group.add_argument("--flash", metavar="DIR", help="Full flash workflow with uartupdatetool") + group.add_argument("--dump", metavar="OUTFILE", + help="Dump the EC's current flash to OUTFILE") group.add_argument("--fmap", metavar="PATH", help="Print the FMAP of an ec.bin (file or dir) and exit") @@ -676,9 +824,11 @@ def main(): parser.add_argument("--enter-flash-mode", action="store_true", help="Enter EC flash mode before primary action") parser.add_argument("--no-reset", action="store_true", - help="Skip reset after --flash") + help="Skip reset after --flash/--dump") parser.add_argument("--section", choices=["all", "ro", "rw"], default="all", help="Which ec.bin region to flash (default: all)") + parser.add_argument("--monitor", metavar="PATH", default=None, + help="npcx_monitor.bin file or dir (default: bundled copy)") args = parser.parse_args() @@ -687,8 +837,10 @@ def main(): print_fmap(args.fmap) return - if not (args.reset or args.reset_hold or args.pty_bridge or args.flash or args.log): - parser.error("One of --reset, --reset-hold, --pty-bridge, --flash, or --log is required") + if not (args.reset or args.reset_hold or args.pty_bridge or args.flash + or args.dump or args.log): + parser.error("One of --reset, --reset-hold, --pty-bridge, --flash, " + "--dump, or --log is required") # Find port binmode_port = args.port or find_bp5_binport() @@ -724,6 +876,9 @@ def sigint_handler(signum, frame): elif args.flash: cmd_flash(bp, args.flash, section=args.section, no_reset=args.no_reset, log=args.log, debug=args.debug) + elif args.dump: + cmd_dump(bp, args.dump, monitor=args.monitor, + no_reset=args.no_reset, debug=args.debug) elif args.log: cmd_log(bp, reset=args.reset, debug=args.debug) elif args.reset: