feat: add TFTP download verify, Step 4 sub-steps, and fix UI layout

- Implement _tftp_get() for downloading files from Zynq via TFTP
- Add TftpSubStepFrame widget for Step 4 progress display
- Refactor Step 4 into 5 sub-steps: Upload, Download Verify, CRC Check, Reboot, Boot Verify
- Fix SubStepFrame.reset() bug that caused collapse after execution
- Fix layout overlap: Step 4 sub-step and progress bar now use separate rows
- Improve ProgressIndicator appearance with better styling and spacing
- Fix UART monitor to start regardless of _uart_available flag
- Add CRC32 verification between uploaded and downloaded files
This commit is contained in:
2026-06-11 11:47:35 +08:00
parent 5ce544cd3c
commit f04fbe2130
3 changed files with 417 additions and 64 deletions
+124 -35
View File
@@ -21,7 +21,7 @@ from zynq_checker import check_zynq_jtag, get_zynq_status_dict
from flash_programmer import full_flash_program, FlashResult
from bitstream_programmer import full_bitstream_program, BitstreamResult
from ip_verifier import ping_ip
from tftp_manager import tftp_upload_verify, TftpResult
from tftp_manager import tftp_upload, tftp_download, tftp_upload_verify, TftpResult, _compute_crc32
from reboot_manager import reboot_zynq, RebootResult
from boot_verifier import verify_boot, verify_zynq_status, BootVerificationResult
from serial_monitor import detect_serial_ports, parse_boot_output, check_uart_available, UartMonitor
@@ -53,6 +53,7 @@ from gui.widgets import (
FileSelector,
LogDisplay,
SubStepFrame,
TftpSubStepFrame,
UartMonitorWindow,
)
@@ -331,7 +332,6 @@ class MainWindow(ctk.CTk):
row=0, column=0, sticky="nsew",
padx=PADDING, pady=(PADDING, 0),
)
panel.grid_rowconfigure(10, weight=1)
panel.grid_columnconfigure(0, weight=1)
# ── Title ──
@@ -364,23 +364,29 @@ class MainWindow(ctk.CTk):
self._steps.append(step)
self._step_statuses[i] = "pending"
# Divider line between steps
# Divider line between steps (skip rows occupied by sub-steps)
if i > 0:
# Dividers go between main steps only, not between step and sub-step
# Row calculation: account for sub-steps from previous steps
prev_steps_with_substeps = sum(1 for j in range(i) if j in (1, 3))
divider_row = i * 2 + 1 + prev_steps_with_substeps
divider = ctk.CTkFrame(
panel,
height=1,
fg_color=CONNECTOR_COLOR,
)
divider.grid(
row=i * 2, column=0,
row=divider_row, column=0,
sticky="ew",
padx=PADDING,
pady=(2, 0),
)
# Position step
# Position step (account for sub-steps from previous steps)
prev_steps_with_substeps = sum(1 for j in range(i) if j in (1, 3))
step_row = i * 2 + 1 + prev_steps_with_substeps
step.grid(
row=i * 2 + 1, column=0,
row=step_row, column=0,
sticky="nsew",
padx=PADDING_SMALL,
pady=(0, 2),
@@ -390,20 +396,31 @@ class MainWindow(ctk.CTk):
if i == 1:
self._sub_step_frame = SubStepFrame(panel)
self._sub_step_frame.grid(
row=i * 2 + 2, column=0,
row=step_row + 1, column=0,
sticky="nsew",
padx=PADDING_SMALL + 16,
pady=(0, 2),
)
# ── Progress indicator ──
# Step 4: add collapsible sub-step frame
if i == 3:
self._tftp_sub_step_frame = TftpSubStepFrame(panel)
self._tftp_sub_step_frame.grid(
row=step_row + 1, column=0,
sticky="nsew",
padx=PADDING_SMALL + 16,
pady=(0, 2),
)
# ── Progress indicator ──
self._progress = ProgressIndicator(panel)
self._progress.grid(
row=8, column=0,
sticky="nsew",
row=11, column=0,
sticky="ew",
padx=PADDING,
pady=(PADDING, PADDING_SMALL),
)
panel.grid_rowconfigure(11, minsize=40)
# ── Run All button ──
self._run_btn = ctk.CTkButton(
@@ -415,10 +432,10 @@ class MainWindow(ctk.CTk):
corner_radius=CORNER_RADIUS,
)
self._run_btn.grid(
row=9, column=0,
row=12, column=0,
sticky="nsew",
padx=PADDING,
pady=PADDING_SMALL,
pady=(0, PADDING_SMALL),
)
def _build_log_panel(self, parent: ctk.CTkFrame) -> None:
@@ -898,7 +915,7 @@ class MainWindow(ctk.CTk):
self._log_message(f" [{status}] {message}")
def _run_step_4_load_main_program(self) -> bool:
"""Step 4: Load main program (TFTP upload + reboot + boot verify)."""
"""Step 4: Load main program (TFTP upload + download verify + CRC + reboot + boot verify)."""
if not self._config:
return False
@@ -907,42 +924,86 @@ class MainWindow(ctk.CTk):
self._log_message(" No Firmware BIN file selected")
return False
bin_path = Path(files["firmware_bin_path"])
remote_name = self._config.tftp_upload_name
all_results: list[bool] = []
# 4a: TFTP upload
self._log_message("Step 4a: TFTP upload & verify...")
# 4a: TFTP Upload
self._log_message("Step 4a: TFTP upload...")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_phase("Upload")
try:
bin_path = Path(files["firmware_bin_path"])
tftp_result = tftp_upload_verify(
self._config, bin_path,
callback=self._tftp_callback,
tftp_result = tftp_upload(
self._config, bin_path, remote_name,
callback=self._tftp_sub_callback,
)
status = "" if tftp_result.success else ""
self._log_message(f" {status} {tftp_result.message}")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_complete("Upload") if tftp_result.success else self._tftp_sub_step_frame.set_step_error("Upload")
all_results.append(tftp_result.success)
except Exception as e:
self._log_message(f" ✗ TFTP error: {e}")
self._log_message(f" ✗ TFTP upload error: {e}")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_error("Upload")
all_results.append(False)
if not all_results[-1]:
self._log_message(" Skipping reboot & boot verify (TFTP failed)")
self._log_message(" Skipping download verify (TFTP upload failed)")
return False
# 4a2: Post-download Zynq status check
self._log_message("Step 4b: Checking Zynq status after download...")
# 4b: TFTP Download Verify
self._log_message("Step 4b: TFTP download verify...")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_phase("Download Verify")
try:
zynq_status = verify_zynq_status(
self._config,
uart_available=self._uart_available,
callback=self._zynq_status_callback,
download_result = tftp_download(
self._config, remote_name,
callback=self._tftp_sub_callback,
)
status = "" if zynq_status.success else ""
self._log_message(f" {status} {zynq_status.message}")
status = "" if download_result.success else ""
self._log_message(f" {status} {download_result.message}")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_complete("Download Verify") if download_result.success else self._tftp_sub_step_frame.set_step_error("Download Verify")
all_results.append(download_result.success)
except Exception as e:
self._log_message(f"Zynq status check error: {e}")
self._log_message(f"TFTP download error: {e}")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_error("Download Verify")
all_results.append(False)
# 4c: Reboot
self._log_message("Step 4c: Rebooting Zynq...")
if not all_results[-1]:
self._log_message(" Skipping CRC check (download failed)")
return False
# 4c: CRC Check
self._log_message("Step 4c: CRC check...")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_phase("CRC Check")
original_crc = _compute_crc32(bin_path)
downloaded_crc = download_result.crc_local
crc_match = original_crc == downloaded_crc
self._log_message(f" Original CRC: {original_crc:#010x}")
self._log_message(f" Downloaded CRC: {downloaded_crc:#010x}")
if crc_match:
self._log_message(" ✓ CRC match")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_complete("CRC Check")
all_results.append(True)
else:
self._log_message(" ✗ CRC mismatch")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_error("CRC Check")
all_results.append(False)
if not all_results[-1]:
self._log_message(" Skipping reboot (CRC mismatch)")
return False
# 4d: Reboot
self._log_message("Step 4d: Rebooting Zynq...")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_phase("Reboot")
try:
reboot_result = reboot_zynq(
self._config,
@@ -951,17 +1012,23 @@ class MainWindow(ctk.CTk):
)
status = "" if reboot_result.success else ""
self._log_message(f" {status} {reboot_result.message}")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_complete("Reboot") if reboot_result.success else self._tftp_sub_step_frame.set_step_error("Reboot")
all_results.append(reboot_result.success)
except Exception as e:
self._log_message(f" ✗ Reboot error: {e}")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_error("Reboot")
all_results.append(False)
if not all_results[-1]:
self._log_message(" Skipping boot verify (reboot failed)")
return False
# 4d: Verify boot
self._log_message("Step 4d: Verifying boot...")
# 4e: Verify boot
self._log_message("Step 4e: Verifying boot...")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_phase("Boot Verify")
try:
boot_result = verify_boot(
self._config,
@@ -975,9 +1042,13 @@ class MainWindow(ctk.CTk):
if boot_result.serial_found:
self._log_message(f" Boot info: {boot_result.boot_info}")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_complete("Boot Verify") if boot_result.success else self._tftp_sub_step_frame.set_step_error("Boot Verify")
all_results.append(boot_result.success)
except Exception as e:
self._log_message(f" ✗ Boot verify error: {e}")
if hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.set_step_error("Boot Verify")
all_results.append(False)
return all(all_results)
@@ -986,6 +1057,10 @@ class MainWindow(ctk.CTk):
"""Callback for TFTP progress."""
self._log_message(f" [{status}] {message}")
def _tftp_sub_callback(self, status: str, message: str) -> None:
"""Callback for TFTP sub-step progress."""
self._log_message(f" [{status}] {message}")
def _reboot_callback(self, status: str, message: str) -> None:
"""Callback for reboot progress."""
self._log_message(f" [{status}] {message}")
@@ -1121,6 +1196,11 @@ class MainWindow(ctk.CTk):
self._sub_step_frame.set_expanded(True)
self._flash_phase = ""
# Step 4: expand sub-step frame and reset
if i == 3 and hasattr(self, '_tftp_sub_step_frame'):
self._tftp_sub_step_frame.reset()
self._tftp_sub_step_frame.set_expanded(True)
# Log step timeout estimate
timeout_key, default_timeout = step_timeout_keys[i] if i < len(step_timeout_keys) else ("", 30)
step_timeout = self._config.step_timeouts.get(timeout_key, default_timeout) if self._config else default_timeout
@@ -1142,6 +1222,13 @@ class MainWindow(ctk.CTk):
self._sub_step_frame.set_phase("done")
else:
self._sub_step_frame.set_phase("error")
# Step 4: mark sub-steps based on result
if i == 3 and hasattr(self, '_tftp_sub_step_frame'):
if success:
self._tftp_sub_step_frame.set_phase("Boot Verify")
else:
self._tftp_sub_step_frame.set_step_error("Boot Verify")
except Exception as e:
results.append(False)
self._set_step_status(i, "error")
@@ -1210,7 +1297,7 @@ class MainWindow(ctk.CTk):
and self._uart_window.is_monitoring):
self._log_message(" [UART] Already monitoring via UART window")
return
if not self._uart_available or not self._config:
if not self._config:
return
port = self._config.serial_port
if not port:
@@ -1226,6 +1313,8 @@ class MainWindow(ctk.CTk):
self._uart_monitor.on_error = lambda e: self._log_message(f" [UART] Error: {e}")
if self._uart_monitor.start():
self._log_message(" [UART] Serial monitor started")
else:
self._log_message(" [UART] Serial monitor failed to start (port may be busy)")
def _stop_uart_monitor(self) -> None:
"""Stop background serial monitoring."""
+154 -13
View File
@@ -392,26 +392,32 @@ class ProgressIndicator(ctk.CTkFrame):
def _create_widgets(self) -> None:
"""Create the progress indicator UI."""
self.grid_rowconfigure(0, weight=1)
self.grid_rowconfigure(0, weight=0)
self.grid_rowconfigure(1, weight=1)
self.grid_columnconfigure(0, weight=1)
self._progress_bar = ctk.CTkProgressBar(
self,
variable=self._progress_var,
corner_radius=CORNER_RADIUS,
height=8,
)
self._progress_bar.grid(
row=0, column=0, sticky="nsew", padx=PADDING
)
# Percentage label on top
self._percent_label = ctk.CTkLabel(
self,
text="0%",
font=FONT_SMALL,
font=(FONT_BODY[0], 13, "bold"),
text_color=PRIMARY_COLOR,
)
self._percent_label.grid(
row=0, column=0, sticky="ne", padx=PADDING, pady=PADDING_SMALL
row=0, column=0, sticky="e", padx=PADDING, pady=(0, 6)
)
# Progress bar below
self._progress_bar = ctk.CTkProgressBar(
self,
variable=self._progress_var,
corner_radius=6,
height=10,
progress_color=PRIMARY_COLOR,
border_width=0,
)
self._progress_bar.grid(
row=1, column=0, sticky="ew", padx=PADDING, pady=2
)
def set_value(self, value: float) -> None:
@@ -765,8 +771,143 @@ class SubStepFrame(ctk.CTkFrame):
if not self._collapsed:
self._toggle()
self._toggle() # collapse then expand to reset
# ════════════════════════════════════════════════════════════════════
# TftpSubStepFrame — collapsible sub-step progress for Step 4
# ════════════════════════════════════════════════════════════════════
class TftpSubStepFrame(ctk.CTkFrame):
"""Collapsible frame showing TFTP workflow sub-steps with status.
Sub-steps: Upload, Download Verify, CRC Check, Reboot, Boot Verify.
Colors match StepHeader: gray pending, blue running (+pulse),
green complete, red error. Default expanded.
"""
def __init__(self, parent: ctk.CTkFrame, substep_names: list[str] | None = None) -> None:
super().__init__(parent, fg_color="transparent")
self._collapsed = False
self._pulse_job: str | None = None
self._substep_names = substep_names or ["Upload", "Download Verify", "CRC Check", "Reboot", "Boot Verify"]
self._num_steps = len(self._substep_names)
self._toggle_btn = ctk.CTkButton(
self, text="", width=22, height=22,
font=(FONT_SMALL[0], 9), command=self._toggle,
)
self._toggle_btn.grid(row=0, column=0, padx=(4, 4), pady=(2, 2), sticky="w")
self._header_label = ctk.CTkLabel(
self, text="".join(self._substep_names),
font=FONT_SMALL,
)
self._header_label.grid(row=0, column=1, padx=(0, 4), pady=(2, 2), sticky="w")
self._sub_items: list[dict] = []
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")
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})
self.grid_columnconfigure(1, weight=1)
def _toggle(self) -> None:
self._collapsed = not self._collapsed
self._toggle_btn.configure(text="" if self._collapsed else "")
for item in self._sub_items:
if self._collapsed:
item["dot"].grid_remove()
item["label"].grid_remove()
else:
item["dot"].grid()
item["label"].grid()
def set_expanded(self, expanded: bool) -> None:
if expanded != self._collapsed:
self._toggle()
def set_phase(self, phase: str, pct: int = -1) -> None:
_C_PENDING = ACCENT_PENDING
_C_RUNNING = CIRCLE_RUNNING
_C_DONE = ACCENT_COMPLETE
idx = self._substep_names.index(phase) if phase in self._substep_names else -1
if idx < 0:
return
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)
elif i == idx:
item["dot"].configure(text="", text_color=_C_RUNNING)
item["label"].configure(text_color=_C_RUNNING)
else:
item["dot"].configure(text="", text_color=_C_PENDING)
item["label"].configure(text_color=_C_PENDING)
if phase in self._substep_names and not self._pulse_job:
self._start_pulse(idx)
def set_step_complete(self, phase: str) -> None:
"""Mark a specific step as complete."""
_C_DONE = ACCENT_COMPLETE
idx = self._substep_names.index(phase) if phase in self._substep_names else -1
if idx < 0:
return
item = self._sub_items[idx]
item["dot"].configure(text="", text_color=_C_DONE)
item["label"].configure(text_color=_C_DONE)
self._stop_pulse()
def set_step_error(self, phase: str) -> None:
"""Mark a specific step as error."""
_C_ERROR = ACCENT_ERROR
idx = self._substep_names.index(phase) if phase in self._substep_names else -1
if idx < 0:
return
item = self._sub_items[idx]
item["dot"].configure(text="", text_color=_C_ERROR)
item["label"].configure(text_color=_C_ERROR)
self._stop_pulse()
def _start_pulse(self, idx: int) -> None:
"""Start pulsing the running sub-step."""
_LIGHT = RUNNING_FADE_LIGHT
_DARK = RUNNING_FADE_DARK
def _pulse(state=[True]):
if idx >= len(self._sub_items):
return
item = self._sub_items[idx]
c = _LIGHT if state[0] else _DARK
item["dot"].configure(text_color=c)
item["label"].configure(text_color=c)
state[0] = not state[0]
self._pulse_job = self.after(PULSE_DURATION, lambda: _pulse(state))
_pulse()
def _stop_pulse(self) -> None:
if self._pulse_job:
self.after_cancel(self._pulse_job)
self._pulse_job = None
def reset(self) -> None:
self._stop_pulse()
_C = ACCENT_PENDING
for i, item in enumerate(self._sub_items):
item["dot"].configure(text="", text_color=_C)
item["label"].configure(text_color=_C)
if not self._collapsed:
self._toggle()
self._toggle() # collapse then expand to reset
# ════════════════════════════════════════════════════════════════════
# UartMonitorWindow — standalone UART monitoring window
+139 -16
View File
@@ -69,18 +69,21 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
"""
data = Path(file_path).read_bytes()
total = len(data)
block_size = _TFTP_BLOCK_SIZE
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(timeout)
# Allow reuse of local port (prevents "Address already in use")
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
# ── Send WRQ ──
# ── Send WRQ to port 69 ──
wrq = struct.pack("!H", _TFTP_WRQ)
wrq += remote_name.encode("ascii") + b"\x00"
wrq += b"octet\x00"
sock.sendto(wrq, (host, 69))
# ── Wait for ACK(0) to get server's data port ──
# ── Wait for ACK(0) — discover server's data port ──
ack_data, server_addr = sock.recvfrom(516)
if len(ack_data) < 4:
return False, "Invalid ACK from server"
@@ -89,24 +92,34 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
err_msg = ack_data[4:].decode("ascii", errors="replace").rstrip("\x00")
return False, f"TFTP error: {err_msg}"
if opcode != _TFTP_ACK:
return False, f"Expected ACK, got opcode {opcode}"
return False, f"Expected ACK(0), got opcode {opcode} from {server_addr}"
srv_ip, srv_port = server_addr
import sys
print(f"[TFTP] ACK(0) received from {srv_ip}:{srv_port} "
f"(file={total}B, blocks={-(-total // block_size)})", file=sys.stderr)
# ── Send data blocks ──
block = 1
offset = 0
while offset < total:
chunk = data[offset:offset + _TFTP_BLOCK_SIZE]
chunk = data[offset:offset + block_size]
offset += len(chunk)
data_pkt = struct.pack("!HH", _TFTP_DATA, block) + chunk
sock.sendto(data_pkt, server_addr)
if block % 100 == 0 or block == -(-total // block_size):
pct = offset * 100 // total
print(f"[TFTP] Block {block}/{-(-total // block_size)} ({pct}%) sent to {srv_ip}:{srv_port}",
file=sys.stderr)
# Wait for ACK
try:
ack_data, _ = sock.recvfrom(516)
ack_op = struct.unpack("!H", ack_data[:2])[0]
ack_block = struct.unpack("!H", ack_data[2:4])[0]
if ack_op != _TFTP_ACK or ack_block != block:
return False, f"Bad ACK: op={ack_op} block={ack_block}"
return False, f"Bad ACK: op={ack_op} block={ack_block} (expected {block})"
except socket.timeout:
return False, f"Timeout waiting for ACK block {block}"
block += 1
@@ -116,6 +129,91 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
sock.close()
def _tftp_get(host: str, remote_name: str, local_dir: str | Path,
timeout: float = 30.0) -> tuple[bool, str, Path]:
"""Download a file from a TFTP server using pure Python.
Args:
host: TFTP server IP address.
remote_name: Remote filename.
local_dir: Local directory to save the downloaded file.
timeout: Socket timeout in seconds.
Returns:
(success, message, local_path).
"""
local_dir = Path(local_dir)
local_dir.mkdir(parents=True, exist_ok=True)
local_path = local_dir / remote_name
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(timeout)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
# ── Send RRQ to port 69 ──
rrq = struct.pack("!H", _TFTP_RRQ)
rrq += remote_name.encode("ascii") + b"\x00"
rrq += b"octet\x00"
sock.sendto(rrq, (host, 69))
# ── Wait for first DATA block — discover server's data port ──
data_pkt, server_addr = sock.recvfrom(516)
if len(data_pkt) < 4:
return False, "Invalid DATA from server", 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} from {server_addr}", local_path
srv_ip, srv_port = server_addr
block = struct.unpack("!H", data_pkt[2:4])[0]
file_data = 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)
# ── ACK block 0 and receive subsequent blocks ──
ack = struct.pack("!HH", _TFTP_ACK, 0)
sock.sendto(ack, server_addr)
total = len(file_data)
while len(file_data) == _TFTP_BLOCK_SIZE:
# Wait for next DATA block
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_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)
# ACK this block
ack = struct.pack("!HH", _TFTP_ACK, block)
sock.sendto(ack, server_addr)
except socket.timeout:
return False, f"Timeout waiting for DATA block {block}", local_path
# ── Write file ──
local_path.write_bytes(file_data)
return True, f"Downloaded {total} bytes to {local_path}", local_path
finally:
sock.close()
def tftp_upload(
config: Config,
file_path: str | Path,
@@ -184,25 +282,50 @@ def tftp_download(
) -> TftpResult:
"""Download a file from the Zynq device via TFTP (pure Python).
Note: This requires the Zynq to act as a TFTP client (U-Boot tftpboot),
which typically means we need a TFTP server on the PC. For simplicity,
this operation is not natively supported — use ping-based verification
or UART log analysis instead.
Args:
config: Application configuration.
remote_name: Remote filename on the TFTP server.
local_dir: Local directory to save the file.
local_dir: Local directory to save the file (default: temp directory).
callback: Optional progress callback.
Returns:
TftpResult with status (always not-implemented).
TftpResult with download status and CRC.
"""
if local_dir is None:
local_dir = tempfile.gettempdir()
local_dir = Path(local_dir)
if callback:
callback("start", f"Downloading {remote_name} from {config.zynq_ip}")
# Verify device is reachable
if not ping_ip(config.zynq_ip, config.ping_count, config.ping_timeout):
return TftpResult(
step="download", success=False,
message=f"Device {config.zynq_ip} not reachable",
remote_path=remote_name,
)
try:
ok, msg, local_path = _tftp_get(config.zynq_ip, remote_name, local_dir)
except OSError as e:
return TftpResult(
step="download", success=False,
message=f"Socket error: {e}",
remote_path=remote_name,
)
crc = _compute_crc32(local_path) if ok else 0
if callback:
callback("complete" if ok else "error",
"Download " + ("succeeded" if ok else "failed"))
return TftpResult(
step="download",
success=False,
message="TFTP download requires PC-side server (not implemented)",
remote_path=remote_name,
step="download", success=ok,
message=msg,
local_path=str(local_path), remote_path=remote_name,
crc_local=crc,
)