✨ 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

config.yaml

@@ -21,5 +21,6 @@ step_timeouts:
   tftp_reboot: 120
 tftp_upload_name: z7bin
 uart_delay: 3
+boot_wait_delay: 10
 xilinx_path: /data/xilinx
 zynq_ip: 192.168.100.11

src/config_manager.py

@@ -33,6 +33,7 @@ DEFAULT_CONFIG: dict[str, Any] = {
     "ping_count": 3,
     "uart_delay": 3,
     "inter_step_delay": 2,
+    "boot_wait_delay": 10,
     "step_timeouts": {
         "check_env": 30,
         "flash_erase": 120,
@@ -70,6 +71,7 @@ class Config:
     ping_count: int = 3
     uart_delay: int = 3
     inter_step_delay: int = 2
+    boot_wait_delay: int = 10
     step_timeouts: dict[str, int] = field(default_factory=lambda: {})
 
     # Internal: path to the config file on disk
@@ -178,6 +180,7 @@ class Config:
             "ping_count": self.ping_count,
             "uart_delay": self.uart_delay,
             "inter_step_delay": self.inter_step_delay,
+            "boot_wait_delay": self.boot_wait_delay,
             "step_timeouts": self.step_timeouts,
         }
         return data

src/gui/main_window.py

@@ -85,6 +85,7 @@ class MainWindow(ctk.CTk):
         self._uart_monitor: UartMonitor | None = None
         self._uart_window: UartMonitorWindow | None = None
         self._flash_phase: str = ""
+        self._tftp_phase: str = ""
 
         # StringVar references for config sync
         self._ip_string_var: ctk.StringVar | None = None
@@ -264,8 +265,8 @@ class MainWindow(ctk.CTk):
 
     def _build_ui(self) -> None:
         """Build the complete UI layout."""
-        # Main container with padding
+        # Scrollable main container — scrollbar appears when content overflows
-        main_frame = ctk.CTkFrame(self)
+        main_frame = ctk.CTkScrollableFrame(self)
         main_frame.pack(fill="both", expand=True, padx=PADDING_LARGE, pady=PADDING_LARGE)
         main_frame.grid_rowconfigure(1, weight=1)
         main_frame.grid_columnconfigure(0, weight=1)
@@ -276,7 +277,7 @@ class MainWindow(ctk.CTk):
         # ── Left Panel: Workflow Steps ──
         left_frame = ctk.CTkFrame(main_frame)
         left_frame.grid(row=1, column=0, sticky="nsew", padx=(0, PADDING))
-        left_frame.grid_rowconfigure(3, weight=1)
+        left_frame.grid_rowconfigure(1, weight=1)
         left_frame.grid_columnconfigure(0, weight=1)
 
         self._build_workflow_panel(left_frame)
@@ -285,10 +286,12 @@ class MainWindow(ctk.CTk):
         # ── Right Panel: Configuration ──
         right_frame = ctk.CTkFrame(main_frame)
         right_frame.grid(row=1, column=1, sticky="nsew", padx=(PADDING, 0))
-        right_frame.grid_rowconfigure(4, weight=1)
+        right_frame.grid_rowconfigure(3, weight=1)
+        right_frame.grid_columnconfigure(0, weight=1)
 
         self._build_config_panel(right_frame)
         self._build_serial_panel(right_frame)
+        self._build_utility_panel(right_frame)
         self._build_status_panel(right_frame)
 
     def _build_header(self, parent: ctk.CTkFrame) -> None:
@@ -601,7 +604,6 @@ class MainWindow(ctk.CTk):
             row=1, column=0, sticky="nsew",
             padx=PADDING, pady=PADDING,
         )
-        panel.grid_rowconfigure(3, weight=1)
         panel.grid_columnconfigure(0, weight=1)
 
         title = ctk.CTkLabel(
@@ -622,6 +624,7 @@ class MainWindow(ctk.CTk):
             port_frame,
             values=[""] + [p.device for p in detect_serial_ports()],
             variable=self._port_var,
+            command=self._on_port_changed,
         )
         self._port_menu.grid(row=0, column=1, sticky="nsew", padx=PADDING_SMALL)
 
