diff --git a/config.yaml b/config.yaml index d3b74ff..3031c8b 100644 --- a/config.yaml +++ b/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 diff --git a/src/config_manager.py b/src/config_manager.py index fd2c5c9..018e87a 100644 --- a/src/config_manager.py +++ b/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 diff --git a/src/gui/main_window.py b/src/gui/main_window.py index e31f21d..ba3c3db 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -268,6 +268,7 @@ class MainWindow(ctk.CTk): # Scrollable main container — scrollbar appears when content overflows 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) # ── Header ── @@ -285,6 +286,7 @@ 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(3, weight=1) right_frame.grid_columnconfigure(0, weight=1) self._build_config_panel(right_frame) @@ -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) @@ -852,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: @@ -937,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 @@ -1077,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'): diff --git a/src/gui/widgets.py b/src/gui/widgets.py index 6bbdacc..456d060 100644 --- a/src/gui/widgets.py +++ b/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 diff --git a/src/reboot_manager.py b/src/reboot_manager.py index 76d8eaf..d9e8676 100644 --- a/src/reboot_manager.py +++ b/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 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) diff --git a/src/tftp_manager.py b/src/tftp_manager.py index 2a03f39..991ba36 100644 --- a/src/tftp_manager.py +++ b/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} " - f"({len(file_data)} bytes)", file=sys.stderr) + print(f"[TFTP] DATA block {block} received from {srv_ip}:{srv_port} " + f"({len(chunk)} bytes)", file=sys.stderr) - # ── ACK block 0 and receive subsequent blocks ── - ack = struct.pack("!HH", _TFTP_ACK, 0) + # ── ACK this block with the actual block number ── + file_data = bytearray(chunk) + ack = struct.pack("!HH", _TFTP_ACK, block) sock.sendto(ack, server_addr) - total = len(file_data) - while len(file_data) == _TFTP_BLOCK_SIZE: - # Wait for next DATA block + if callback: + pct = len(file_data) * 100 // expected_size if expected_size > 0 else -1 + 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}/{block} ({pct}%) received from {srv_ip}:{srv_port}", - file=sys.stderr) + print(f"[TFTP] Block {block} ({len(file_data)} bytes) " + f"received from {srv_ip}:{srv_port}", 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) - return True, f"Downloaded {total} bytes to {local_path}", local_path + total = len(file_data) + 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,