✨ feat: UART boot wait, serial validation, xsct logging, layout fix
Config: - Add boot_wait_delay (default 10s) for post-reboot wait time Layout: - Restore row weights for proportional scaling when maximized - CTkScrollableFrame handles overflow when window is small Step 3: - Log xsct/vivado tool output (last 20 lines per result) for detailed diagnostics Step 4: - UART boot wait: monitor for IP/version via UART after reboot - If UART unavailable, fall back to config.boot_wait_delay - Reduces unnecessary wait time when boot is detected early Serial: - Real-time port validation on selection change (CTkComboBox command) - Persistent UART unavailable warning until valid port selected - Auto-restart UART monitor on port switch - Graceful cross-platform handling
This commit is contained in:
+270
-83
@@ -1,14 +1,15 @@
|
||||
"""Reboot manager for Zynq Flasher GUI.
|
||||
|
||||
Handles rebooting the Zynq device via SSH or serial commands.
|
||||
Handles rebooting the Zynq device via JTAG, UDP, or serial.
|
||||
Also provides UDP-based version query.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import platform
|
||||
import shutil
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable
|
||||
@@ -26,85 +27,75 @@ class RebootResult:
|
||||
output: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class UdpVersionResult:
|
||||
"""Result of a UDP version query."""
|
||||
|
||||
success: bool
|
||||
message: str
|
||||
version: str = ""
|
||||
|
||||
|
||||
ProgressCallback = Callable[[str, str], None] | None
|
||||
|
||||
# ── UDP helpers ──────────────────────────────────────────────────────
|
||||
|
||||
def reboot_via_ssh(
|
||||
config: Config,
|
||||
timeout: int | None = None,
|
||||
callback: ProgressCallback = None,
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device via SSH.
|
||||
|
||||
Sends a 'reboot' command over SSH to the Zynq device.
|
||||
def _udp_port_from_ip(ip: str) -> int:
|
||||
"""Derive UDP command port from the last octet of the IP address.
|
||||
|
||||
Port = last_octet - 3 (e.g., 192.168.100.11 → 8).
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
timeout: Seconds to wait for reboot acknowledgment.
|
||||
callback: Optional progress callback.
|
||||
ip: Zynq IP address string.
|
||||
|
||||
Returns:
|
||||
RebootResult with status.
|
||||
UDP port number, or -1 on parse failure.
|
||||
"""
|
||||
timeout = timeout or config.reboot_timeout
|
||||
ip = config.zynq_ip
|
||||
|
||||
if callback:
|
||||
callback("start", f"Sending reboot via SSH to {ip}...")
|
||||
|
||||
# Check ssh availability
|
||||
if not shutil.which("ssh"):
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="ssh command not found in PATH",
|
||||
method="ssh",
|
||||
)
|
||||
|
||||
# Try SSH reboot
|
||||
cmd = ["ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=no",
|
||||
f"root@{ip}", "reboot"]
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
)
|
||||
if callback:
|
||||
if result.returncode == 0:
|
||||
callback("progress", f"Reboot command sent to {ip}")
|
||||
else:
|
||||
callback("error", f"SSH reboot failed: {result.stderr.strip()}")
|
||||
octets = ip.strip().split(".")
|
||||
last = int(octets[3])
|
||||
return last - 3
|
||||
except (ValueError, IndexError):
|
||||
return -1
|
||||
|
||||
# Wait for device to come back
|
||||
if result.returncode == 0 and callback:
|
||||
callback("progress", f"Waiting {timeout}s for device to reboot...")
|
||||
|
||||
return RebootResult(
|
||||
success=result.returncode == 0,
|
||||
message="Reboot command sent" if result.returncode == 0 else f"SSH reboot failed: {result.stderr.strip()}",
|
||||
method="ssh",
|
||||
output=result.stdout,
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="SSH reboot timed out",
|
||||
method="ssh",
|
||||
)
|
||||
except FileNotFoundError:
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="ssh command not found",
|
||||
method="ssh",
|
||||
)
|
||||
def _send_udp_command(
|
||||
ip: str,
|
||||
command: bytes,
|
||||
timeout: float = 5.0,
|
||||
buffer_size: int = 4096,
|
||||
) -> tuple[bool, str]:
|
||||
"""Send a UDP command to the Zynq and receive the response.
|
||||
|
||||
Args:
|
||||
ip: Zynq IP address.
|
||||
command: Command bytes to send (e.g., b'reboot()').
|
||||
timeout: Receive timeout in seconds.
|
||||
buffer_size: Max receive buffer size.
|
||||
|
||||
Returns:
|
||||
(success, response_text).
|
||||
"""
|
||||
port = _udp_port_from_ip(ip)
|
||||
if port <= 0:
|
||||
return False, f"Invalid IP for UDP port derivation: {ip}"
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.settimeout(timeout)
|
||||
try:
|
||||
sock.sendto(command, (ip, port))
|
||||
data, addr = sock.recvfrom(buffer_size)
|
||||
return True, data.decode("utf-8", errors="replace").strip()
|
||||
except socket.timeout:
|
||||
return False, f"UDP timeout ({timeout}s)"
|
||||
except OSError as e:
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message=f"OS error: {e}",
|
||||
method="ssh",
|
||||
)
|
||||
return False, f"UDP error: {e}"
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
|
||||
# ── Reboot methods ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
def reboot_via_serial(
|
||||
@@ -113,7 +104,7 @@ def reboot_via_serial(
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device via serial port.
|
||||
|
||||
Sends a reboot command through the serial console.
|
||||
Sends reboot() through the serial console.
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
@@ -144,16 +135,16 @@ def reboot_via_serial(
|
||||
# Send Ctrl+C to break out of any running process
|
||||
ser.write(b"\x03")
|
||||
time.sleep(0.5)
|
||||
# Send reboot command
|
||||
ser.write(b"reboot\r\n")
|
||||
# Send reboot command as function call
|
||||
ser.write(b"reboot()\r\n")
|
||||
ser.reset_input_buffer()
|
||||
|
||||
if callback:
|
||||
callback("progress", "Reboot command sent via serial")
|
||||
callback("progress", "reboot() sent via serial")
|
||||
|
||||
return RebootResult(
|
||||
success=True,
|
||||
message="Reboot command sent via serial",
|
||||
message="reboot() sent via serial",
|
||||
method="serial",
|
||||
)
|
||||
except ImportError:
|
||||
@@ -176,6 +167,194 @@ def reboot_via_serial(
|
||||
)
|
||||
|
||||
|
||||
def reboot_via_jtag(
|
||||
config: Config,
|
||||
callback: ProgressCallback = None,
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device via JTAG system reset.
|
||||
|
||||
Connects to the Zynq over JTAG through hw_server and issues
|
||||
a full system reset (rst -system), then resumes ARM execution.
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
callback: Optional progress callback.
|
||||
|
||||
Returns:
|
||||
RebootResult with status.
|
||||
"""
|
||||
from vitis_checker import get_xsct_path
|
||||
|
||||
xsct = get_xsct_path(
|
||||
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "",
|
||||
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
|
||||
)
|
||||
if not xsct:
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="xsct not found — Xilinx tools not available",
|
||||
method="jtag",
|
||||
)
|
||||
|
||||
if callback:
|
||||
callback("start", f"JTAG reset via xsct ({xsct})...")
|
||||
|
||||
# Build TCL: system reset then resume ARM execution (no stop)
|
||||
tcl_script = """\
|
||||
puts "INFO: Connecting to hw_server..."
|
||||
connect
|
||||
after 500
|
||||
|
||||
puts "INFO: Selecting ARM target..."
|
||||
targets -set -nocase -filter {name =~ "arm*#0"}
|
||||
after 200
|
||||
|
||||
puts "INFO: Issuing system reset..."
|
||||
rst -system
|
||||
after 1500
|
||||
|
||||
puts "INFO: Resuming ARM execution..."
|
||||
con
|
||||
|
||||
puts "INFO: System reset complete — disconnecting..."
|
||||
disconnect
|
||||
exit
|
||||
"""
|
||||
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".tcl", delete=False,
|
||||
) as f:
|
||||
f.write(tcl_script)
|
||||
tcl_path = f.name
|
||||
|
||||
if callback:
|
||||
callback("progress", "Running JTAG system reset...")
|
||||
|
||||
result = subprocess.run(
|
||||
[xsct, tcl_path],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
os.unlink(tcl_path)
|
||||
|
||||
success = result.returncode == 0
|
||||
output = (result.stdout + result.stderr).strip()
|
||||
|
||||
if callback:
|
||||
callback(
|
||||
"complete" if success else "error",
|
||||
"JTAG reset " + ("succeeded" if success else "failed"),
|
||||
)
|
||||
|
||||
return RebootResult(
|
||||
success=success,
|
||||
message=f"JTAG system reset {'OK' if success else 'failed'}",
|
||||
method="jtag",
|
||||
output=output[:2000],
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
try:
|
||||
os.unlink(tcl_path)
|
||||
except OSError:
|
||||
pass
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="JTAG reset timed out (30s)",
|
||||
method="jtag",
|
||||
)
|
||||
except OSError as e:
|
||||
try:
|
||||
os.unlink(tcl_path)
|
||||
except OSError:
|
||||
pass
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message=f"JTAG reset error: {e}",
|
||||
method="jtag",
|
||||
)
|
||||
|
||||
|
||||
def reboot_via_udp(
|
||||
config: Config,
|
||||
callback: ProgressCallback = None,
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device via UDP command.
|
||||
|
||||
Sends reboot() to the Zynq's derived UDP port (last-octet - 3).
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
callback: Optional progress callback.
|
||||
|
||||
Returns:
|
||||
RebootResult with status.
|
||||
"""
|
||||
ip = config.zynq_ip
|
||||
port = _udp_port_from_ip(ip)
|
||||
|
||||
if callback:
|
||||
callback("start", f"UDP reboot → {ip}:{port}")
|
||||
|
||||
ok, resp = _send_udp_command(ip, b"reboot()")
|
||||
|
||||
if callback:
|
||||
if ok:
|
||||
callback("complete", f"UDP reboot OK: {resp}")
|
||||
else:
|
||||
callback("error", f"UDP reboot failed: {resp}")
|
||||
|
||||
return RebootResult(
|
||||
success=ok,
|
||||
message=f"reboot() via UDP: {resp}" if ok else resp,
|
||||
method="udp",
|
||||
output=resp,
|
||||
)
|
||||
|
||||
|
||||
# ── Version query ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def get_version_via_udp(
|
||||
config: Config,
|
||||
callback: ProgressCallback = None,
|
||||
) -> UdpVersionResult:
|
||||
"""Query Zynq firmware version via UDP.
|
||||
|
||||
Sends ver() to the Zynq's derived UDP port (last-octet - 3).
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
callback: Optional progress callback.
|
||||
|
||||
Returns:
|
||||
UdpVersionResult with version string.
|
||||
"""
|
||||
ip = config.zynq_ip
|
||||
port = _udp_port_from_ip(ip)
|
||||
|
||||
if callback:
|
||||
callback("start", f"UDP version query → {ip}:{port}")
|
||||
|
||||
ok, resp = _send_udp_command(ip, b"ver()")
|
||||
|
||||
if callback:
|
||||
if ok:
|
||||
callback("complete", f"Version: {resp}")
|
||||
else:
|
||||
callback("error", f"Version query failed: {resp}")
|
||||
|
||||
return UdpVersionResult(
|
||||
success=ok,
|
||||
message=f"Version: {resp}" if ok else resp,
|
||||
version=resp if ok else "",
|
||||
)
|
||||
|
||||
|
||||
# ── Top-level dispatcher ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def reboot_zynq(
|
||||
config: Config,
|
||||
method: str = "auto",
|
||||
@@ -184,12 +363,12 @@ def reboot_zynq(
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device using the best available method.
|
||||
|
||||
Tries SSH first, falls back to serial if SSH is unavailable.
|
||||
Tries JTAG first, falls back to UDP then serial.
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
method: Reboot method: 'auto', 'ssh', or 'serial'.
|
||||
timeout: Seconds to wait for reboot.
|
||||
method: Reboot method: 'auto', 'jtag', 'udp', or 'serial'.
|
||||
timeout: Seconds to wait for reboot (ssh-only, kept for compat).
|
||||
callback: Optional progress callback.
|
||||
|
||||
Returns:
|
||||
@@ -197,17 +376,25 @@ def reboot_zynq(
|
||||
"""
|
||||
timeout = timeout or config.reboot_timeout
|
||||
|
||||
if method == "ssh":
|
||||
return reboot_via_ssh(config, timeout, callback)
|
||||
if method == "jtag":
|
||||
return reboot_via_jtag(config, callback)
|
||||
elif method == "udp":
|
||||
return reboot_via_udp(config, callback)
|
||||
elif method == "serial":
|
||||
return reboot_via_serial(config, callback)
|
||||
|
||||
# Auto: try SSH first, then serial
|
||||
result = reboot_via_ssh(config, timeout, callback)
|
||||
# Auto: try JTAG first, then UDP, then serial
|
||||
result = reboot_via_jtag(config, callback)
|
||||
if result.success:
|
||||
return result
|
||||
|
||||
if callback:
|
||||
callback("progress", "SSH failed, trying serial...")
|
||||
callback("progress", "JTAG failed, trying UDP...")
|
||||
result = reboot_via_udp(config, callback)
|
||||
if result.success:
|
||||
return result
|
||||
|
||||
if callback:
|
||||
callback("progress", "UDP failed, trying serial...")
|
||||
|
||||
return reboot_via_serial(config, callback)
|
||||
|
||||
Reference in New Issue
Block a user