@@ -644,20 +647,65 @@ class MainWindow(ctk.CTk):
         )
         uart_btn.grid(row=2, column=0, padx=PADDING, pady=PADDING_SMALL)
 
-        # Test version button
+    def _build_utility_panel(self, parent: ctk.CTkFrame) -> None:
+        """Build the utility tools panel."""
+        panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS)
+        panel.grid(
+            row=2, column=0, sticky="nsew",
+            padx=PADDING, pady=PADDING,
+        )
+        panel.grid_columnconfigure((0, 1), weight=1)
+
+        title = ctk.CTkLabel(
+            panel,
+            text="Utilities",
+            font=FONT_HEADING,
+        )
+        title.grid(row=0, column=0, columnspan=2, padx=PADDING, pady=PADDING)
+
+        # Row 1: Reboot buttons (both red)
+        self._jtag_reboot_btn = ctk.CTkButton(
+            panel,
+            text="Reboot (JTAG)",
+            font=FONT_BODY,
+            fg_color=DANGER_COLOR,
+            hover_color="#B92B2F",
+            command=self._jtag_reboot,
+        )
+        self._jtag_reboot_btn.grid(row=1, column=0, padx=PADDING_SMALL, pady=PADDING_SMALL, sticky="ew")
+
+        self._udp_reboot_btn = ctk.CTkButton(
+            panel,
+            text="Reboot (UDP)",
+            font=FONT_BODY,
+            fg_color=DANGER_COLOR,
+            hover_color="#B92B2F",
+            command=self._udp_reboot,
+        )
+        self._udp_reboot_btn.grid(row=1, column=1, padx=PADDING_SMALL, pady=PADDING_SMALL, sticky="ew")
+
+        # Row 2: Version buttons
+        self._udp_version_btn = ctk.CTkButton(
+            panel,
+            text="Version (UDP)",
+            font=FONT_BODY,
+            command=self._udp_version,
+        )
+        self._udp_version_btn.grid(row=2, column=0, padx=PADDING_SMALL, pady=PADDING_SMALL, sticky="ew")
+
         self._test_version_btn = ctk.CTkButton(
             panel,
-            text="Test Version",
+            text="Version (Serial)",
             font=FONT_BODY,
             command=self._test_serial_version,
         )
-        self._test_version_btn.grid(row=3, column=0, padx=PADDING, pady=PADDING_SMALL)
+        self._test_version_btn.grid(row=2, column=1, padx=PADDING_SMALL, pady=PADDING_SMALL, sticky="ew")
 
     def _build_status_panel(self, parent: ctk.CTkFrame) -> None:
         """Build the status display panel."""
         panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS)
         panel.grid(
-            row=2, column=0, sticky="nsew",
+            row=3, column=0, sticky="nsew",
             padx=PADDING, pady=PADDING,
         )
 
@@ -807,6 +855,49 @@ class MainWindow(ctk.CTk):
         self._uart_warning_label.configure(text=msg)
         self._uart_warning_label.pack(fill="x", padx=PADDING, pady=(0, PADDING_SMALL))
 
