اضافه شدن قابلیت تنظیم ایپی به ورک فلو
This commit is contained in:
@@ -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)
|
||||
|
||||
مرجع در شماره جدید
مسدود کردن کاربر