Files
workflow-flash-with-uboot/scripts/serial_command.py
T
2026-11-09 20:53:02 +03:30

188 خطوط
9.5 KiB
Python

"""Check an existing U-Boot prompt or configure its volatile environment."""
import argparse
import fcntl
import re
import sys
import time
from pathlib import Path
import serial
# Match the complete final line, never a prompt embedded in command output.
PROMPT = re.compile(r"(?:^|\n)([^\r\n]*=>) ?$")
ERROR = re.compile(
r"Unknown command|## Error|^Error:|^Usage:|^.*(?:not defined|not found|failed|failure).*$",
re.IGNORECASE | re.MULTILINE
)
class UBoot:
def __init__(self, port):
self.port = port
self.prompt = None
def read_prompt(self, timeout):
output = ""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
data = self.port.read(self.port.in_waiting or 1)
if data:
chunk = data.decode("utf-8", errors="replace").replace("\r", "")
print(chunk, end="", flush=True)
output += chunk
match = PROMPT.search(output)
if match and (self.prompt is None or match[1] == self.prompt):
# Wait for a quiet prompt rather than a partial output line.
time.sleep(0.15)
if self.port.in_waiting:
continue
return output, match[1]
raise RuntimeError("Timed out waiting for the CLI prompt")
def detect(self):
# Observe before requesting a fresh prompt.
try:
_, candidate = self.read_prompt(3)
except RuntimeError:
candidate = None
if candidate is None:
# An idle prompt may have been printed before the port was opened.
# A newline only requests a fresh prompt; no U-Boot command is sent.
self.port.write(b"\r")
self.port.flush()
_, candidate = self.read_prompt(5)
if not candidate.endswith("=>"):
raise RuntimeError("Response is not a recognized U-Boot CLI prompt")
self.prompt = candidate
print("\nU-Boot-style CLI prompt detected: " + candidate, flush=True)
def command(self, command, expected=None, timeout=10, allow_missing=()):
print("\n>>> " + command, flush=True)
self.port.write((command + "\r").encode())
self.port.flush()
output, _ = self.read_prompt(timeout)
# Query immediately, before another command changes the return status.
self.port.write(b"echo __UBOOT_STATUS_ $?\r")
self.port.flush()
status, _ = self.read_prompt(5)
codes = re.findall(r"(?m)^__UBOOT_STATUS_ (\d+) *$", status)
# Only the initial environment inspection may report unset variables.
missing_pattern = r'^## Error: "(' + "|".join(map(re.escape, allow_missing)) + r')" not defined$'
missing = re.findall(missing_pattern, output, re.MULTILINE) if allow_missing else []
checked_output = re.sub(missing_pattern, "", output, flags=re.MULTILINE) if missing else output
if codes != ["0"] and not (codes == ["1"] and missing):
raise RuntimeError("Command failed or its exit status could not be verified: " + command)
if ERROR.search(checked_output):
raise RuntimeError("Error in command response: " + command)
if expected and not re.search(expected, output, re.MULTILINE):
raise RuntimeError("Expected response missing: " + command)
if missing:
print("\nInitially unset: " + ", ".join(missing), flush=True)
print("\nVerified: " + command, flush=True)
return output
def transfer(cli, firmware, *, dtb=False, kernel=False, rootfs=False, rootfs_size=None):
address = 0x41000000 if (dtb or rootfs) else 0x42000000
offset = 0x310000 if kernel else (0x200000 if dtb else 0)
limit = 0x500000 if kernel else (0x110000 if dtb else 0x200000)
dump_size = 64 if kernel else 32
filename = "zImage" if kernel else ("sun8i-v3s-licheepi-zero-dock.dtb" if dtb else "u-boot-sunxi-with-spl.bin")
if rootfs:
if rootfs_size is None or rootfs_size < 64:
raise RuntimeError("Actual RootFS partition size is required (at least 64 bytes)")
offset, limit, dump_size, filename = 0x810000, rootfs_size, 64, "rootfs.squashfs"
image = firmware.read_bytes()
if not dump_size <= len(image) <= limit:
raise RuntimeError(f"Firmware size must be between {dump_size} and {limit:#x} bytes")
if rootfs and len(image) > 0x1800000:
raise RuntimeError("RootFS exceeds the non-overlapping RAM buffer capacity (0x1800000)")
# Invalidate any previous transfer length before attempting a new transfer.
cli.command("setenv filesize")
output = cli.command(
f"tftp {address:#x} {filename}",
r"^Bytes transferred = \d+", timeout=120,
)
transferred = re.findall(r"^Bytes transferred = (\d+)", output, re.MULTILINE)
if transferred != [str(len(image))]:
raise RuntimeError("TFTP byte count differs from the local firmware file")
output = cli.command("echo ${filesize}", r"^[0-9a-fA-F]+$")
sizes = re.findall(r"^([0-9a-fA-F]+)$", output, re.MULTILINE)
if len(sizes) != 1 or int(sizes[0], 16) != len(image):
raise RuntimeError("U-Boot filesize differs from the local firmware file")
output = cli.command(f"md.b {address:#x} {dump_size:#x}", rf"^{address:08x}:")
dumped = bytearray()
for row_offset in range(0, dump_size, 16):
row_address = f"{address + row_offset:08x}"
rows = re.findall(r"^" + row_address + r":((?: [0-9a-fA-F]{2}){16})(?:\s|$)", output, re.MULTILINE)
if len(rows) != 1:
raise RuntimeError("Missing or invalid memory dump at " + row_address)
dumped.extend(bytes.fromhex(rows[0]))
if dumped != image[:dump_size]:
raise RuntimeError(f"First {dump_size} bytes in RAM differ from the local firmware file")
print(f"\nTransfer verified: size and first {dump_size} bytes match.")
cli.command("sf probe 0:0", r"^SF: Detected .+", timeout=20)
cli.command(
f"sf write {address:#x} {offset:#x} ${{filesize}}",
rf"^SF: {len(image)} bytes @ {offset:#x} Written: OK\s*$", timeout=120,
)
cli.command(
f"sf read 0x42800000 {offset:#x} ${{filesize}}",
rf"^SF: {len(image)} bytes @ {offset:#x} Read: OK\s*$", timeout=120,
)
output = cli.command(
f"cmp.b {address:#x} 0x42800000 ${{filesize}}",
r"^Total of \d+ byte\(s\) were the same", timeout=60,
)
compared = re.findall(r"^Total of (\d+) byte\(s\) were the same", output, re.MULTILINE)
if compared != [str(len(image))]:
raise RuntimeError("Flash comparison byte count differs from firmware size")
print("\nFirmware flash write and full read-back comparison verified.")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("mode", choices=("check", "configure", "transfer", "transfer-dtb", "transfer-kernel", "transfer-rootfs"), default="check", nargs="?")
parser.add_argument("--port", default="/dev/ttyUSB0")
parser.add_argument("--firmware", type=Path, default=None)
parser.add_argument("--rootfs-size", type=lambda value: int(value, 0), help="Actual RootFS partition size in bytes (decimal or 0x-prefixed hex)")
args = parser.parse_args()
if args.mode == "transfer-rootfs" and (args.rootfs_size is None or args.rootfs_size < 64):
parser.error("--rootfs-size must specify the actual RootFS partition size (at least 64 bytes)")
if args.firmware is None:
filename = "rootfs.squashfs" if args.mode == "transfer-rootfs" else "zImage" if args.mode == "transfer-kernel" else ("sun8i-v3s-licheepi-zero-dock.dtb" if args.mode == "transfer-dtb" else "u-boot-sunxi-with-spl.bin")
args.firmware = Path(__file__).resolve().parent.parent / "firmware" / filename
if args.mode in ("transfer", "transfer-dtb", "transfer-kernel", "transfer-rootfs") and not args.firmware.is_file():
raise RuntimeError("Firmware file not found: " + str(args.firmware))
# Shared by both workflows: fail rather than interleave serial commands.
with open("/tmp/uboot-serial.lock", "w") as lock:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
with serial.Serial(args.port, 115200, timeout=0.1, write_timeout=2, exclusive=True) as port:
cli = UBoot(port)
cli.detect()
if args.mode == "check":
return
if args.mode in ("transfer", "transfer-dtb", "transfer-kernel", "transfer-rootfs"):
transfer(cli, args.firmware, dtb=args.mode == "transfer-dtb", kernel=args.mode == "transfer-kernel", rootfs=args.mode == "transfer-rootfs", rootfs_size=args.rootfs_size)
return
cli.command("version", r"^U-Boot \d")
cli.command(
"printenv ipaddr serverip ethaddr bootcmd bootargs",
allow_missing=("ipaddr", "serverip", "ethaddr", "bootcmd", "bootargs"),
)
cli.command("setenv ipaddr 192.168.75.100")
cli.command("printenv ipaddr", r"^ipaddr=192\.168\.75\.100$")
cli.command("setenv serverip 192.168.75.247")
cli.command("printenv serverip", r"^serverip=192\.168\.75\.247$")
cli.command("ping 192.168.75.247", r"^host 192\.168\.75\.247 is alive$", timeout=30)
cli.command("sf probe 0:0", r"^SF: Detected .+", timeout=20)
print("\nAll commands verified; environment changes were not saved to flash.")
if __name__ == "__main__":
try:
main()
except (RuntimeError, OSError, serial.SerialException) as exc:
print("\nERROR: " + str(exc), file=sys.stderr)
sys.exit(1)