+    def _hide_uart_warning(self) -> None:
+        """Hide the persistent UART warning label."""
+        self._uart_warning_label.pack_forget()
+
+    def _on_port_changed(self, new_port: str) -> None:
+        """Handle serial port selection change — validate immediately.
+
+        Args:
+            new_port: The newly selected serial port path.
+        """
+        if not new_port:
+            self._uart_available = False
+            self._show_uart_notification("No port selected")
+            if self._uart_monitor:
+                self._stop_uart_monitor()
+            return
+
+        # Attempt to open port to validate
+        try:
+            port = new_port
+            baudrate = self._config.serial_baudrate if self._config else 115200
+            available, reason = check_uart_available(port, baudrate)
+            self._uart_available = available
+            if available:
+                self._uart_message = reason
+                self._hide_uart_warning()
+                self._log_message(f"  ✓ UART {port}: {reason}")
+                # If UART monitor was running, restart on new port
+                if self._uart_monitor and self._uart_monitor.is_running():
+                    self._stop_uart_monitor()
+                    self._start_uart_monitor()
+            else:
+                self._uart_available = False
+                self._uart_message = reason
+                self._show_uart_notification(reason)
+                if self._uart_monitor:
+                    self._stop_uart_monitor()
+        except Exception as e:
+            self._uart_available = False
+            self._show_uart_notification(str(e))
+            if self._uart_monitor:
+                self._stop_uart_monitor()
+
     def _run_step_2_program_flash(self) -> bool:
         """Step 2: Program and verify Flash using program_flash."""
         if not self._config:
@@ -892,6 +983,11 @@ class MainWindow(ctk.CTk):
             for result in results:
                 status = "✓" if result.success else "✗"
                 self._log_message(f"  {status} {result.step}: {result.message}")
+                if result.output:
+                    # Log tool output for diagnostics (truncated)
+                    output_lines = result.output.strip().splitlines()
+                    for line in output_lines[-20:]:  # last 20 lines
+                        self._log_message(f"    [xsct] {line}")
                 if "DAP" in result.message and not result.success:
                     dap_fail = True
 
@@ -930,6 +1026,7 @@ class MainWindow(ctk.CTk):
 
         # 4a: TFTP Upload
         self._log_message("Step 4a: TFTP upload...")
+        self._tftp_phase = "Upload"
         if hasattr(self, '_tftp_sub_step_frame'):
             self._tftp_sub_step_frame.set_phase("Upload")
         try:
@@ -952,14 +1049,20 @@ class MainWindow(ctk.CTk):
             self._log_message("  Skipping download verify (TFTP upload failed)")
             return False
 
+        # Delay to let Zynq TFTP server stabilize
+        self._log_message("  ⏳ Waiting 2s before download...")
+        time.sleep(2)
+
         # 4b: TFTP Download Verify
         self._log_message("Step 4b: TFTP download verify...")
+        self._tftp_phase = "Download Verify"
         if hasattr(self, '_tftp_sub_step_frame'):
             self._tftp_sub_step_frame.set_phase("Download Verify")
         try:
             download_result = tftp_download(
                 self._config, remote_name,
                 callback=self._tftp_sub_callback,
+                expected_size=bin_path.stat().st_size,
             )
             status = "✓" if download_result.success else "✗"
             self._log_message(f"  {status} {download_result.message}")
@@ -1007,7 +1110,7 @@ class MainWindow(ctk.CTk):
         try:
             reboot_result = reboot_zynq(
                 self._config,
-                method="auto",
+                method="jtag",
                 callback=self._reboot_callback,
             )
             status = "✓" if reboot_result.success else "✗"
@@ -1025,6 +1128,29 @@ class MainWindow(ctk.CTk):
             self._log_message("  Skipping boot verify (reboot failed)")
             return False
 
+        # Wait for board to boot — UART if available, else config delay
+        boot_delay = self._config.boot_wait_delay
+        uart_alive = (self._uart_monitor is not None
+                      and self._uart_monitor.is_running())
+        if uart_alive:
+            self._log_message(f"  ⏳ Waiting for UART boot signal (max {boot_delay}s)...")
+            elapsed = 0.0
+            while elapsed < boot_delay:
+                info = self._uart_monitor.latest_info
+                if info.ip_address or info.version:
+                    self._log_message(
+                        f"  ✓ Boot detected after {elapsed:.1f}s"
+                        f" (IP={info.ip_address}, Ver={info.version})"
+                    )
+                    break
+                time.sleep(0.5)
+                elapsed += 0.5
+            else:
+                self._log_message(f"  ⏱ UART timeout — no boot signal in {boot_delay}s")
+        else:
+            self._log_message(f"  ⏳ UART unavailable — sleeping {boot_delay}s for boot...")
+            time.sleep(boot_delay)
+
         # 4e: Verify boot
         self._log_message("Step 4e: Verifying boot...")
         if hasattr(self, '_tftp_sub_step_frame'):
