diff --git a/.gitea/workflows/serial.yml b/.gitea/workflows/serial.yml index d2b6bda..cf03a61 100644 --- a/.gitea/workflows/serial.yml +++ b/.gitea/workflows/serial.yml @@ -1,16 +1,16 @@ -name: U-Boot Serial Command +name: Check U-Boot CLI on: workflow_dispatch: jobs: - uboot-command: + uboot-check: runs-on: serial-runner steps: - name: Checkout Repository uses: actions/checkout@v4 - - name: Execute U-Boot bdinfo + - name: Check current U-Boot CLI prompt run: | - /usr/bin/python3 scripts/serial_command.py + /usr/bin/python3 scripts/serial_command.py check diff --git a/.gitea/workflows/uboot-setup.yml b/.gitea/workflows/uboot-setup.yml new file mode 100644 index 0000000..88a9bfb --- /dev/null +++ b/.gitea/workflows/uboot-setup.yml @@ -0,0 +1,16 @@ +name: Configure and Verify U-Boot + +on: + workflow_dispatch: + +jobs: + uboot-setup: + runs-on: serial-runner + + steps: + - name: Checkout Repository + uses: actions/checkout@v4 + + - name: Configure network and verify U-Boot commands + run: | + /usr/bin/python3 scripts/serial_command.py configure diff --git a/scripts/serial_command.py b/scripts/serial_command.py index 003f8eb..46cc290 100644 --- a/scripts/serial_command.py +++ b/scripts/serial_command.py @@ -1,67 +1,107 @@ -import serial +"""Check an existing U-Boot prompt or configure its volatile environment.""" +import argparse +import fcntl +import re import sys import time -PORT = "/dev/ttyUSB0" -BAUDRATE = 115200 -TIMEOUT = 2 -MAX_WAIT = 10 +import serial -COMMAND = "bdinfo" +# 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 +) -try: - with serial.Serial( - port=PORT, - baudrate=BAUDRATE, - timeout=0.2, - write_timeout=TIMEOUT - ) as ser: - print(f"Connected to {PORT}", flush=True) +class UBoot: + def __init__(self, port): + self.port = port + self.prompt = None - # Clear old serial data - ser.reset_input_buffer() + 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") - # Send command to U-Boot - ser.write((COMMAND + "\r").encode()) - ser.flush() + 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) - print(f"Command sent: {COMMAND}", flush=True) - print("------ U-Boot Response ------", flush=True) + def command(self, command, expected=None, timeout=10): + 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) + if codes != ["0"]: + raise RuntimeError("Command failed or its exit status could not be verified: " + command) + if ERROR.search(output): + raise RuntimeError("Error in command response: " + command) + if expected and not re.search(expected, output, re.MULTILINE): + raise RuntimeError("Expected response missing: " + command) + print("\nVerified: " + command, flush=True) + return output - response = bytearray() - start_time = time.monotonic() - last_data_time = start_time - while time.monotonic() - start_time < MAX_WAIT: +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("mode", choices=("check", "configure"), default="check", nargs="?") + parser.add_argument("--port", default="/dev/ttyUSB0") + args = parser.parse_args() + # 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 + cli.command("version", r"^U-Boot \d") + cli.command("printenv 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 ser.in_waiting: - data = ser.read(ser.in_waiting) - response.extend(data) - # Print response in real time - print( - data.decode("utf-8", errors="replace"), - end="", - flush=True - ) - - last_data_time = time.monotonic() - - # Stop when no new data arrives for 2 seconds - if response and time.monotonic() - last_data_time > TIMEOUT: - break - - time.sleep(0.05) - - print("\n-----------------------------") - - if not response: - print("ERROR: No response from U-Boot") - sys.exit(1) - - print("Command completed successfully") - -except serial.SerialException as e: - print(f"Serial error: {e}") - sys.exit(1) +if __name__ == "__main__": + try: + main() + except (RuntimeError, OSError, serial.SerialException) as exc: + print("\nERROR: " + str(exc), file=sys.stderr) + sys.exit(1)