From f04fbe2130f9cc33b2282ef69319e6994b852fef Mon Sep 17 00:00:00 2001 From: Jeremy Shen Date: Thu, 11 Jun 2026 11:47:35 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20add=20TFTP=20download=20ver?= =?UTF-8?q?ify,=20Step=204=20sub-steps,=20and=20fix=20UI=20layout?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- src/gui/main_window.py | 159 ++++++++++++++++++++++++++++++--------- src/gui/widgets.py | 167 +++++++++++++++++++++++++++++++++++++---- src/tftp_manager.py | 155 ++++++++++++++++++++++++++++++++++---- 3 files changed, 417 insertions(+), 64 deletions(-) diff --git a/src/gui/main_window.py b/src/gui/main_window.py index f56b0fe..5ac1546 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -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.""" diff --git a/src/gui/widgets.py b/src/gui/widgets.py index 9230fb4..6bbdacc 100644 --- a/src/gui/widgets.py +++ b/src/gui/widgets.py @@ -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 diff --git a/src/tftp_manager.py b/src/tftp_manager.py index 3053343..2a03f39 100644 --- a/src/tftp_manager.py +++ b/src/tftp_manager.py @@ -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, )