@@ -1058,7 +1184,15 @@ class MainWindow(ctk.CTk):
         self._log_message(f"  [{status}] {message}")
 
     def _tftp_sub_callback(self, status: str, message: str) -> None:
-        """Callback for TFTP sub-step progress."""
+        """Callback for TFTP sub-step progress — drives TftpSubStepFrame."""
+        if status == "progress":
+            try:
+                pct = int(message.rstrip('%'))
+            except ValueError:
+                pct = -1
+            if hasattr(self, '_tftp_sub_step_frame'):
+                self._tftp_sub_step_frame.set_phase(self._tftp_phase, pct)
+            return
         self._log_message(f"  [{status}] {message}")
 
     def _reboot_callback(self, status: str, message: str) -> None:
@@ -1402,10 +1536,111 @@ class MainWindow(ctk.CTk):
                 self._status.set_status(f"Error: {e}", "error")
             finally:
                 # Re-enable button
-                self.after(0, lambda: self._test_version_btn.configure(state="normal", text="Test Version"))
+                self.after(0, lambda: self._test_version_btn.configure(state="normal", text="Version (Serial)"))
 
         threading.Thread(target=test_version_thread, daemon=True).start()
 
+    def _jtag_reboot(self) -> None:
+        """Reboot the Zynq board via JTAG system reset."""
+        if not self._config:
+            self._log_message("No config loaded — cannot reboot")
+            return
+
+        self._log_message(f"JTAG reboot: sending system reset to {self._config.zynq_ip}...")
+        self._jtag_reboot_btn.configure(state="disabled", text="Rebooting...")
+
+        def _reboot_thread():
+            try:
+                from reboot_manager import reboot_via_jtag
+
+                result = reboot_via_jtag(
+                    self._config,
+                    callback=lambda s, m: self._log_message(f"  [{s}] {m}"),
+                )
+                if result.success:
+                    self._log_message(f"  ✓ JTAG reset: {result.message}")
+                    self._status.set_status("JTAG reset complete", "success")
+                else:
+                    self._log_message(f"  ✗ JTAG reset: {result.message}")
+                    self._status.set_status(f"JTAG reset failed: {result.message}", "error")
+                    if result.output:
+                        self._log_message(f"  Output: {result.output[:500]}")
+            except Exception as e:
+                self._log_message(f"  ✗ JTAG reboot error: {e}")
+                self._status.set_status(f"Error: {e}", "error")
+            finally:
+                self.after(0, lambda: self._jtag_reboot_btn.configure(
+                    state="normal", text="Reboot (JTAG)",
+                ))
+
+        threading.Thread(target=_reboot_thread, daemon=True).start()
+
+    def _udp_reboot(self) -> None:
+        """Reboot the Zynq board via UDP command."""
+        if not self._config:
+            self._log_message("No config loaded — cannot reboot")
+            return
+
+        self._log_message(f"UDP reboot: sending reboot() to {self._config.zynq_ip}...")
+        self._udp_reboot_btn.configure(state="disabled", text="Rebooting...")
+
+        def _thread():
+            try:
+                from reboot_manager import reboot_via_udp
+
+                result = reboot_via_udp(
+                    self._config,
+                    callback=lambda s, m: self._log_message(f"  [{s}] {m}"),
+                )
+                if result.success:
+                    self._log_message(f"  ✓ UDP reboot: {result.message}")
+                    self._status.set_status("UDP reboot complete", "success")
+                else:
+                    self._log_message(f"  ✗ UDP reboot: {result.message}")
+                    self._status.set_status(f"UDP reboot failed", "error")
+            except Exception as e:
+                self._log_message(f"  ✗ UDP reboot error: {e}")
+                self._status.set_status(f"Error: {e}", "error")
+            finally:
+                self.after(0, lambda: self._udp_reboot_btn.configure(
+                    state="normal", text="Reboot (UDP)",
+                ))
+
+        threading.Thread(target=_thread, daemon=True).start()
+
+    def _udp_version(self) -> None:
+        """Query Zynq firmware version via UDP."""
+        if not self._config:
+            self._log_message("No config loaded")
+            return
+
+        self._log_message(f"UDP version: sending ver() to {self._config.zynq_ip}...")
+        self._udp_version_btn.configure(state="disabled", text="Querying...")
+
+        def _thread():
+            try:
+                from reboot_manager import get_version_via_udp
+
+                result = get_version_via_udp(
+                    self._config,
+                    callback=lambda s, m: self._log_message(f"  [{s}] {m}"),
+                )
+                if result.success:
+                    self._log_message(f"  ✓ Version: {result.version}")
+                    self._status.set_status(f"Version: {result.version}", "success")
+                else:
+                    self._log_message(f"  ✗ {result.message}")
+                    self._status.set_status(f"Query failed", "error")
+            except Exception as e:
+                self._log_message(f"  ✗ UDP version error: {e}")
+                self._status.set_status(f"Error: {e}", "error")
+            finally:
+                self.after(0, lambda: self._udp_version_btn.configure(
+                    state="normal", text="Version (UDP)",
+                ))
+
+        threading.Thread(target=_thread, daemon=True).start()
+
     # ── Vitis Check ────────────────────────────────────────────
 
     def _check_vitis(self) -> None:

src/gui/widgets.py

@@ -619,6 +619,7 @@ class LogDisplay(ctk.CTkFrame):
         """
         self._text_widget.configure(state="normal")
         self._text_widget.insert("end", message + "\n")
+        self._text_widget.update_idletasks()
         self._text_widget.see("end")
         self._text_widget.configure(state="disabled")
 
@@ -696,7 +697,7 @@ class SubStepFrame(ctk.CTkFrame):
                 item["pct"].grid()
 
     def set_expanded(self, expanded: bool) -> None:
-        if expanded != self._collapsed:
+        if expanded == self._collapsed:
             self._toggle()
 
     def set_phase(self, phase: str, pct: int = -1) -> None:
@@ -810,9 +811,11 @@ class TftpSubStepFrame(ctk.CTkFrame):
         for i, name in enumerate(self._substep_names):
             dot = ctk.CTkLabel(self, text="○", font=FONT_BODY, text_color=ACCENT_PENDING, anchor="w")
             label = ctk.CTkLabel(self, text=name, font=FONT_BODY, text_color=ACCENT_PENDING, anchor="w")
+            pct = ctk.CTkLabel(self, text="", font=FONT_BODY, text_color=ACCENT_PENDING, anchor="e")
             dot.grid(row=i + 1, column=0, padx=(16, 2), pady=(2, 2), sticky="w")
             label.grid(row=i + 1, column=1, padx=(2, 4), pady=(2, 2), sticky="w")
-            self._sub_items.append({"dot": dot, "label": label})
+            pct.grid(row=i + 1, column=2, padx=(4, 8), pady=(2, 2), sticky="e")
+            self._sub_items.append({"dot": dot, "label": label, "pct": pct})
 
         self.grid_columnconfigure(1, weight=1)
 
@@ -823,12 +826,14 @@ class TftpSubStepFrame(ctk.CTkFrame):
             if self._collapsed:
                 item["dot"].grid_remove()
                 item["label"].grid_remove()
+                item["pct"].grid_remove()
             else:
                 item["dot"].grid()
                 item["label"].grid()
+                item["pct"].grid()
 
     def set_expanded(self, expanded: bool) -> None:
-        if expanded != self._collapsed:
+        if expanded == self._collapsed:
             self._toggle()
 
     def set_phase(self, phase: str, pct: int = -1) -> None:
@@ -840,16 +845,20 @@ class TftpSubStepFrame(ctk.CTkFrame):
         if idx < 0:
             return
 
+        pct_text = f"{pct}%" if pct >= 0 else ""
         for i, item in enumerate(self._sub_items):
             if i < idx:
                 item["dot"].configure(text="●", text_color=_C_DONE)
                 item["label"].configure(text_color=_C_DONE)
+                item["pct"].configure(text="OK", text_color=_C_DONE)
             elif i == idx:
                 item["dot"].configure(text="◉", text_color=_C_RUNNING)
                 item["label"].configure(text_color=_C_RUNNING)
+                item["pct"].configure(text=pct_text, text_color=_C_RUNNING)
             else:
                 item["dot"].configure(text="○", text_color=_C_PENDING)
                 item["label"].configure(text_color=_C_PENDING)
+                item["pct"].configure(text="", text_color=_C_PENDING)
 
         if phase in self._substep_names and not self._pulse_job:
             self._start_pulse(idx)
@@ -863,6 +872,7 @@ class TftpSubStepFrame(ctk.CTkFrame):
         item = self._sub_items[idx]
         item["dot"].configure(text="●", text_color=_C_DONE)
         item["label"].configure(text_color=_C_DONE)
+        item["pct"].configure(text="OK", text_color=_C_DONE)
         self._stop_pulse()
 
     def set_step_error(self, phase: str) -> None:
@@ -904,6 +914,7 @@ class TftpSubStepFrame(ctk.CTkFrame):
         for i, item in enumerate(self._sub_items):
             item["dot"].configure(text="○", text_color=_C)
             item["label"].configure(text_color=_C)
+            item["pct"].configure(text="", text_color=_C)
         if not self._collapsed:
             self._toggle()
             self._toggle()  # collapse then expand to reset

src/reboot_manager.py

@@ -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 os
-import platform
+import socket
-import shutil
 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.
+        ip: Zynq IP address string.
-        timeout: Seconds to wait for reboot acknowledgment.
-        callback: Optional progress callback.
 
     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(
+        octets = ip.strip().split(".")
-            cmd,
+        last = int(octets[3])
-            capture_output=True,
+        return last - 3
-            text=True,
+    except (ValueError, IndexError):
-            timeout=10,
+        return -1
-        )
-        if callback:
-            if result.returncode == 0:
-                callback("progress", f"Reboot command sent to {ip}")
-            else:
-                callback("error", f"SSH reboot failed: {result.stderr.strip()}")
 
-        # Wait for device to come back
-        if result.returncode == 0 and callback:
-            callback("progress", f"Waiting {timeout}s for device to reboot...")
 
-        return RebootResult(
+def _send_udp_command(
-            success=result.returncode == 0,
+    ip: str,
-            message="Reboot command sent" if result.returncode == 0 else f"SSH reboot failed: {result.stderr.strip()}",
+    command: bytes,
-            method="ssh",
+    timeout: float = 5.0,
-            output=result.stdout,
+    buffer_size: int = 4096,
-        )
+) -> tuple[bool, str]:
-    except subprocess.TimeoutExpired:
+    """Send a UDP command to the Zynq and receive the response.
-        return RebootResult(
+
-            success=False,
+    Args:
-            message="SSH reboot timed out",
+        ip: Zynq IP address.
-            method="ssh",
+        command: Command bytes to send (e.g., b'reboot()').
-        )
+        timeout: Receive timeout in seconds.
-    except FileNotFoundError:
+        buffer_size: Max receive buffer size.
-        return RebootResult(
+
-            success=False,
+    Returns:
-            message="ssh command not found",
+        (success, response_text).
-            method="ssh",
+    """
-        )
+    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(
+        return False, f"UDP error: {e}"
-            success=False,
+    finally:
-            message=f"OS error: {e}",
+        sock.close()
-            method="ssh",
+
-        )
+
+# ── 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
+            # Send reboot command as function call
-            ser.write(b"reboot\r\n")
+            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'.
+        method: Reboot method: 'auto', 'jtag', 'udp', or 'serial'.
-        timeout: Seconds to wait for reboot.
+        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":
+    if method == "jtag":
-        return reboot_via_ssh(config, timeout, callback)
+        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
+    # Auto: try JTAG first, then UDP, then serial
-    result = reboot_via_ssh(config, timeout, callback)
+    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)

src/tftp_manager.py

@@ -9,7 +9,6 @@ from __future__ import annotations
 import binascii
 import socket
 import struct
-import tempfile
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable
@@ -55,7 +54,8 @@ def _compute_crc32(file_path: str | Path) -> int:
 
 
 def _tftp_put(host: str, file_path: str, remote_name: str,
-              timeout: float = 10.0) -> tuple[bool, str]:
+              timeout: float = 10.0,
+              callback: Callable[[str, str], None] | None = None) -> tuple[bool, str]:
     """Upload a file to a TFTP server using pure Python.
 
     Args:
@@ -63,6 +63,7 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
         file_path: Local file path.
         remote_name: Remote filename.
         timeout: Socket timeout in seconds.
+        callback: Optional progress callback (status, message).
 
     Returns:
         (success, message).
@@ -112,6 +113,8 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
                 pct = offset * 100 // total
                 print(f"[TFTP] Block {block}/{-(-total // block_size)} ({pct}%) sent to {srv_ip}:{srv_port}",
                       file=sys.stderr)
+                if callback:
+                    callback("progress", f"{pct}%")
 
             # Wait for ACK
             try:
@@ -130,7 +133,9 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
 
 
 def _tftp_get(host: str, remote_name: str, local_dir: str | Path,
-              timeout: float = 30.0) -> tuple[bool, str, Path]:
+              timeout: float = 30.0,
+              callback: Callable[[str, str], None] | None = None,
+              expected_size: int = 0) -> tuple[bool, str, Path]:
     """Download a file from a TFTP server using pure Python.
 
     Args:
@@ -138,6 +143,8 @@ def _tftp_get(host: str, remote_name: str, local_dir: str | Path,
         remote_name: Remote filename.
         local_dir: Local directory to save the downloaded file.
         timeout: Socket timeout in seconds.
+        callback: Optional progress callback (status, message).
+        expected_size: Expected file size in bytes (for progress percentage).
 
     Returns:
         (success, message, local_path).
@@ -170,46 +177,56 @@ def _tftp_get(host: str, remote_name: str, local_dir: str | Path,
 
         srv_ip, srv_port = server_addr
         block = struct.unpack("!H", data_pkt[2:4])[0]
-        file_data = data_pkt[4:]
+        chunk = data_pkt[4:]
         import sys
-        print(f"[TFTP] DATA block 1 received from {srv_ip}:{srv_port} "
+        print(f"[TFTP] DATA block {block} received from {srv_ip}:{srv_port} "
-              f"({len(file_data)} bytes)", file=sys.stderr)
+              f"({len(chunk)} bytes)", file=sys.stderr)
 
-        # ── ACK block 0 and receive subsequent blocks ──
+        # ── ACK this block with the actual block number ──
-        ack = struct.pack("!HH", _TFTP_ACK, 0)
+        file_data = bytearray(chunk)
+        ack = struct.pack("!HH", _TFTP_ACK, block)
         sock.sendto(ack, server_addr)
 
-        total = len(file_data)
+        if callback:
-        while len(file_data) == _TFTP_BLOCK_SIZE:
+            pct = len(file_data) * 100 // expected_size if expected_size > 0 else -1
-            # Wait for next DATA block
+            callback("progress", f"{pct}%")
+
+        # ── Receive subsequent blocks while last chunk was full-size ──
+        while len(chunk) == _TFTP_BLOCK_SIZE:
             try:
                 data_pkt, _ = sock.recvfrom(516)
                 if len(data_pkt) < 4:
                     return False, "Incomplete DATA packet", local_path
                 opcode = struct.unpack("!H", data_pkt[:2])[0]
+                if opcode == _TFTP_ERROR:
+                    err_msg = data_pkt[4:].decode("ascii", errors="replace").rstrip("\x00")
+                    return False, f"TFTP error: {err_msg}", local_path
                 if opcode != _TFTP_DATA:
                     return False, f"Expected DATA, got opcode {opcode}", local_path
 
                 new_block = struct.unpack("!H", data_pkt[2:4])[0]
                 chunk = data_pkt[4:]
                 file_data += chunk
-                total = len(file_data)
                 block = new_block
 
                 if block % 100 == 0:
-                    pct = total * 100 // max(total, 1)
+                    print(f"[TFTP] Block {block} ({len(file_data)} bytes) "
-                    print(f"[TFTP] Block {block}/{block} ({pct}%) received from {srv_ip}:{srv_port}",
+                          f"received from {srv_ip}:{srv_port}", file=sys.stderr)
-                          file=sys.stderr)
 
                 # ACK this block
                 ack = struct.pack("!HH", _TFTP_ACK, block)
                 sock.sendto(ack, server_addr)
+
+                if callback:
+                    pct = len(file_data) * 100 // expected_size if expected_size > 0 else -1
+                    callback("progress", f"{pct}%")
             except socket.timeout:
-                return False, f"Timeout waiting for DATA block {block}", local_path
+                return False, f"Timeout waiting for DATA block {block + 1}", local_path
 
         # ── Write file ──
-        local_path.write_bytes(file_data)
+        total = len(file_data)
-        return True, f"Downloaded {total} bytes to {local_path}", local_path
+        local_path.write_bytes(bytes(file_data))
+        return True, f"Downloaded {total} bytes in {block} blocks to {local_path}", local_path
     finally:
         sock.close()
 
@@ -254,7 +271,8 @@ def tftp_upload(
         )
 
     try:
-        ok, msg = _tftp_put(config.zynq_ip, str(file_path), remote_name)
+        ok, msg = _tftp_put(config.zynq_ip, str(file_path), remote_name,
+                            callback=callback)
     except OSError as e:
         return TftpResult(
             step="upload", success=False,
@@ -279,20 +297,23 @@ def tftp_download(
     remote_name: str,
     local_dir: str | Path | None = None,
     callback: ProgressCallback = None,
+    expected_size: int = 0,
 ) -> TftpResult:
     """Download a file from the Zynq device via TFTP (pure Python).
 
     Args:
         config: Application configuration.
         remote_name: Remote filename on the TFTP server.
-        local_dir: Local directory to save the file (default: temp directory).
+        local_dir: Local directory to save the file (default: project .tmp/tftp/).
         callback: Optional progress callback.
+        expected_size: Expected file size in bytes (for progress percentage).
 
     Returns:
         TftpResult with download status and CRC.
     """
     if local_dir is None:
-        local_dir = tempfile.gettempdir()
+        # Use project-local temp dir for cross-platform compatibility
+        local_dir = Path(__file__).resolve().parent.parent / ".tmp" / "tftp"
     local_dir = Path(local_dir)
 
     if callback:
@@ -307,7 +328,8 @@ def tftp_download(
         )
 
     try:
-        ok, msg, local_path = _tftp_get(config.zynq_ip, remote_name, local_dir)
+        ok, msg, local_path = _tftp_get(config.zynq_ip, remote_name, local_dir,
+                                        callback=callback, expected_size=expected_size)
     except OSError as e:
         return TftpResult(
             step="download", success=False,