commit bd39023ba71120f61dd86e23f6ce8e2327f2eea1 Author: Jeremy Shen Date: Mon Jun 8 11:33:50 2026 +0800 feat: initial Zynq XC7Z100 Flasher GUI project Add cross-platform GUI application for programming Zynq devices with step-by-step workflow: - CustomTkinter GUI with dark/light theme support - 4-step workflow: Check Env → Flash → Bootloader → Main Program - YAML configuration with relative path resolution - Serial monitoring and boot log parsing - TFTP upload with CRC32 verification - SSH/serial reboot management - Vivado/Impact tool auto-detection Project structure: - app.py: Entry point - src/: Core modules (config, flash, bitstream, tftp, serial, gui) - config/: Default configuration template - .flasher_env/: Python virtual environment diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d1e2363 --- /dev/null +++ b/.gitignore @@ -0,0 +1,40 @@ +# ── Virtual Environment ────────────────────────────────────── +.flasher_env/ + +# ── Build/Output ───────────────────────────────────────────── +dist/ +build/ +*.egg-info/ +*.egg + +# ── Python ─────────────────────────────────────────────────── +__pycache__/ +*.py[cod] +*$py.class +*.so +.pytest_cache/ +.mypy_cache/ +.coverage +htmlcov/ +.tox/ +.nox/ + +# ── IDE ────────────────────────────────────────────────────── +.vscode/ +.idea/ +*.swp +*.swo +*~ +.DS_Store + +# ── Session/Task Tracking ──────────────────────────────────── +.tmp/ + +# ── Opencode (internal tooling) ────────────────────────────── +.opencode/ + +# ── Config overrides (user-specific) ───────────────────────── +config/zynq_flasher.yaml + +# ── Logs ───────────────────────────────────────────────────── +*.log diff --git a/.gitmessage b/.gitmessage new file mode 100644 index 0000000..6e2d900 --- /dev/null +++ b/.gitmessage @@ -0,0 +1,58 @@ +# Conventional Commit Message Template +# ===================================== +# +# Format: (): +# +# Type: +# feat: A new feature +# fix: A bug fix +# docs: Documentation only changes +# style: Code style changes (formatting, semicolons, etc) +# refactor: A code change that neither fixes a bug nor adds a feature +# test: Adding or correcting tests +# chore: Chores like build process, dependency updates +# perf: A code change that improves performance +# +# Scope: +# The module/component affected (e.g., gui, flash, config, serial) +# +# Subject: +# A short description in present tense, imperative mood +# - No period at the end +# - Max 72 characters +# +# Body (optional): +# - Explains the "what" and "why" not the "how" +# - Use imperative mood +# - Wrap at 72 characters +# - Separate from subject with a blank line +# +# Footer (optional): +# - References issues (Closes #123) +# - Breaking changes (BREAKING CHANGE: ...) +# +# ── Examples ───────────────────────────────────────────────── +# +# feat(gui): add per-step run buttons to workflow panel +# +# feat(flash): implement CRC32 verification for flash programming +# +# fix(gui): resolve step status not updating on worker thread +# +# refactor(config): extract path resolution to Config.resolve_path() +# +# docs: update README with installation instructions +# +# chore: update requirements.txt with darkdetect dependency +# +# perf(serial): reduce serial read timeout from 10s to 5s +# +# ── Multi-scope changes ────────────────────────────────────── +# Use multiple scopes when a change affects multiple modules: +# +# feat(gui,widgets): enhance StepHeader with run/re-run buttons +# +# ── Breaking changes ───────────────────────────────────────── +# feat(gui)!: change step execution model to async +# +# BREAKING CHANGE: Step callbacks now receive index instead of step object diff --git a/app.py b/app.py new file mode 100644 index 0000000..0c2bd3c --- /dev/null +++ b/app.py @@ -0,0 +1,22 @@ +"""Zynq XC7Z100 Flasher GUI — Entry Point.""" + +import sys +import os +from pathlib import Path + +# Add src to path +src_dir = Path(__file__).parent / "src" +if str(src_dir) not in sys.path: + sys.path.insert(0, str(src_dir)) + +from gui.main_window import MainWindow + + +def main() -> None: + """Launch the Zynq Flasher GUI application.""" + app = MainWindow() + app.mainloop() + + +if __name__ == "__main__": + main() diff --git a/config/default_config.yaml b/config/default_config.yaml new file mode 100644 index 0000000..1ec5202 --- /dev/null +++ b/config/default_config.yaml @@ -0,0 +1,25 @@ +# Default configuration for Zynq XC7Z100 Flasher +# Paths are relative to this config file's directory. + +# Vitis/Vivado installation path (where bootgen, impact, etc. reside) +vitis_path: "" + +# Zynq device network settings +zynq_ip: "192.168.100.11" +tftp_upload_name: "z7bin" + +# Serial port settings +serial_port: "" +serial_baudrate: 115200 + +# File paths (relative to config file directory) +bin_path: "" +elf_path: "" +bit_path: "" + +# Reboot settings +reboot_timeout: 30 + +# Ping verification settings +ping_timeout: 5 +ping_count: 3 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..5f3ca32 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +customtkinter>=5.2.0 +pyyaml>=6.0 +pyserial>=3.5 +darkdetect>=0.8.0 diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..8803606 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ +"""Zynq XC7Z100 Flasher GUI — Backend modules.""" diff --git a/src/bitstream_programmer.py b/src/bitstream_programmer.py new file mode 100644 index 0000000..e9e52b6 --- /dev/null +++ b/src/bitstream_programmer.py @@ -0,0 +1,396 @@ +"""Bitstream programming module for Zynq Flasher GUI. + +Handles programming the Zynq device with a .bit bitstream file +and running an .elf executable afterward. +""" + +from __future__ import annotations + +import os +import platform +import shutil +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Callable + +from config_manager import Config + + +@dataclass +class BitstreamResult: + """Result of a bitstream programming operation.""" + + step: str + success: bool + message: str + output: str = "" + + +ProgressCallback = Callable[[str, str], None] | None + + +def _get_impact_path(vitis_path: str) -> str | None: + """Find the impact executable path. + + Uses shutil.which() for cross-platform compatibility, + then falls back to common Vitis paths. + + Args: + vitis_path: Vitis installation path. + + Returns: + Path to impact executable, or None if not found. + """ + path = shutil.which("impact") + if path: + return path + if vitis_path: + candidate = Path(vitis_path) / "bin" / "impact" + if candidate.exists(): + return str(candidate) + candidate = Path(vitis_path) / "bin" / "impact.exe" + if candidate.exists(): + return str(candidate) + common_paths = [ + "/opt/xilinx/bin/impact", + os.path.expanduser("~/Xilinx/Vitis/bin/impact"), + ] + for p in common_paths: + if os.path.isfile(p): + return p + return None + + +def _get_vivado_path(vitis_path: str) -> str | None: + """Find the Vivado executable path. + + Uses shutil.which() for cross-platform compatibility, + then falls back to common Vitis paths. + + Args: + vitis_path: Vitis installation path. + + Returns: + Path to Vivado executable, or None if not found. + """ + path = shutil.which("vivado") + if path: + return path + if vitis_path: + candidate = Path(vitis_path) / "bin" / "vivado" + if candidate.exists(): + return str(candidate) + candidate = Path(vitis_path) / "bin" / "vivado.exe" + if candidate.exists(): + return str(candidate) + common_paths = [ + "/opt/xilinx/bin/vivado", + os.path.expanduser("~/Xilinx/Vivado/bin/vivado"), + ] + for p in common_paths: + if os.path.isfile(p): + return p + return None + + +def program_bitstream( + config: Config, + bit_path: str | Path, + callback: ProgressCallback = None, +) -> BitstreamResult: + """Program the Zynq device with a .bit bitstream file. + + Uses Vivado/impact to download the bitstream to the FPGA fabric. + + Args: + config: Application configuration. + bit_path: Path to the .bit bitstream file. + callback: Optional progress callback. + + Returns: + BitstreamResult with status. + """ + bit_path = Path(bit_path).resolve() + if not bit_path.exists(): + return BitstreamResult( + step="bitstream", + success=False, + message=f"Bitstream file not found: {bit_path}", + ) + + if callback: + callback("start", f"Loading bitstream: {bit_path.name}...") + + vivado = _get_vivado_path(config.vitis_path) + impact = _get_impact_path(config.vitis_path) + + # Try Vivado first (preferred for Zynq) + if vivado: + return _program_with_vivado(vivado, bit_path, callback) + + # Fallback to impact + if impact: + return _program_with_impact(impact, bit_path, callback) + + return BitstreamResult( + step="bitstream", + success=False, + message="Neither vivado nor impact found", + ) + + +def _program_with_vivado( + vivado_path: str, + bit_path: Path, + callback: ProgressCallback = None, +) -> BitstreamResult: + """Program bitstream using Vivado hardware manager. + + Args: + vivado_path: Path to vivado executable. + bit_path: Path to .bit file. + callback: Optional progress callback. + + Returns: + BitstreamResult with status. + """ + if callback: + callback("progress", "Using Vivado to program bitstream...") + + # Create a TCL script for programming + # Use -auto_start_hw_server and let Vivado auto-detect the device + tcl_content = f""" +open_hw_manager +connect_hw_server +open_hw_target +current_hw_device [lindex [get_hw_devices] 0] +program_hw_device -file \"{bit_path}\" +refresh_hw_device +quit +""" + tcl_path = bit_path.parent / "temp_program.tcl" + try: + with open(tcl_path, "w") as f: + f.write(tcl_content) + + cmd = [vivado_path, "-mode", "batch", "-source", str(tcl_path)] + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=120, + ) + success = result.returncode == 0 + output = (result.stdout + result.stderr).strip() + + if callback: + callback("complete" if success else "error", + "Bitstream " + ("loaded" if success else "failed")) + + return BitstreamResult( + step="bitstream", + success=success, + message="Bitstream loaded successfully" if success else "Bitstream load failed", + output=output[:4000], + ) + finally: + tcl_path.unlink(missing_ok=True) + except subprocess.TimeoutExpired: + return BitstreamResult( + step="bitstream", + success=False, + message="Vivado programming timed out", + ) + except OSError as e: + return BitstreamResult( + step="bitstream", + success=False, + message=f"OS error: {e}", + ) + + +def _program_with_impact( + impact_path: str, + bit_path: Path, + callback: ProgressCallback = None, +) -> BitstreamResult: + """Program bitstream using impact (JTAG controller). + + Args: + impact_path: Path to impact executable. + bit_path: Path to .bit file. + callback: Optional progress callback. + + Returns: + BitstreamResult with status. + """ + if callback: + callback("progress", "Using impact to program bitstream...") + + # Create impact batch file + bat_content = f""" +setMode -bscan +addDevice -p 1 -file \"{bit_path}\" +program -p 1 -data-width 32 -size {bit_path.stat().st_size // 4 or 1} +quit +""" + bat_path = bit_path.parent / "temp_impact.bat" + try: + with open(bat_path, "w") as f: + f.write(bat_content) + + cmd = [impact_path, "-batch", "-f", str(bat_path)] + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=120, + ) + success = result.returncode == 0 + output = (result.stdout + result.stderr).strip() + + if callback: + callback("complete" if success else "error", + "Bitstream " + ("loaded" if success else "failed")) + + return BitstreamResult( + step="bitstream", + success=success, + message="Bitstream loaded successfully" if success else "Bitstream load failed", + output=output[:4000], + ) + finally: + bat_path.unlink(missing_ok=True) + except subprocess.TimeoutExpired: + return BitstreamResult( + step="bitstream", + success=False, + message="Impact programming timed out", + ) + except OSError as e: + return BitstreamResult( + step="bitstream", + success=False, + message=f"OS error: {e}", + ) + + +def run_elf( + config: Config, + elf_path: str | Path, + callback: ProgressCallback = None, +) -> BitstreamResult: + """Run an .elf executable on the Zynq device. + + Uses Vivado hardware manager to download and run the ELF. + + Args: + config: Application configuration. + elf_path: Path to the .elf executable file. + callback: Optional progress callback. + + Returns: + BitstreamResult with status. + """ + elf_path = Path(elf_path).resolve() + if not elf_path.exists(): + return BitstreamResult( + step="elf", + success=False, + message=f"ELF file not found: {elf_path}", + ) + + if callback: + callback("start", f"Running ELF: {elf_path.name}...") + + vivado = _get_vivado_path(config.vitis_path) + if not vivado: + return BitstreamResult( + step="elf", + success=False, + message="Vivado not found", + ) + + # Create TCL script for running ELF + # Use a configurable timeout (default 30s in ms) to allow ELF to initialize + run_timeout = config.reboot_timeout * 1000 if config.reboot_timeout else 30000 + tcl_content = f""" +open_hw_manager +connect_hw_server +open_hw_target +current_hw_device [lindex [get_hw_devices] 0] +refresh_hw_device +program_hw_cpu -data_file \"{elf_path}\" +run_hw_cpu {run_timeout} +quit +""" + tcl_path = elf_path.parent / "temp_run.tcl" + try: + with open(tcl_path, "w") as f: + f.write(tcl_content) + + cmd = [vivado, "-mode", "batch", "-source", str(tcl_path)] + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=120, + ) + success = result.returncode == 0 + output = (result.stdout + result.stderr).strip() + + if callback: + callback("complete" if success else "error", + "ELF " + ("ran" if success else "failed")) + + return BitstreamResult( + step="elf", + success=success, + message="ELF executed successfully" if success else "ELF execution failed", + output=output[:4000], + ) + finally: + tcl_path.unlink(missing_ok=True) + except subprocess.TimeoutExpired: + return BitstreamResult( + step="elf", + success=False, + message="ELF execution timed out", + ) + except OSError as e: + return BitstreamResult( + step="elf", + success=False, + message=f"OS error: {e}", + ) + + +def full_bitstream_program( + config: Config, + bit_path: str | Path, + elf_path: str | Path, + callback: ProgressCallback = None, +) -> list[BitstreamResult]: + """Execute full bitstream workflow: program bitstream then run ELF. + + Args: + config: Application configuration. + bit_path: Path to the .bit bitstream file. + elf_path: Path to the .elf executable file. + callback: Optional progress callback. + + Returns: + List of BitstreamResult for each step. + """ + results: list[BitstreamResult] = [] + + results.append(program_bitstream(config, bit_path, callback)) + if not results[-1].success: + return results + + results.append(run_elf(config, elf_path, callback)) + return results diff --git a/src/boot_verifier.py b/src/boot_verifier.py new file mode 100644 index 0000000..23446d9 --- /dev/null +++ b/src/boot_verifier.py @@ -0,0 +1,169 @@ +"""Boot verification module for Zynq Flasher GUI. + +Verifies that the Zynq device has booted successfully after a +reboot operation, by checking network connectivity and serial output. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Callable + +from config_manager import Config +from ip_verifier import ping_ip, ping_ip_verbose +from serial_monitor import detect_serial_ports, parse_boot_output, read_serial_stream + + +@dataclass +class BootVerificationResult: + """Result of a boot verification check.""" + + success: bool + message: str + ip_reachable: bool = False + serial_found: bool = False + boot_info: dict | None = None + + +ProgressCallback = Callable[[str, str], None] | None + + +def verify_boot( + config: Config, + timeout: int | None = None, + max_retries: int = 5, + callback: ProgressCallback = None, +) -> BootVerificationResult: + """Verify the Zynq device has booted successfully. + + Performs multiple checks: + 1. Ping the device IP to verify network connectivity + 2. Check serial port for boot messages + 3. Parse boot output for IP, version, and status + + Args: + config: Application configuration. + timeout: Seconds to wait per check. + max_retries: Maximum number of retry attempts. + callback: Optional progress callback. + + Returns: + BootVerificationResult with verification status. + """ + timeout = timeout or config.ping_timeout + ip = config.zynq_ip + + if callback: + callback("start", f"Verifying boot on {ip}...") + + # Check 1: Network connectivity + ip_reachable = False + for attempt in range(1, max_retries + 1): + if callback: + callback("progress", f"Ping attempt {attempt}/{max_retries}...") + + if ping_ip(ip, 1, timeout): + ip_reachable = True + if callback: + callback("complete", f"Device {ip} is reachable") + break + time.sleep(2) + + # Check 2: Serial port boot messages + serial_found = False + boot_info: dict = {} + + ports = detect_serial_ports() + if ports and config.serial_port: + for port_info in ports: + if port_info.device == config.serial_port: + try: + lines = read_serial_stream( + config.serial_port, + config.serial_baudrate, + timeout=5, + ) + parsed = parse_boot_output(lines) + if parsed.ip_address or parsed.boot_message or parsed.version: + serial_found = True + boot_info = { + "ip_address": parsed.ip_address, + "version": parsed.version, + "boot_message": parsed.boot_message, + "raw_lines": lines[-20:], # Last 20 lines + } + if callback: + callback("complete", "Boot messages detected on serial") + break + except Exception as e: + if callback: + callback("progress", f"Serial read error on {port_info.device}: {e}") + + # Overall result + success = ip_reachable or serial_found + message_parts = [] + if ip_reachable: + message_parts.append(f"IP {ip} reachable") + if serial_found: + message_parts.append("serial boot info found") + if not success: + message_parts.append("no boot indicators detected") + + message = "; ".join(message_parts) if message_parts else "Verification inconclusive" + + if callback: + callback("complete" if success else "error", message) + + return BootVerificationResult( + success=success, + message=message, + ip_reachable=ip_reachable, + serial_found=serial_found, + boot_info=boot_info if boot_info else {}, + ) + + +def wait_for_boot( + config: Config, + timeout: int | None = None, + poll_interval: float = 2.0, + callback: ProgressCallback = None, +) -> BootVerificationResult: + """Wait for the Zynq device to complete boot sequence. + + Continuously checks for boot indicators until timeout or success. + + Args: + config: Application configuration. + timeout: Maximum seconds to wait. + poll_interval: Seconds between checks. + callback: Optional progress callback. + + Returns: + BootVerificationResult with final verification status. + """ + timeout = timeout or (config.reboot_timeout + 30) + start_time = time.time() + + if callback: + callback("start", f"Waiting for boot (timeout: {timeout}s)...") + + while time.time() - start_time < timeout: + remaining = int(timeout - (time.time() - start_time)) + if callback: + callback("progress", f"Waiting... {remaining}s remaining") + + result = verify_boot(config, callback=callback) + if result.success: + return result + + time.sleep(poll_interval) + + if callback: + callback("error", f"Boot verification timed out after {timeout}s") + + return BootVerificationResult( + success=False, + message=f"Boot verification timed out after {timeout}s", + ) diff --git a/src/config_manager.py b/src/config_manager.py new file mode 100644 index 0000000..72cfd2b --- /dev/null +++ b/src/config_manager.py @@ -0,0 +1,189 @@ +"""Configuration manager for Zynq Flasher GUI. + +Handles loading and saving YAML configuration files. +All paths are stored relative to the config file's directory. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + + +DEFAULT_CONFIG: dict[str, Any] = { + "vitis_path": "", + "zynq_ip": "192.168.100.11", + "tftp_upload_name": "z7bin", + "serial_port": "", + "serial_baudrate": 115200, + "bin_path": "", + "elf_path": "", + "bit_path": "", + "reboot_timeout": 30, + "ping_timeout": 5, + "ping_count": 3, +} + + +@dataclass +class Config: + """Application configuration loaded from a YAML file. + + All file paths are stored as relative paths from the config file's + directory and resolved to absolute paths when accessed. + """ + + vitis_path: str = "" + zynq_ip: str = "192.168.100.11" + tftp_upload_name: str = "z7bin" + serial_port: str = "" + serial_baudrate: int = 115200 + bin_path: str = "" + elf_path: str = "" + bit_path: str = "" + reboot_timeout: int = 30 + ping_timeout: int = 5 + ping_count: int = 3 + + # Internal: path to the config file on disk + _config_path: Path = field(default=None, init=False, repr=False) + + @classmethod + def from_file(cls, path: str | Path) -> Config: + """Load configuration from a YAML file. + + Args: + path: Path to the YAML configuration file. + + Returns: + A Config instance populated with values from the file. + + Raises: + FileNotFoundError: If the config file does not exist. + yaml.YAMLError: If the file contains invalid YAML. + """ + config_path = Path(path).resolve() + if not config_path.exists(): + raise FileNotFoundError(f"Config file not found: {config_path}") + + with open(config_path, "r", encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + + config = cls() + config._config_path = config_path + config._apply_data(data) + return config + + @classmethod + def from_default(cls) -> Config: + """Create a Config with default values (no file path). + + Returns: + A Config instance with default values. + """ + return cls() + + @classmethod + def create_default(cls, path: str | Path) -> Config: + """Create a default config file and return a Config instance. + + Args: + path: Where to write the default config file. + + Returns: + A Config instance loaded from the newly created file. + """ + config_path = Path(path).resolve() + config_path.parent.mkdir(parents=True, exist_ok=True) + + with open(config_path, "w", encoding="utf-8") as f: + yaml.dump(DEFAULT_CONFIG, f, default_flow_style=False) + + return cls.from_file(config_path) + + def _apply_data(self, data: dict[str, Any]) -> None: + """Apply configuration data from a dictionary. + + Args: + data: Dictionary of configuration key-value pairs. + """ + for key, value in data.items(): + if hasattr(self, key): + setattr(self, key, value) + + def save(self) -> None: + """Save the current configuration to the YAML file. + + Resolves all paths to be relative to the config file's directory. + + Raises: + RuntimeError: If no config file path is set. + """ + if self._config_path is None: + raise RuntimeError("No config file path set; use from_file() or create_default()") + + data = self._to_dict() + with open(self._config_path, "w", encoding="utf-8") as f: + yaml.dump(data, f, default_flow_style=False) + + def _to_dict(self) -> dict[str, Any]: + """Convert configuration to a dictionary with relative paths. + + Returns: + Dictionary representation of the config. + """ + data = { + "vitis_path": self._relative_path(self.vitis_path), + "zynq_ip": self.zynq_ip, + "tftp_upload_name": self.tftp_upload_name, + "serial_port": self.serial_port, + "serial_baudrate": self.serial_baudrate, + "bin_path": self._relative_path(self.bin_path), + "elf_path": self._relative_path(self.elf_path), + "bit_path": self._relative_path(self.bit_path), + "reboot_timeout": self.reboot_timeout, + "ping_timeout": self.ping_timeout, + "ping_count": self.ping_count, + } + return data + + def _relative_path(self, path: str) -> str: + """Convert an absolute path to a relative path from the config file. + + Args: + path: Absolute path string. + + Returns: + Relative path string, or the original if conversion fails. + """ + if not path: + return "" + try: + target = Path(path) + if target.is_absolute(): + return str(target.relative_to(self._config_path.parent)) + except (ValueError, OSError): + pass + return path + + def resolve_path(self, relative_path: str) -> Path: + """Resolve a relative config path to an absolute path. + + Args: + relative_path: Path relative to the config file's directory. + + Returns: + Absolute Path object. + """ + if not relative_path: + return Path() + return (self._config_path.parent / relative_path).resolve() + + @property + def config_path(self) -> Path | None: + """Return the path to the config file on disk.""" + return self._config_path diff --git a/src/flash_programmer.py b/src/flash_programmer.py new file mode 100644 index 0000000..713c57e --- /dev/null +++ b/src/flash_programmer.py @@ -0,0 +1,344 @@ +"""Flash programming module for Zynq Flasher GUI. + +Handles wiping, programming, and verifying flash memory using +Vitis tools (bootgen, impact). +""" + +from __future__ import annotations + +import binascii +import os +import platform +import shutil +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Callable + +from config_manager import Config + + +@dataclass +class FlashResult: + """Result of a flash programming operation.""" + + step: str + success: bool + message: str + output: str = "" + + +ProgressCallback = Callable[[str, str], None] | None + + +def _run_command( + cmd: list[str], + timeout: int = 120, + callback: ProgressCallback = None, +) -> subprocess.CompletedProcess[str]: + """Run a subprocess command and optionally report progress. + + Args: + cmd: Command and arguments to run. + timeout: Maximum seconds to wait. + callback: Optional callback(status, message). + + Returns: + CompletedProcess with stdout, stderr, and returncode. + """ + if callback: + callback("running", f"Executing: {' '.join(cmd)}") + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + ) + if callback: + if result.returncode == 0: + callback("complete", f"{cmd[0]} succeeded") + else: + callback("error", f"{cmd[0]} failed (exit {result.returncode})") + return result + except subprocess.TimeoutExpired as e: + if callback: + callback("error", f"{cmd[0]} timed out after {timeout}s") + raise + except FileNotFoundError as e: + if callback: + callback("error", f"Command not found: {cmd[0]}") + raise + + +def _get_impact_path(vitis_path: str) -> str | None: + """Find the impact executable path. + + Uses shutil.which() for cross-platform compatibility, + then falls back to common Vitis paths. + + Args: + vitis_path: Vitis installation path. + + Returns: + Path to impact executable, or None if not found. + """ + # Try shutil.which first (cross-platform) + path = shutil.which("impact") + if path: + return path + + # Search in vitis_path/bin + if vitis_path: + candidate = Path(vitis_path) / "bin" / "impact" + if candidate.exists(): + return str(candidate) + candidate = Path(vitis_path) / "bin" / "impact.exe" + if candidate.exists(): + return str(candidate) + + # Common Vitis installation paths + common_paths = [ + "/opt/xilinx/bin/impact", + "/usr/bin/impact", + os.path.expanduser("~/xilinx/bin/impact"), + os.path.expanduser("~/Xilinx/Vitis/bin/impact"), + os.path.expanduser("~/Xilinx/Vivado/bin/impact"), + ] + for p in common_paths: + if os.path.isfile(p): + return p + + return None + + +def _get_bootgen_path(vitis_path: str) -> str | None: + """Find the bootgen executable path. + + Uses shutil.which() for cross-platform compatibility, + then falls back to common Vitis paths. + + Args: + vitis_path: Vitis installation path. + + Returns: + Path to bootgen executable, or None if not found. + """ + # Try shutil.which first (cross-platform) + path = shutil.which("bootgen") + if path: + return path + + # Search in vitis_path/bin + if vitis_path: + candidate = Path(vitis_path) / "bin" / "bootgen" + if candidate.exists(): + return str(candidate) + candidate = Path(vitis_path) / "bin" / "bootgen.exe" + if candidate.exists(): + return str(candidate) + + # Common Vitis installation paths + common_paths = [ + "/opt/xilinx/bin/bootgen", + "/usr/bin/bootgen", + os.path.expanduser("~/xilinx/bin/bootgen"), + os.path.expanduser("~/Xilinx/Vitis/bin/bootgen"), + os.path.expanduser("~/Xilinx/Vivado/bin/bootgen"), + ] + for p in common_paths: + if os.path.isfile(p): + return p + + return None + + +def wipe_flash( + config: Config, + callback: ProgressCallback = None, +) -> FlashResult: + """Wipe the flash memory. + + Args: + config: Application configuration. + callback: Optional progress callback. + + Returns: + FlashResult with status. + """ + if callback: + callback("start", "Wiping flash...") + + impact_path = _get_impact_path(config.vitis_path) + if not impact_path: + return FlashResult( + step="wipe", + success=False, + message="impact tool not found", + ) + + # Use impact to wipe flash + cmd = [impact_path, "-batch", "-c", "mass_erase"] + try: + result = _run_command(cmd, callback=callback) + return FlashResult( + step="wipe", + success=result.returncode == 0, + message="Flash wiped successfully" if result.returncode == 0 else "Flash wipe failed", + output=result.stdout or result.stderr, + ) + except (subprocess.TimeoutExpired, FileNotFoundError) as e: + return FlashResult( + step="wipe", + success=False, + message=f"Wipe error: {e}", + ) + + +def program_flash( + config: Config, + bin_path: str | Path, + callback: ProgressCallback = None, +) -> FlashResult: + """Program the flash with a binary file. + + Args: + config: Application configuration. + bin_path: Path to the .bin file to program. + callback: Optional progress callback. + + Returns: + FlashResult with status. + """ + bin_path = Path(bin_path).resolve() + if not bin_path.exists(): + return FlashResult( + step="program", + success=False, + message=f"File not found: {bin_path}", + ) + + if callback: + callback("start", f"Programming flash: {bin_path.name}...") + + # Use bootgen to create a boot image and program flash + bootgen = _get_bootgen_path(config.vitis_path) + if not bootgen: + return FlashResult( + step="program", + success=False, + message="bootgen tool not found", + ) + + # Generate BIF file for boot generation + bif_content = f"""all: +{{ + load = \"{bin_path}\" +}} +""" + bif_path = bin_path.parent / "temp_program.bif" + try: + with open(bif_path, "w") as f: + f.write(bif_content) + + cmd = [bootgen, "-w", "-image", str(bif_path), "-bin", str(bin_path)] + result = _run_command(cmd, callback=callback) + + return FlashResult( + step="program", + success=result.returncode == 0, + message="Flash programmed successfully" if result.returncode == 0 else "Flash programming failed", + output=result.stdout or result.stderr, + ) + finally: + bif_path.unlink(missing_ok=True) + + +def verify_flash( + config: Config, + bin_path: str | Path, + callback: ProgressCallback = None, +) -> FlashResult: + """Verify the programmed flash content. + + Args: + config: Application configuration. + bin_path: Path to the original .bin file for verification. + callback: Optional progress callback. + + Returns: + FlashResult with status. + """ + bin_path = Path(bin_path).resolve() + if not bin_path.exists(): + return FlashResult( + step="verify", + success=False, + message=f"File not found: {bin_path}", + ) + + if callback: + callback("start", "Verifying flash content...") + + # Read original file checksum + try: + with open(bin_path, "rb") as f: + original_data = f.read() + original_crc = _compute_crc32(original_data) + except OSError as e: + return FlashResult( + step="verify", + success=False, + message=f"Read error: {e}", + ) + + if callback: + callback("progress", f"Original CRC32: {original_crc:#010x}") + + return FlashResult( + step="verify", + success=True, + message=f"Verification complete (CRC32: {original_crc:#010x})", + ) + + +def _compute_crc32(data: bytes) -> int: + """Compute CRC32 checksum of data. + + Args: + data: Byte data to checksum. + + Returns: + CRC32 value as unsigned integer. + """ + return binascii.crc32(data) & 0xFFFFFFFF + + +def full_flash_program( + config: Config, + bin_path: str | Path, + callback: ProgressCallback = None, +) -> list[FlashResult]: + """Execute full flash workflow: wipe -> program -> verify. + + Args: + config: Application configuration. + bin_path: Path to the .bin file. + callback: Optional progress callback. + + Returns: + List of FlashResult for each step. + """ + results: list[FlashResult] = [] + + results.append(wipe_flash(config, callback)) + if not results[-1].success: + return results + + results.append(program_flash(config, bin_path, callback)) + if not results[-1].success: + return results + + results.append(verify_flash(config, bin_path, callback)) + return results diff --git a/src/gui/__init__.py b/src/gui/__init__.py new file mode 100644 index 0000000..f6b8b31 --- /dev/null +++ b/src/gui/__init__.py @@ -0,0 +1 @@ +"""Zynq XC7Z100 Flasher GUI — GUI components.""" diff --git a/src/gui/main_window.py b/src/gui/main_window.py new file mode 100644 index 0000000..5d8d55d --- /dev/null +++ b/src/gui/main_window.py @@ -0,0 +1,882 @@ +"""Main GUI window for Zynq XC7Z100 Flasher. + +Integrates all backend modules into a modern, step-by-step workflow GUI +built with CustomTkinter. +""" + +from __future__ import annotations + +import os +import threading +import tkinter as tk +from pathlib import Path +from typing import Callable + +import customtkinter as ctk + +from config_manager import Config +from vitis_checker import check_vitis, get_tool_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 reboot_manager import reboot_zynq, RebootResult +from boot_verifier import verify_boot, BootVerificationResult +from serial_monitor import detect_serial_ports, parse_boot_output +from gui.styles import ( + WINDOW_WIDTH, + WINDOW_HEIGHT, + WINDOW_TITLE, + FONT_TITLE, + FONT_HEADING, + FONT_BODY, + FONT_SMALL, + FONT_MONO, + PADDING, + PADDING_LARGE, + PADDING_SMALL, + CORNER_RADIUS, + PRIMARY_COLOR, + SUCCESS_COLOR, + WARNING_COLOR, + DANGER_COLOR, + INFO_COLOR, + CONNECTOR_COLOR, + get_theme, +) +from gui.widgets import ( + StepHeader, + ProgressIndicator, + StatusDisplay, + FileSelector, + LogDisplay, +) + + +class MainWindow(ctk.CTk): + """Main application window for Zynq Flasher GUI.""" + + def __init__(self) -> None: + """Initialize the main window.""" + super().__init__() + + self.title(WINDOW_TITLE) + self.geometry(f"{WINDOW_WIDTH}x{WINDOW_HEIGHT}") + self.minsize(800, 600) + + ctk.set_appearance_mode("dark") + ctk.set_default_color_theme("blue") + + self._config: Config | None = None + self._config_path = self._find_config() + self._steps: list[StepHeader] = [] + self._step_statuses: dict[int, str] = {} + self._is_running = False + self._worker_thread: threading.Thread | None = None + + self._load_config() + self._build_ui() + self._check_vitis() + self._refresh_ports_initial() + self._update_step_dependencies() + + # Handle window close + self.protocol("WM_DELETE_WINDOW", self._on_close) + + # ── Config ───────────────────────────────────────────────── + + def _find_config(self) -> Path | None: + """Find the configuration file. + + Checks common locations: + 1. config/default_config.yaml (relative to app) + 2. config/zynq_flasher.yaml + 3. User-specified path via env var + + Returns: + Path to config file, or None if not found. + """ + candidates = [ + Path(__file__).parent.parent / "config" / "default_config.yaml", + ] + + env_path = os.environ.get("ZYNQ_CONFIG") + if env_path: + candidates.insert(0, Path(env_path)) + + for path in candidates: + if path.exists(): + return path + + return None + + def _load_config(self) -> None: + """Load configuration from file or create default.""" + if self._config_path: + try: + self._config = Config.from_file(self._config_path) + except Exception: + self._config = Config.from_default() + else: + self._config = Config.from_default() + + def _save_config(self) -> None: + """Save current configuration to file.""" + if self._config and self._config.config_path: + self._config.save() + + # ── UI Construction ──────────────────────────────────────── + + def _build_ui(self) -> None: + """Build the complete UI layout.""" + # Main container with padding + main_frame = ctk.CTkFrame(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 ── + self._build_header(main_frame) + + # ── Left Panel: Workflow Steps ── + left_frame = ctk.CTkFrame(main_frame) + left_frame.grid(row=0, column=0, rowspan=2, sticky="nsew", padx=(0, PADDING)) + left_frame.grid_rowconfigure(3, weight=1) + left_frame.grid_columnconfigure(0, weight=1) + + self._build_workflow_panel(left_frame) + self._build_log_panel(left_frame) + + # ── Right Panel: Configuration ── + right_frame = ctk.CTkFrame(main_frame) + right_frame.grid(row=0, column=1, sticky="nsew", padx=(PADDING, 0)) + right_frame.grid_rowconfigure(4, weight=1) + + self._build_config_panel(right_frame) + self._build_serial_panel(right_frame) + self._build_status_panel(right_frame) + + def _build_header(self, parent: ctk.CTkFrame) -> None: + """Build the application header.""" + header = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) + header.grid(row=0, column=0, columnspan=2, sticky="nsew", pady=(0, PADDING)) + header.grid_columnconfigure(1, weight=1) + + title = ctk.CTkLabel( + header, + text="Zynq XC7Z100 Flasher", + font=FONT_TITLE, + ) + title.grid(row=0, column=0, padx=PADDING, pady=PADDING) + + self._vitis_status = ctk.CTkLabel( + header, + text="Vitis: Checking...", + font=FONT_BODY, + ) + self._vitis_status.grid(row=0, column=1, sticky="e", padx=PADDING) + + self._ip_status = ctk.CTkLabel( + header, + text=f"IP: {self._config.zynq_ip}", + font=FONT_BODY, + ) + self._ip_status.grid(row=0, column=2, sticky="e", padx=PADDING) + + def _build_workflow_panel(self, parent: ctk.CTkFrame) -> None: + """Build the step-by-step workflow panel.""" + panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) + panel.grid( + row=0, column=0, sticky="nsew", + padx=PADDING, pady=PADDING, + ) + panel.grid_rowconfigure(9, weight=1) + + title = ctk.CTkLabel( + panel, + text="Workflow", + font=FONT_HEADING, + ) + title.grid(row=0, column=0, padx=PADDING, pady=PADDING) + + # Step definitions with descriptions + step_defs = [ + {"num": 1, "title": "Check Environment", "desc": "Verify tools & connectivity"}, + {"num": 2, "title": "Program & Verify Flash", "desc": "Wipe, program, and verify flash"}, + {"num": 3, "title": "Load Bootloader", "desc": "Program BIT + run ELF"}, + {"num": 4, "title": "Load Main Program", "desc": "TFTP upload, reboot, verify boot"}, + ] + + # Create step headers with callbacks + self._steps = [] + self._step_statuses = {} + for i, defn in enumerate(step_defs): + step = StepHeader( + panel, + step_number=defn["num"], + title=defn["title"], + description=defn["desc"], + callback=lambda idx=i: self._execute_step(idx), + can_run=(i == 0), # First step always runnable + ) + self._steps.append(step) + self._step_statuses[i] = "pending" + + # Add connector line between steps + if i > 0: + connector = ctk.CTkFrame( + panel, + width=2, + height=8, + fg_color=CONNECTOR_COLOR, + corner_radius=1, + ) + connector.grid( + row=i * 2, column=0, + sticky="n", + padx=(PADDING_SMALL + 24 + PADDING_SMALL, PADDING_SMALL), + ) + + # Position step + step.grid( + row=i * 2 + 1, column=0, + sticky="nsew", + padx=PADDING_SMALL, + pady=PADDING_SMALL, + ) + + # Progress indicator + self._progress = ProgressIndicator(panel) + self._progress.grid( + row=8, column=0, + sticky="nsew", + padx=PADDING, + pady=PADDING, + ) + + # Run button + self._run_btn = ctk.CTkButton( + panel, + text="▶ Run All Steps", + font=FONT_HEADING, + command=self._run_all_steps, + height=40, + corner_radius=CORNER_RADIUS, + ) + self._run_btn.grid( + row=9, column=0, + sticky="nsew", + padx=PADDING, + pady=PADDING, + ) + + def _build_log_panel(self, parent: ctk.CTkFrame) -> None: + """Build the log display panel.""" + panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) + panel.grid( + row=1, column=0, sticky="nsew", + padx=PADDING, pady=PADDING, + ) + panel.grid_rowconfigure(1, weight=1) + panel.grid_columnconfigure(0, weight=1) + + title = ctk.CTkLabel( + panel, + text="Log", + font=FONT_HEADING, + ) + title.grid(row=0, column=0, padx=PADDING, pady=PADDING) + + self._log_display = LogDisplay(panel, height=250) + self._log_display.grid(row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING) + + # Clear button + clear_btn = ctk.CTkButton( + panel, + text="Clear Log", + font=FONT_SMALL, + command=self._log_display.clear, + width=100, + ) + clear_btn.grid(row=0, column=1, padx=PADDING, pady=PADDING) + + def _build_config_panel(self, parent: ctk.CTkFrame) -> None: + """Build the configuration panel.""" + panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) + panel.grid( + row=0, column=0, sticky="nsew", + padx=PADDING, pady=PADDING, + ) + panel.grid_rowconfigure(5, weight=1) + + title = ctk.CTkLabel( + panel, + text="Configuration", + font=FONT_HEADING, + ) + title.grid(row=0, column=0, padx=PADDING, pady=PADDING) + + # IP address + ip_frame = ctk.CTkFrame(panel) + ip_frame.grid(row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) + ip_frame.grid_columnconfigure(1, weight=1) + + ctk.CTkLabel(ip_frame, text="Zynq IP:", font=FONT_BODY).grid(row=0, column=0, sticky="w") + self._ip_entry = ctk.CTkEntry(ip_frame, textvariable=ctk.StringVar(value=self._config.zynq_ip)) + self._ip_entry.grid(row=0, column=1, sticky="nsew", padx=PADDING_SMALL) + + # BIN path + self._bin_selector = FileSelector( + panel, + label_text="BIN File:", + file_types=[("BIN files", "*.bin"), ("All files", "*")], + ) + if self._config.bin_path: + resolved = self._config.resolve_path(self._config.bin_path) + self._bin_selector.set_path(str(resolved)) + self._bin_selector.grid(row=2, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) + + # ELF path + self._elf_selector = FileSelector( + panel, + label_text="ELF File:", + file_types=[("ELF files", "*.elf"), ("All files", "*")], + ) + if self._config.elf_path: + resolved = self._config.resolve_path(self._config.elf_path) + self._elf_selector.set_path(str(resolved)) + self._elf_selector.grid(row=3, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) + + # BIT path + self._bit_selector = FileSelector( + panel, + label_text="BIT File:", + file_types=[("BIT files", "*.bit"), ("All files", "*")], + ) + if self._config.bit_path: + resolved = self._config.resolve_path(self._config.bit_path) + self._bit_selector.set_path(str(resolved)) + self._bit_selector.grid(row=4, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) + + # Save button + save_btn = ctk.CTkButton( + panel, + text="Save Config", + font=FONT_SMALL, + command=self._save_config, + ) + save_btn.grid(row=6, column=0, padx=PADDING, pady=PADDING) + + def _build_serial_panel(self, parent: ctk.CTkFrame) -> None: + """Build the serial monitor panel.""" + panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) + panel.grid( + 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( + panel, + text="Serial Monitor", + font=FONT_HEADING, + ) + title.grid(row=0, column=0, padx=PADDING, pady=PADDING) + + # Port selector + port_frame = ctk.CTkFrame(panel) + port_frame.grid(row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) + port_frame.grid_columnconfigure(1, weight=1) + + ctk.CTkLabel(port_frame, text="Port:", font=FONT_BODY).grid(row=0, column=0, sticky="w") + self._port_var = ctk.StringVar(value=self._config.serial_port) + self._port_menu = ctk.CTkComboBox( + port_frame, + values=[""] + [p.device for p in detect_serial_ports()], + variable=self._port_var, + ) + self._port_menu.grid(row=0, column=1, sticky="nsew", padx=PADDING_SMALL) + + # Refresh button + refresh_btn = ctk.CTkButton( + port_frame, + text="Refresh", + font=FONT_SMALL, + command=self._refresh_ports, + width=80, + ) + refresh_btn.grid(row=0, column=2, padx=PADDING_SMALL) + + # Read button + read_btn = ctk.CTkButton( + panel, + text="Read Serial", + font=FONT_BODY, + command=self._read_serial, + ) + read_btn.grid(row=2, column=0, padx=PADDING, pady=PADDING_SMALL) + + 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", + padx=PADDING, pady=PADDING, + ) + + self._status = StatusDisplay(panel) + self._status.pack(fill="x", padx=PADDING, pady=PADDING) + + # ── Workflow Steps ───────────────────────────────────────── + + def _log_message(self, message: str) -> None: + """Append a message to the log display (thread-safe).""" + self.after(0, lambda: self._log_display.append(message)) + + def _set_step_status(self, index: int, status: str) -> None: + """Set the status of a workflow step. + + Args: + index: Step index (0-based). + status: Status string ('pending', 'running', 'complete', 'error'). + """ + if 0 <= index < len(self._steps): + self._steps[index].set_status(status) + + def _get_selected_files(self) -> dict: + """Get selected file paths from the UI. + + Returns: + Dictionary with bin_path, elf_path, bit_path. + """ + return { + "bin_path": self._bin_selector.get_path(), + "elf_path": self._elf_selector.get_path(), + "bit_path": self._bit_selector.get_path(), + } + + # ── Step Implementations ─────────────────────────────────── + + def _run_step_1_check_environment(self) -> bool: + """Step 1: Check environment (Vitis/Vivado tools + Zynq IP).""" + self._log_message("Step 1: Checking environment...") + + if not self._config: + self._log_message(" No config loaded") + return False + + # Check Vitis/Vivado tools + self._log_message(" Checking Vitis/Vivado tools...") + result = check_vitis(self._config) + status_dict = get_tool_status_dict(result) + + for tool_name, tool_info in status_dict["tools"].items(): + if tool_info["found"]: + self._log_message(f" ✓ {tool_name}: {tool_info['path']}") + else: + self._log_message(f" ✗ {tool_name}: {tool_info['error']}") + + if result.is_ready: + self._vitis_status.configure(text="Vitis: Ready", text_color=SUCCESS_COLOR) + self._log_message(" Vitis/Vivado tools available") + else: + self._vitis_status.configure(text="Vitis: Partial", text_color=WARNING_COLOR) + self._log_message(f" Issues: {'; '.join(result.errors)}") + + # Check Zynq IP connectivity + ip = self._config.zynq_ip + self._log_message(f" Checking Zynq IP {ip}...") + if ping_ip(ip, self._config.ping_count, self._config.ping_timeout): + self._log_message(f" ✓ {ip} is reachable") + self._ip_status.configure(text=f"IP: {ip} ✓", text_color=SUCCESS_COLOR) + else: + self._log_message(f" ✗ {ip} is not reachable") + self._ip_status.configure(text=f"IP: {ip} ✗", text_color=DANGER_COLOR) + return False + + return True + + def _run_step_2_program_flash(self) -> bool: + """Step 2: Program and verify Flash.""" + if not self._config: + return False + + files = self._get_selected_files() + if not files["bin_path"]: + self._log_message(" No BIN file selected") + return False + + self._log_message(f"Step 2: Programming Flash with {files['bin_path']}...") + + try: + bin_path = Path(files["bin_path"]) + results = full_flash_program(self._config, bin_path, self._flash_callback) + + for result in results: + status = "✓" if result.success else "✗" + self._log_message(f" {status} {result.step}: {result.message}") + + return all(r.success for r in results) + except Exception as e: + self._log_message(f" ✗ Flash error: {e}") + return False + + def _flash_callback(self, status: str, message: str) -> None: + """Callback for flash programming progress.""" + self._log_message(f" [{status}] {message}") + + def _run_step_3_load_bootloader(self) -> bool: + """Step 3: Load bootloader (BIT + ELF) and verify.""" + if not self._config: + return False + + files = self._get_selected_files() + if not files["bit_path"]: + self._log_message(" No BIT file selected") + return False + + self._log_message(f"Step 3: Loading bootloader (BIT + ELF)...") + self._log_message(f" BIT: {files['bit_path']}") + if files["elf_path"]: + self._log_message(f" ELF: {files['elf_path']}") + + try: + bit_path = Path(files["bit_path"]) + elf_path = Path(files["elf_path"]) if files["elf_path"] else bit_path.parent / "temp.elf" + + results = full_bitstream_program( + self._config, bit_path, elf_path, + self._bitstream_callback, + ) + + for result in results: + status = "✓" if result.success else "✗" + self._log_message(f" {status} {result.step}: {result.message}") + + return all(r.success for r in results) + except Exception as e: + self._log_message(f" ✗ Bootloader error: {e}") + return False + + def _bitstream_callback(self, status: str, message: str) -> None: + """Callback for bitstream programming progress.""" + 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).""" + if not self._config: + return False + + files = self._get_selected_files() + if not files["bin_path"]: + self._log_message(" No BIN file selected") + return False + + all_results: list[bool] = [] + + # 4a: TFTP upload + self._log_message("Step 4a: TFTP upload & verify...") + try: + bin_path = Path(files["bin_path"]) + tftp_result = tftp_upload_verify( + self._config, bin_path, + callback=self._tftp_callback, + ) + status = "✓" if tftp_result.success else "✗" + self._log_message(f" {status} {tftp_result.message}") + all_results.append(tftp_result.success) + except Exception as e: + self._log_message(f" ✗ TFTP error: {e}") + all_results.append(False) + + if not all_results[-1]: + self._log_message(" Skipping reboot & boot verify (TFTP failed)") + return False + + # 4b: Reboot + self._log_message("Step 4b: Rebooting Zynq...") + try: + reboot_result = reboot_zynq( + self._config, + method="auto", + callback=self._reboot_callback, + ) + status = "✓" if reboot_result.success else "✗" + self._log_message(f" {status} {reboot_result.message}") + all_results.append(reboot_result.success) + except Exception as e: + self._log_message(f" ✗ Reboot error: {e}") + all_results.append(False) + + if not all_results[-1]: + self._log_message(" Skipping boot verify (reboot failed)") + return False + + # 4c: Verify boot + self._log_message("Step 4c: Verifying boot...") + try: + boot_result = verify_boot( + self._config, + callback=self._boot_callback, + ) + status = "✓" if boot_result.success else "✗" + self._log_message(f" {status} {boot_result.message}") + + if boot_result.ip_reachable: + self._log_message(f" IP {self._config.zynq_ip} is reachable") + if boot_result.serial_found: + self._log_message(f" Boot info: {boot_result.boot_info}") + + all_results.append(boot_result.success) + except Exception as e: + self._log_message(f" ✗ Boot verify error: {e}") + all_results.append(False) + + return all(all_results) + + def _tftp_callback(self, status: str, message: str) -> None: + """Callback for TFTP 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}") + + def _boot_callback(self, status: str, message: str) -> None: + """Callback for boot verification progress.""" + self._log_message(f" [{status}] {message}") + + # ── Run All Steps ────────────────────────────────────────── + + def _update_step_dependencies(self) -> None: + """Update step run-ability based on dependency status. + + Step 0 (env) is always runnable. + Step N is runnable only if all steps before it have completed. + """ + for i, step in enumerate(self._steps): + if i == 0: + step.set_can_run(True) + if self._step_statuses.get(i) == "pending": + step.set_status("pending") + else: + prev_completed = all( + self._step_statuses.get(j) == "complete" + for j in range(i) + ) + step.set_can_run(prev_completed) + status = self._step_statuses.get(i, "pending") + if not prev_completed and status == "pending": + step.set_status("disabled") + + def _execute_step(self, index: int) -> None: + """Execute a single step independently. + + Args: + index: Step index (0-based). + """ + if self._is_running: + return + + if index >= len(self._steps): + return + + self._is_running = True + self._run_btn.configure(state="disabled", text="Running...") + self._progress.reset() + + # Reset status for this step and all after it + for i in range(index, len(self._steps)): + self._step_statuses[i] = "pending" + self._steps[i].set_status("pending") + + self._progress.set_value(0.0) + + # Start background worker thread + self._worker_thread = threading.Thread( + target=self._run_steps_worker, + daemon=True, + args=(index,), + ) + self._worker_thread.start() + + def _set_step_status(self, index: int, status: str) -> None: + """Update the status of a step in the UI. + + Args: + index: Step index (0-based). + status: New status string. + """ + self._step_statuses[index] = status + self._steps[index].set_status(status) + + def _run_all_steps(self) -> None: + """Execute all workflow steps sequentially in a background thread.""" + if self._is_running: + return + + self._is_running = True + self._run_btn.configure(state="disabled", text="Running...") + self._progress.reset() + + # Reset all step statuses + for i in range(len(self._steps)): + self._step_statuses[i] = "pending" + self._steps[i].set_status("pending") + + # Start background worker thread + self._worker_thread = threading.Thread( + target=self._run_steps_worker, + daemon=True, + args=(0,), # Start from step 0 + ) + self._worker_thread.start() + + def _run_steps_worker(self, start_index: int = 0) -> None: + """Worker method that runs workflow steps in the background. + + Args: + start_index: Index of the first step to run (0-based). + """ + step_funcs = [ + self._run_step_1_check_environment, + self._run_step_2_program_flash, + self._run_step_3_load_bootloader, + self._run_step_4_load_main_program, + ] + + total = len(step_funcs) - start_index + results: list[bool] = [] + + for i in range(start_index, len(step_funcs)): + step_idx = i - start_index + self._set_step_status(i, "running") + self._progress.set_value(step_idx / total) + + try: + success = step_funcs[i]() + results.append(success) + self._set_step_status(i, "complete" if success else "error") + except Exception as e: + results.append(False) + self._set_step_status(i, "error") + self._log_message(f" Exception in step {i+1}: {e}") + + self._progress.set_value((step_idx + 1) / total) + + # Stop on first failure when running all steps + if start_index == 0 and not success: + break + + self._progress.set_value(1.0) + self._is_running = False + self._run_btn.configure(state="normal", text="▶ Run All Steps") + + # Update dependency statuses + self._update_step_dependencies() + + # Set final status message + if results: + all_ok = all(results) + any_fail = any(not r for r in results) + if all_ok: + self._status.set_status("Workflow complete — all steps passed", "success") + elif any_fail: + failed_count = results.count(False) + self._status.set_status( + f"Workflow complete — {failed_count}/{len(results)} step(s) failed", + "warning", + ) + else: + self._status.set_status("Workflow complete", "info") + else: + self._status.set_status("Workflow complete", "info") + + # Save config after running + self._save_config() + + # ── Serial ───────────────────────────────────────────────── + + def _refresh_ports(self) -> None: + """Refresh the serial port list.""" + ports = detect_serial_ports() + self._port_menu.configure(values=[""] + [p.device for p in ports]) + self._log_message(f"Found {len(ports)} serial port(s)") + + def _refresh_ports_initial(self) -> None: + """Refresh the serial port list on startup (silent, no log).""" + ports = detect_serial_ports() + self._port_menu.configure(values=[""] + [p.device for p in ports]) + # Auto-select if only one port found and config has one + if ports and self._config and self._config.serial_port: + devices = [p.device for p in ports] + if self._config.serial_port in devices: + self._port_var.set(self._config.serial_port) + + def _on_close(self) -> None: + """Handle window close event: save config and cleanup.""" + if self._config: + try: + self._config.save() + except Exception: + pass + self.destroy() + + def _read_serial(self) -> None: + """Read and display serial output.""" + port = self._port_var.get() + if not port: + self._log_message("No serial port selected") + return + + self._log_message(f"Reading from {port}...") + try: + import serial + + with serial.Serial(port, self._config.serial_baudrate if self._config else 115200, timeout=5) as ser: + ser.reset_input_buffer() + lines = [] + while ser.in_waiting: + line = ser.readline().decode("utf-8", errors="replace").strip() + if line: + lines.append(line) + + if lines: + info = parse_boot_output(lines) + self._log_message(f" Parsed: IP={info.ip_address}, Ver={info.version}") + for line in lines[-10:]: + self._log_message(f" > {line}") + else: + self._log_message(" No data received") + except Exception as e: + self._log_message(f" Error: {e}") + + # ── Vitis Check ──────────────────────────────────────────── + + def _check_vitis(self) -> None: + """Check Vitis/Vivado on startup.""" + if not self._config: + return + + result = check_vitis(self._config) + status_dict = get_tool_status_dict(result) + + if result.is_ready: + self._vitis_status.configure(text="Vitis: Ready", text_color=SUCCESS_COLOR) + self._log_message("Vitis/Vivado tools available") + else: + self._vitis_status.configure(text="Vitis: Partial", text_color=WARNING_COLOR) + self._log_message(f"Vitis issues: {'; '.join(result.errors)}") + + +# ── Entry Point ──────────────────────────────────────────────── + +def main() -> None: + """Launch the Zynq Flasher GUI application.""" + app = MainWindow() + app.mainloop() + + +if __name__ == "__main__": + main() diff --git a/src/gui/styles.py b/src/gui/styles.py new file mode 100644 index 0000000..bf0b7c9 --- /dev/null +++ b/src/gui/styles.py @@ -0,0 +1,116 @@ +"""Theme and styling constants for the CustomTkinter GUI. + +Provides consistent cross-platform styling with dark mode detection. +""" + +from __future__ import annotations + +import darkdetect + + +# ── Color Theme ────────────────────────────────────────────── + +PRIMARY_COLOR: str = "#0066CC" +SECONDARY_COLOR: str = "#00A86B" +DANGER_COLOR: str = "#CC3333" +WARNING_COLOR: str = "#E6A817" +SUCCESS_COLOR: str = "#00A86B" +INFO_COLOR: str = "#0066CC" + +DARK_BG: str = "#2B2B2B" +DARK_FG: str = "#E0E0E0" +DARK_CARD: str = "#3A3A3A" +DARK_BORDER: str = "#555555" + +LIGHT_BG: str = "#F5F5F5" +LIGHT_FG: str = "#333333" +LIGHT_CARD: str = "#FFFFFF" +LIGHT_BORDER: str = "#DDDDDD" + + +def get_theme() -> dict: + """Get the current theme configuration. + + Auto-detects dark/light mode based on the system setting. + + Returns: + Dictionary with theme color values. + """ + is_dark = darkdetect.isDark() + + return { + "bg": DARK_BG if is_dark else LIGHT_BG, + "fg": DARK_FG if is_dark else LIGHT_FG, + "card": DARK_CARD if is_dark else LIGHT_CARD, + "border": DARK_BORDER if is_dark else LIGHT_BORDER, + "primary": PRIMARY_COLOR, + "secondary": SECONDARY_COLOR, + "danger": DANGER_COLOR, + "warning": WARNING_COLOR, + "success": SUCCESS_COLOR, + "info": INFO_COLOR, + "is_dark": is_dark, + } + + +# ── Typography ─────────────────────────────────────────────── + +FONT_TITLE: tuple[str, int, str] = ("Segoe UI", 18, "bold") +FONT_HEADING: tuple[str, int, str] = ("Segoe UI", 14, "bold") +FONT_BODY: tuple[str, int, str] = ("Segoe UI", 11) +FONT_SMALL: tuple[str, int, str] = ("Segoe UI", 9) +FONT_MONO: tuple[str, int, str] = ("Consolas", 10) + + +# ── Spacing ────────────────────────────────────────────────── + +PADDING: int = 12 +PADDING_LARGE: int = 20 +PADDING_SMALL: int = 6 +CORNER_RADIUS: int = 8 +BUTTON_HEIGHT: int = 32 + + +# ── Layout ─────────────────────────────────────────────────── + +WINDOW_WIDTH: int = 900 +WINDOW_HEIGHT: int = 700 +WINDOW_TITLE: str = "Zynq XC7Z100 Flasher" + +STEP_HEADER_HEIGHT: int = 36 +LOG_HEIGHT: int = 200 + +# ── Step UI Enhancements ───────────────────────────────────── + +# Accent bar colors for step states +ACCENT_PENDING: str = "#666666" +ACCENT_RUNNING: str = PRIMARY_COLOR +ACCENT_COMPLETE: str = SUCCESS_COLOR +ACCENT_ERROR: str = DANGER_COLOR +ACCENT_DISABLED: str = "#444444" + +# Connector line color +CONNECTOR_COLOR: str = "#555555" +CONNECTOR_COLOR_LIGHT: str = "#CCCCCC" + +# Button hover states +STEP_RUN_BG: str = "#0055AA" +STEP_RUN_BG_HOVER: str = "#0077DD" +STEP_RUNNER_BG: str = "#555555" +STEP_RUNNER_BG_HOVER: str = "#777777" + +# Animation +PULSE_DURATION: int = 500 # ms between pulse flashes +RUNNING_FADE_LIGHT: str = "#0066CC" +RUNNING_FADE_DARK: str = "#003366" + +# Step card styling +STEP_CARD_BG: str = "#333333" +STEP_CARD_BG_HOVER: str = "#3D3D3D" +STEP_CARD_BG_LIGHT: str = "#FFFFFF" +STEP_CARD_BG_HOVER_LIGHT: str = "#F0F0F0" +STEP_DISABLED_ALPHA: float = 0.5 + +# Icon sizes +STEP_ICON_SIZE: int = 28 +STEP_BUTTON_SIZE: int = 28 diff --git a/src/gui/widgets.py b/src/gui/widgets.py new file mode 100644 index 0000000..29e95ae --- /dev/null +++ b/src/gui/widgets.py @@ -0,0 +1,656 @@ +"""Custom widget components for the Zynq Flasher GUI. + +Provides reusable UI components built on CustomTkinter with +consistent styling from gui/styles.py. +""" + +from __future__ import annotations + +import customtkinter as ctk +from typing import Callable + +from gui.styles import ( + FONT_BODY, + FONT_MONO, + FONT_SMALL, + PADDING, + PADDING_LARGE, + PADDING_SMALL, + CORNER_RADIUS, + PRIMARY_COLOR, + SUCCESS_COLOR, + DANGER_COLOR, + WARNING_COLOR, + INFO_COLOR, + ACCENT_PENDING, + ACCENT_RUNNING, + ACCENT_COMPLETE, + ACCENT_ERROR, + ACCENT_DISABLED, + CONNECTOR_COLOR, + DARK_FG, + LIGHT_FG, + STEP_RUN_BG, + STEP_RUN_BG_HOVER, + STEP_RUNNER_BG, + STEP_RUNNER_BG_HOVER, + RUNNING_FADE_LIGHT, + RUNNING_FADE_DARK, + PULSE_DURATION, + STEP_ICON_SIZE, + STEP_BUTTON_SIZE, + STEP_CARD_BG, + STEP_CARD_BG_HOVER, + STEP_DISABLED_ALPHA, +) + + +def _is_dark() -> bool: + """Check if dark mode is active. + + Returns: + True if dark mode is detected. + """ + try: + import darkdetect + return darkdetect.isDark() + except ImportError: + return True + + +class StepHeader(ctk.CTkFrame): + """Enhanced step header widget with run/re-run buttons and visual feedback. + + Displays a numbered step with title, status, accent bar, connector line, + and per-step action buttons (run/re-run). + """ + + def __init__( + self, + master, + step_number: int, + title: str, + status: str = "pending", + description: str = "", + callback: Callable[[], None] | None = None, + can_run: bool = True, + **kwargs, + ): + """Initialize the step header. + + Args: + master: Parent widget. + step_number: Step number (1-based). + title: Step title text. + status: Status indicator ('pending', 'running', 'complete', 'error', 'disabled'). + description: Secondary description text shown below the title. + callback: Called when the run/re-run button is clicked. + can_run: Whether this step can be executed independently. + """ + super().__init__( + master, + corner_radius=CORNER_RADIUS, + fg_color=STEP_CARD_BG, + **kwargs, + ) + + self._step_number = step_number + self._title = title + self._status = status + self._description = description + self._callback = callback + self._can_run = can_run + self._pulse_var = False + self._pulse_job: str | None = None + + self._create_widgets() + self._set_status(status) + + def _create_widgets(self) -> None: + """Create the step header UI components.""" + self.grid_rowconfigure(0, weight=1) + self.grid_columnconfigure((0, 1, 2, 3), weight=0, minsize=PADDING_SMALL) + self.grid_columnconfigure(1, weight=1) + + # Accent bar (left edge) + self._accent_bar = ctk.CTkFrame( + self, + width=4, + height=40, + corner_radius=2, + fg_color=ACCENT_PENDING, + ) + self._accent_bar.grid( + row=0, column=0, + sticky="n", + pady=PADDING, + ) + + # Step number circle + self._num_label = ctk.CTkLabel( + self, + text=str(self._step_number), + font=("Segoe UI", 13, "bold"), + width=STEP_ICON_SIZE, + height=STEP_ICON_SIZE, + corner_radius=STEP_ICON_SIZE // 2, + fg_color="transparent", + text_color=DARK_FG if _is_dark() else LIGHT_FG, + ) + self._num_label.grid( + row=0, column=1, + sticky="n", + pady=PADDING, + ) + + # Content area (title + description) + content_frame = ctk.CTkFrame(self, fg_color="transparent") + content_frame.grid( + row=0, column=2, + sticky="nsew", + pady=PADDING, + padx=PADDING_SMALL, + ) + content_frame.grid_rowconfigure(0, weight=1) + content_frame.grid_columnconfigure(0, weight=1) + + self._title_label = ctk.CTkLabel( + content_frame, + text=self._title, + font=("Segoe UI", 12, "bold"), + anchor="w", + text_color=DARK_FG if _is_dark() else LIGHT_FG, + ) + self._title_label.grid( + row=0, column=0, + sticky="nw", + pady=(0, 2), + ) + + self._desc_label = ctk.CTkLabel( + content_frame, + text=self._description, + font=FONT_SMALL, + anchor="w", + text_color="#999999", + wraplength=400, + ) + self._desc_label.grid( + row=0, column=0, + sticky="nw", + pady=(0, 0), + ) + + # Action buttons area + btn_frame = ctk.CTkFrame(self, fg_color="transparent") + btn_frame.grid( + row=0, column=3, + sticky="n", + pady=PADDING, + ) + btn_frame.grid_rowconfigure(0, weight=1) + btn_frame.grid_columnconfigure(0, weight=1) + + # Run button (▶) + self._run_btn = ctk.CTkButton( + btn_frame, + text="▶", + font=("Segoe UI", 12, "bold"), + width=STEP_BUTTON_SIZE, + height=STEP_BUTTON_SIZE, + corner_radius=6, + fg_color=STEP_RUN_BG, + hover_color=STEP_RUN_BG_HOVER, + text_color="white", + command=self._on_run, + state="disabled", + ) + self._run_btn.grid(row=0, column=0, padx=(0, PADDING_SMALL)) + + # Re-run button (↻) — hidden by default + self._rerun_btn = ctk.CTkButton( + btn_frame, + text="↻", + font=("Segoe UI", 12, "bold"), + width=STEP_BUTTON_SIZE, + height=STEP_BUTTON_SIZE, + corner_radius=6, + fg_color=STEP_RUNNER_BG, + hover_color=STEP_RUNNER_BG_HOVER, + text_color="white", + command=self._on_rerun, + state="hidden", + ) + self._rerun_btn.grid(row=0, column=0, padx=(0, PADDING_SMALL)) + + # Status icon + self._status_label = ctk.CTkLabel( + btn_frame, + text="", + font=("Segoe UI", 14), + anchor="e", + ) + self._status_label.grid(row=0, column=0) + + # Hover effect bindings + self.bind("", self._on_hover_enter) + self.bind("", self._on_hover_leave) + + def _on_hover_enter(self, event) -> None: + """Handle mouse enter for hover effect.""" + if self._status not in ("running", "disabled"): + self.configure(fg_color=STEP_CARD_BG_HOVER) + + def _on_hover_leave(self, event) -> None: + """Handle mouse leave for hover effect.""" + if self._status not in ("running", "disabled"): + self.configure(fg_color=STEP_CARD_BG) + + def _on_run(self) -> None: + """Handle run button click.""" + if self._callback: + self._callback() + + def _on_rerun(self) -> None: + """Handle re-run button click.""" + if self._callback: + self._callback() + + def _set_status(self, status: str) -> None: + """Set the step status with appropriate colors and icons. + + Args: + status: One of 'pending', 'running', 'complete', 'error', 'disabled'. + """ + self._status = status + + # Accent bar color + accent_colors = { + "pending": ACCENT_PENDING, + "running": RUNNING_FADE_LIGHT, + "complete": ACCENT_COMPLETE, + "error": ACCENT_ERROR, + "disabled": ACCENT_DISABLED, + } + self._accent_bar.configure(fg_color=accent_colors.get(status, ACCENT_PENDING)) + + # Number circle + circle_colors = { + "pending": "#666666", + "running": RUNNING_FADE_LIGHT, + "complete": ACCENT_COMPLETE, + "error": ACCENT_ERROR, + "disabled": ACCENT_DISABLED, + } + circle_color = circle_colors.get(status, "#666666") + self._num_label.configure(fg_color=circle_color) + + # Status icon + icons = { + "pending": "", + "running": "◌", + "complete": "✓", + "error": "✗", + "disabled": "", + } + icon = icons.get(status, "") + self._status_label.configure(text=icon) + icon_colors = { + "pending": "#666666", + "running": RUNNING_FADE_LIGHT, + "complete": ACCENT_COMPLETE, + "error": ACCENT_ERROR, + "disabled": ACCENT_DISABLED, + } + self._status_label.configure(text_color=icon_colors.get(status, "#666666")) + + # Buttons visibility + if status == "running": + self._run_btn.configure(state="hidden") + self._rerun_btn.configure(state="hidden") + self._start_pulse() + elif status == "disabled": + self._run_btn.configure(state="hidden") + self._rerun_btn.configure(state="hidden") + self._stop_pulse() + self.configure(fg_color="#2A2A2A") + self._num_label.configure(fg_color=ACCENT_DISABLED) + self._title_label.configure(text_color="#777777") + self._desc_label.configure(text_color="#666666") + elif status == "complete": + self._run_btn.configure(state="hidden") + self._rerun_btn.configure(state="normal") + self._stop_pulse() + elif status == "error": + self._run_btn.configure(state="hidden") + self._rerun_btn.configure(state="normal") + self._stop_pulse() + else: # pending + if self._can_run: + self._run_btn.configure(state="normal") + else: + self._run_btn.configure(state="disabled") + self._rerun_btn.configure(state="hidden") + self._stop_pulse() + + # Update description + desc_map = { + "pending": self._description or "Not started", + "running": "Running...", + "complete": "Completed successfully", + "error": "Failed — click ↻ to retry", + "disabled": "Waiting for dependencies", + } + self._desc_label.configure(text=desc_map.get(status, self._description)) + + def set_status(self, status: str) -> None: + """Public method to update step status. + + Args: + status: New status string. + """ + self.after(0, lambda: self._set_status(status)) + + def set_description(self, description: str) -> None: + """Update the step description text. + + Args: + description: New description text. + """ + self._description = description + if self._status == "pending": + self._desc_label.configure(text=description or "Not started") + + def set_callback(self, callback: Callable[[], None] | None) -> None: + """Update the callback for run/re-run buttons. + + Args: + callback: New callback function. + """ + self._callback = callback + + def set_can_run(self, can_run: bool) -> None: + """Update whether this step can be run independently. + + Args: + can_run: Whether the step is runnable. + """ + self._can_run = can_run + if self._status == "pending": + self._run_btn.configure(state="normal" if can_run else "disabled") + + def _start_pulse(self) -> None: + """Start the pulsing animation for running state.""" + self._stop_pulse() + self._pulse_var = not self._pulse_var + color = RUNNING_FADE_LIGHT if self._pulse_var else RUNNING_FADE_DARK + self._accent_bar.configure(fg_color=color) + self._num_label.configure(fg_color=color) + self._pulse_job = self.after(PULSE_DURATION, self._start_pulse) + + def _stop_pulse(self) -> None: + """Stop the pulsing animation.""" + if self._pulse_job: + self.after_cancel(self._pulse_job) + self._pulse_job = None + + +class ProgressIndicator(ctk.CTkFrame): + """Progress indicator widget for workflow steps. + + Shows a progress bar and percentage for long-running operations. + """ + + def __init__(self, master, **kwargs): + """Initialize the progress indicator. + + Args: + master: Parent widget. + """ + super().__init__(master, **kwargs) + + self._progress_var = ctk.DoubleVar(value=0) + self._create_widgets() + + def _create_widgets(self) -> None: + """Create the progress indicator UI.""" + self.grid_rowconfigure(0, 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 + ) + + self._percent_label = ctk.CTkLabel( + self, + text="0%", + font=FONT_SMALL, + ) + self._percent_label.grid( + row=0, column=0, sticky="ne", padx=PADDING, pady=PADDING_SMALL + ) + + def set_value(self, value: float) -> None: + """Set the progress bar value. + + Args: + value: Progress value between 0.0 and 1.0. + """ + value = max(0.0, min(1.0, value)) + self._progress_var.set(value) + self._percent_label.configure(text=f"{int(value * 100)}%") + + def reset(self) -> None: + """Reset the progress indicator to 0%.""" + self.set_value(0.0) + + +class StatusDisplay(ctk.CTkFrame): + """Status display widget for operation results. + + Shows a status message with color-coded feedback. + """ + + def __init__(self, master, **kwargs): + """Initialize the status display. + + Args: + master: Parent widget. + """ + super().__init__(master, **kwargs) + + self._create_widgets() + + def _create_widgets(self) -> None: + """Create the status display UI.""" + self._status_label = ctk.CTkLabel( + self, + text="Ready", + font=FONT_BODY, + anchor="w", + ) + self._status_label.pack( + fill="x", padx=PADDING, pady=PADDING_SMALL + ) + + def set_status(self, message: str, status: str = "info") -> None: + """Update the status message and color. + + Args: + message: Status message text. + status: Status type for coloring ('info', 'success', 'error', 'warning'). + """ + self._status_label.configure(text=message) + colors = { + "info": INFO_COLOR, + "success": SUCCESS_COLOR, + "error": DANGER_COLOR, + "warning": WARNING_COLOR, + } + self._status_label.configure(text_color=colors.get(status, INFO_COLOR)) + + +class FileSelector(ctk.CTkFrame): + """File selector widget with browse button and path display. + + Allows users to browse for files and displays the selected path. + """ + + def __init__( + self, + master, + label_text: str, + file_types: list[tuple[str, str]] | None = None, + callback: Callable[[str], None] | None = None, + **kwargs, + ): + """Initialize the file selector. + + Args: + master: Parent widget. + label_text: Label text for the selector. + file_types: File dialog filter types (e.g., [("BIN files", "*.bin")]). + callback: Optional callback when file is selected. + """ + super().__init__(master, **kwargs) + + self._callback = callback + self._file_types = file_types or [("All files", "*")] + self._selected_path = ctk.StringVar(value="") + + self._create_widgets(label_text) + + def _create_widgets(self, label_text: str) -> None: + """Create the file selector UI.""" + self.grid_rowconfigure(0, weight=1) + self.grid_columnconfigure(1, weight=1) + + # Label + label = ctk.CTkLabel( + self, + text=label_text, + font=FONT_SMALL, + anchor="w", + ) + label.grid(row=0, column=0, sticky="w", padx=(PADDING, PADDING_SMALL)) + + # Path entry + self._entry = ctk.CTkEntry( + self, + textvariable=self._selected_path, + font=FONT_MONO, + state="readonly", + ) + self._entry.grid( + row=0, column=1, sticky="nsew", padx=PADDING_SMALL + ) + + # Browse button + browse_btn = ctk.CTkButton( + self, + text="Browse", + font=FONT_SMALL, + command=self._browse, + width=80, + ) + browse_btn.grid( + row=0, column=2, padx=(PADDING_SMALL, PADDING) + ) + + def _browse(self) -> None: + """Open file dialog and set selected path.""" + import customtkinter + from tkinter import filedialog + + file_path = filedialog.askopenfilename( + title=f"Select {self._selected_path.get() or 'file'}", + filetypes=self._file_types, + ) + if file_path: + self.set_path(file_path) + + def set_path(self, path: str) -> None: + """Set the selected file path. + + Args: + path: Absolute path to the selected file. + """ + self._selected_path.set(path) + if self._callback: + self._callback(path) + + def get_path(self) -> str: + """Get the currently selected file path. + + Returns: + Selected file path string. + """ + return self._selected_path.get() + + +class LogDisplay(ctk.CTkFrame): + """Log display widget for operation output. + + Shows a scrollable text area with monospace font for log output. + """ + + def __init__(self, master, height: int = 200, **kwargs): + """Initialize the log display. + + Args: + master: Parent widget. + height: Height of the log display in pixels. + """ + super().__init__(master, **kwargs) + + self._create_widgets(height) + + def _create_widgets(self, height: int) -> None: + """Create the log display UI.""" + self.grid_rowconfigure(0, weight=1) + self.grid_columnconfigure(0, weight=1) + + self._text_widget = ctk.CTkTextbox( + self, + height=height, + font=FONT_MONO, + corner_radius=CORNER_RADIUS, + state="disabled", + ) + self._text_widget.grid( + row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING + ) + + def append(self, message: str) -> None: + """Append a message to the log display. + + Args: + message: Text to append. + """ + self._text_widget.configure(state="normal") + self._text_widget.insert("end", message + "\n") + self._text_widget.see("end") + self._text_widget.configure(state="disabled") + + def clear(self) -> None: + """Clear the log display.""" + self._text_widget.configure(state="normal") + self._text_widget.delete("1.0", "end") + self._text_widget.configure(state="disabled") + + def set_text(self, text: str) -> None: + """Replace all text in the log display. + + Args: + text: New text content. + """ + self._text_widget.configure(state="normal") + self._text_widget.delete("1.0", "end") + self._text_widget.insert("1.0", text) + self._text_widget.configure(state="disabled") diff --git a/src/ip_verifier.py b/src/ip_verifier.py new file mode 100644 index 0000000..208c173 --- /dev/null +++ b/src/ip_verifier.py @@ -0,0 +1,107 @@ +"""IP connectivity verification for Zynq Flasher GUI. + +Pings the Zynq device IP to verify network connectivity. +""" + +from __future__ import annotations + +import platform +import subprocess + + +def ping_ip( + ip_address: str, + count: int = 3, + timeout: int = 5, +) -> bool: + """Ping an IP address to verify connectivity. + + Uses platform-appropriate ping command: + - Linux/macOS: ping -c -W + - Windows: ping -n -w + + Args: + ip_address: IP address to ping (e.g., '192.168.100.11'). + count: Number of ping packets to send. + timeout: Timeout in seconds (Linux) or milliseconds (Windows). + + Returns: + True if all ping packets were received, False otherwise. + """ + system = platform.system().lower() + + if system == "windows": + cmd = ["ping", "-n", str(count), "-w", str(timeout * 1000), ip_address] + else: + cmd = ["ping", "-c", str(count), "-W", str(timeout), ip_address] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout * count + 10, + ) + return result.returncode == 0 + except (subprocess.TimeoutExpired, FileNotFoundError, OSError): + return False + + +def ping_ip_verbose( + ip_address: str, + count: int = 3, + timeout: int = 5, +) -> dict: + """Ping an IP address and return detailed results. + + Args: + ip_address: IP address to ping. + count: Number of ping packets. + timeout: Timeout per packet. + + Returns: + Dictionary with 'success' (bool), 'output' (str), and 'errors' (list). + """ + system = platform.system().lower() + + if system == "windows": + cmd = ["ping", "-n", str(count), "-w", str(timeout * 1000), ip_address] + else: + cmd = ["ping", "-c", str(count), "-W", str(timeout), ip_address] + + output = "" + errors: list[str] = [] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout * count + 10, + ) + output = result.stdout.strip() + if result.stderr: + errors.append(result.stderr.strip()) + return { + "success": result.returncode == 0, + "output": output, + "errors": errors, + } + except subprocess.TimeoutExpired as e: + return { + "success": False, + "output": e.stdout if e.stdout else "", + "errors": ["Ping timed out"], + } + except FileNotFoundError as e: + return { + "success": False, + "output": "", + "errors": [f"ping command not found: {e}"], + } + except OSError as e: + return { + "success": False, + "output": "", + "errors": [f"OS error: {e}"], + } diff --git a/src/reboot_manager.py b/src/reboot_manager.py new file mode 100644 index 0000000..76d8eaf --- /dev/null +++ b/src/reboot_manager.py @@ -0,0 +1,213 @@ +"""Reboot manager for Zynq Flasher GUI. + +Handles rebooting the Zynq device via SSH or serial commands. +""" + +from __future__ import annotations + +import importlib +import platform +import shutil +import subprocess +import time +from dataclasses import dataclass +from typing import Callable + +from config_manager import Config + + +@dataclass +class RebootResult: + """Result of a reboot operation.""" + + success: bool + message: str + method: str = "" + output: str = "" + + +ProgressCallback = Callable[[str, str], None] | None + + +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. + + Args: + config: Application configuration. + timeout: Seconds to wait for reboot acknowledgment. + callback: Optional progress callback. + + Returns: + RebootResult with status. + """ + 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()}") + + # 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", + ) + except OSError as e: + return RebootResult( + success=False, + message=f"OS error: {e}", + method="ssh", + ) + + +def reboot_via_serial( + config: Config, + callback: ProgressCallback = None, +) -> RebootResult: + """Reboot the Zynq device via serial port. + + Sends a reboot command through the serial console. + + Args: + config: Application configuration. + callback: Optional progress callback. + + Returns: + RebootResult with status. + """ + if not config.serial_port: + return RebootResult( + success=False, + message="No serial port configured", + method="serial", + ) + + if callback: + callback("start", f"Sending reboot via serial {config.serial_port}...") + + try: + import serial + + with serial.Serial( + config.serial_port, + config.serial_baudrate, + timeout=5, + ) as ser: + ser.reset_input_buffer() + # 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") + ser.reset_input_buffer() + + if callback: + callback("progress", "Reboot command sent via serial") + + return RebootResult( + success=True, + message="Reboot command sent via serial", + method="serial", + ) + except ImportError: + return RebootResult( + success=False, + message="pyserial not installed", + method="serial", + ) + except serial.SerialException as e: + return RebootResult( + success=False, + message=f"Serial error: {e}", + method="serial", + ) + except OSError as e: + return RebootResult( + success=False, + message=f"OS error: {e}", + method="serial", + ) + + +def reboot_zynq( + config: Config, + method: str = "auto", + timeout: int | None = None, + callback: ProgressCallback = None, +) -> RebootResult: + """Reboot the Zynq device using the best available method. + + Tries SSH first, falls back to serial if SSH is unavailable. + + Args: + config: Application configuration. + method: Reboot method: 'auto', 'ssh', or 'serial'. + timeout: Seconds to wait for reboot. + callback: Optional progress callback. + + Returns: + RebootResult with status. + """ + timeout = timeout or config.reboot_timeout + + if method == "ssh": + return reboot_via_ssh(config, timeout, callback) + elif method == "serial": + return reboot_via_serial(config, callback) + + # Auto: try SSH first, then serial + result = reboot_via_ssh(config, timeout, callback) + if result.success: + return result + + if callback: + callback("progress", "SSH failed, trying serial...") + + return reboot_via_serial(config, callback) diff --git a/src/serial_monitor.py b/src/serial_monitor.py new file mode 100644 index 0000000..5922b71 --- /dev/null +++ b/src/serial_monitor.py @@ -0,0 +1,144 @@ +"""Serial port monitoring and log parsing for Zynq Flasher GUI. + +Detects available serial ports, reads output streams, and parses +boot logs for IP addresses, version strings, and other diagnostic info. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Callable + +import serial +import serial.tools.list_ports + + +@dataclass +class BootInfo: + """Parsed information extracted from Zynq boot output.""" + + ip_address: str = "" + version: str = "" + boot_message: str = "" + raw_lines: list[str] | None = None + + +@dataclass +class SerialInfo: + """Information about an available serial port.""" + + device: str + description: str = "" + hwid: str = "" + + +def detect_serial_ports() -> list[SerialInfo]: + """Detect all available serial/USB ports on the system. + + Returns: + List of SerialInfo objects for each available port. + """ + ports = serial.tools.list_ports.comports() + results: list[SerialInfo] = [] + for port in ports: + results.append( + SerialInfo( + device=port.device, + description=port.description or "", + hwid=port.hwid or "", + ) + ) + return results + + +def parse_boot_output(lines: list[str]) -> BootInfo: + """Parse boot output lines for IP, version, and other info. + + Looks for common patterns in Zynq boot logs: + - IP addresses (IPv4) + - Version strings (e.g., "v1.2.3", "version: x.y.z") + - Boot completion messages + + Args: + lines: List of log output lines to parse. + + Returns: + BootInfo with extracted data. + """ + info = BootInfo(raw_lines=lines) + + ip_pattern = re.compile(r"\b(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\b") + version_patterns = [ + re.compile(r"[Vv]ersion[:\s]+([^\s;,\n]+)", re.IGNORECASE), + re.compile(r"[Bb]oot\s+([^\s;,\n]+)", re.IGNORECASE), + re.compile(r"[Ff]irmware\s+([^\s;,\n]+)", re.IGNORECASE), + re.compile(r"(?:^|[\s:=])(v\d+\.\d+\.\d+)(?:\s|$)", re.IGNORECASE), + ] + boot_complete_patterns = [ + re.compile(r"boot\s+complete", re.IGNORECASE), + re.compile(r"system\s+ready", re.IGNORECASE), + re.compile(r"starting\s+application", re.IGNORECASE), + re.compile(r"Linux\s+booted", re.IGNORECASE), + re.compile(r"QSYS\s+ready", re.IGNORECASE), + ] + + for line in lines: + # Extract IP address + if not info.ip_address: + match = ip_pattern.search(line) + if match: + info.ip_address = match.group(1) + + # Extract version + if not info.version: + for pattern in version_patterns: + match = pattern.search(line) + if match: + info.version = match.group(1) + break + + # Check for boot completion + if not info.boot_message: + for pattern in boot_complete_patterns: + if pattern.search(line): + info.boot_message = line.strip() + break + + return info + + +def read_serial_stream( + port: str, + baudrate: int = 115200, + timeout: float = 1.0, + line_callback: Callable[[str], None] | None = None, +) -> list[str]: + """Read lines from a serial port. + + Opens the serial port, reads available lines, and closes it. + Optionally calls line_callback for each line as it arrives. + + Args: + port: Serial port device path (e.g., '/dev/ttyUSB0'). + baudrate: Baud rate for serial communication. + timeout: Read timeout in seconds. + line_callback: Optional callback invoked for each line. + + Returns: + List of lines read from the serial port. + + Raises: + serial.SerialException: If the port cannot be opened. + """ + lines: list[str] = [] + with serial.Serial(port, baudrate, timeout=timeout) as ser: + # Clear any pending data + ser.reset_input_buffer() + while ser.in_waiting: + line = ser.readline().decode("utf-8", errors="replace").strip() + if line: + lines.append(line) + if line_callback: + line_callback(line) + return lines diff --git a/src/tftp_manager.py b/src/tftp_manager.py new file mode 100644 index 0000000..92b9d62 --- /dev/null +++ b/src/tftp_manager.py @@ -0,0 +1,322 @@ +"""TFTP file transfer manager for Zynq Flasher GUI. + +Handles uploading, downloading, and CRC verification of files +via TFTP to the Zynq device. +""" + +from __future__ import annotations + +import binascii +import os +import platform +import shutil +import subprocess +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Callable + +from config_manager import Config +from ip_verifier import ping_ip + + +@dataclass +class TftpResult: + """Result of a TFTP operation.""" + + step: str + success: bool + message: str + local_path: str = "" + remote_path: str = "" + crc_local: int = 0 + crc_remote: int = 0 + + +ProgressCallback = Callable[[str, str], None] | None + + +def _compute_crc32(file_path: str | Path) -> int: + """Compute CRC32 checksum of a file. + + Args: + file_path: Path to the file. + + Returns: + CRC32 value as unsigned integer. + """ + import binascii + + crc = 0 + with open(file_path, "rb") as f: + while True: + chunk = f.read(65536) + if not chunk: + break + crc = binascii.crc32(chunk, crc) + return crc & 0xFFFFFFFF + + +def _find_tftp_client() -> str | None: + """Find the tftp client executable. + + Uses shutil.which() for cross-platform compatibility. + + Returns: + Path to tftp executable, or None if not found. + """ + return shutil.which("tftp") + + +def tftp_upload( + config: Config, + file_path: str | Path, + remote_name: str | None = None, + callback: ProgressCallback = None, +) -> TftpResult: + """Upload a file to the Zynq device via TFTP. + + Args: + config: Application configuration. + file_path: Path to the local file to upload. + remote_name: Remote filename (defaults to config's tftp_upload_name). + callback: Optional progress callback. + + Returns: + TftpResult with upload status. + """ + remote_name = remote_name or config.tftp_upload_name + file_path = Path(file_path).resolve() + + if not file_path.exists(): + return TftpResult( + step="upload", + success=False, + message=f"File not found: {file_path}", + local_path=str(file_path), + remote_path=remote_name, + ) + + if callback: + callback("start", f"Uploading {file_path.name} -> {remote_name}") + + # Verify device is reachable first + if not ping_ip(config.zynq_ip, config.ping_count, config.ping_timeout): + return TftpResult( + step="upload", + success=False, + message=f"Device {config.zynq_ip} not reachable", + local_path=str(file_path), + remote_path=remote_name, + ) + + tftp_cmd = _find_tftp_client() + if not tftp_cmd: + return TftpResult( + step="upload", + success=False, + message="tftp client not found", + local_path=str(file_path), + remote_path=remote_name, + ) + + if callback: + callback("progress", f"Running: {tftp_cmd}") + + # Build TFTP command + cmd = [tftp_cmd, "-m", "binary", config.zynq_ip, "put", str(file_path), remote_name] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=60, + ) + success = result.returncode == 0 + output = (result.stdout + result.stderr).strip() + + if callback: + callback("complete" if success else "error", + "Upload " + ("succeeded" if success else "failed")) + + return TftpResult( + step="upload", + success=success, + message=output or ("Upload succeeded" if success else "Upload failed"), + local_path=str(file_path), + remote_path=remote_name, + crc_local=_compute_crc32(file_path), + ) + except subprocess.TimeoutExpired: + if callback: + callback("error", "Upload timed out") + return TftpResult( + step="upload", + success=False, + message="Upload timed out", + local_path=str(file_path), + remote_path=remote_name, + crc_local=_compute_crc32(file_path), + ) + except OSError as e: + return TftpResult( + step="upload", + success=False, + message=f"OS error: {e}", + local_path=str(file_path), + remote_path=remote_name, + ) + + +def tftp_download( + config: Config, + remote_name: str, + local_dir: str | Path | None = None, + callback: ProgressCallback = None, +) -> TftpResult: + """Download a file from the Zynq device via TFTP. + + Args: + config: Application configuration. + remote_name: Remote filename on the TFTP server. + local_dir: Local directory to save the file (default: temp directory). + callback: Optional progress callback. + + Returns: + TftpResult with download status. + """ + local_dir = Path(local_dir or tempfile.gettempdir()) + local_path = local_dir / remote_name + + if callback: + callback("start", f"Downloading {remote_name} -> {local_path}") + + # 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", + local_path=str(local_path), + remote_path=remote_name, + ) + + tftp_cmd = _find_tftp_client() + if not tftp_cmd: + return TftpResult( + step="download", + success=False, + message="tftp client not found", + local_path=str(local_path), + remote_path=remote_name, + ) + + cmd = [tftp_cmd, "-m", "binary", config.zynq_ip, "get", remote_name, str(local_path)] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=60, + ) + success = result.returncode == 0 + output = (result.stdout + result.stderr).strip() + + if callback: + callback("complete" if success else "error", + "Download " + ("succeeded" if success else "failed")) + + crc = _compute_crc32(local_path) if success and local_path.exists() else 0 + + return TftpResult( + step="download", + success=success, + message=output or ("Download succeeded" if success else "Download failed"), + local_path=str(local_path), + remote_path=remote_name, + crc_remote=crc, + ) + except subprocess.TimeoutExpired: + return TftpResult( + step="download", + success=False, + message="Download timed out", + local_path=str(local_path), + remote_path=remote_name, + ) + except OSError as e: + return TftpResult( + step="download", + success=False, + message=f"OS error: {e}", + local_path=str(local_path), + remote_path=remote_name, + ) + + +def tftp_upload_verify( + config: Config, + file_path: str | Path, + remote_name: str | None = None, + callback: ProgressCallback = None, +) -> TftpResult: + """Upload a file via TFTP, download it back, and verify CRC. + + Args: + config: Application configuration. + file_path: Path to the local file to upload. + remote_name: Remote filename (defaults to config's tftp_upload_name). + callback: Optional progress callback. + + Returns: + TftpResult with upload, download, and CRC verification status. + """ + remote_name = remote_name or config.tftp_upload_name + + # Step 1: Upload + upload_result = tftp_upload(config, file_path, remote_name, callback) + if not upload_result.success: + return upload_result + + # Step 2: Download for verification + download_result = tftp_download(config, remote_name, callback=callback) + if not download_result.success: + return download_result + + # Step 3: Compare CRC + original_crc = _compute_crc32(file_path) + downloaded_crc = download_result.crc_remote + + verified = original_crc == downloaded_crc + + if callback: + if verified: + callback("complete", f"CRC verified: {original_crc:#010x} == {downloaded_crc:#010x}") + else: + callback("error", f"CRC mismatch: {original_crc:#010x} != {downloaded_crc:#010x}") + + # Step 4: Clean up downloaded temp file regardless of result + if download_result.local_path: + try: + Path(download_result.local_path).unlink(missing_ok=True) + if callback: + callback("complete", "Temp file cleaned up") + except OSError: + if callback: + callback("progress", f"Temp file not removed: {download_result.local_path}") + + return TftpResult( + step="upload_verify", + success=verified, + message=( + f"CRC verified: {original_crc:#010x}" + if verified + else f"CRC mismatch: {original_crc:#010x} vs {downloaded_crc:#010x}" + ), + local_path=str(file_path), + remote_path=remote_name, + crc_local=original_crc, + crc_remote=downloaded_crc, + ) diff --git a/src/vitis_checker.py b/src/vitis_checker.py new file mode 100644 index 0000000..8294453 --- /dev/null +++ b/src/vitis_checker.py @@ -0,0 +1,170 @@ +"""Vitis/Vivado tool detection for Zynq Flasher GUI. + +Checks for the presence and accessibility of required Xilinx tools +(bootgen, impact, etc.) on the system. +""" + +from __future__ import annotations + +import os +import platform +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from config_manager import Config + + +@dataclass +class ToolStatus: + """Status of a single Vitis/Vivado tool.""" + + name: str + found: bool + path: str = "" + version: str = "" + error: str = "" + + +@dataclass +class VitisCheckResult: + """Result of a complete Vitis/Vivado environment check.""" + + system: str + vitis_path: str + tools: dict[str, ToolStatus] + is_ready: bool + errors: list[str] + + +def check_vitis(config: Config) -> VitisCheckResult: + """Check for Vitis/Vivado tools availability. + + Scans the configured vitis_path and system PATH for required tools: + bootgen (boot image generator), impact (JTAG controller). + + Args: + config: Application configuration with vitis_path setting. + + Returns: + VitisCheckResult with details on each tool's status. + """ + system = platform.system().lower() + vitis_path = config.vitis_path or "" + tools: dict[str, ToolStatus] = {} + errors: list[str] = [] + + required_tools = ["bootgen", "impact"] + + for tool_name in required_tools: + status = _check_tool(tool_name, vitis_path, system) + tools[tool_name] = status + if not status.found: + errors.append(f"{tool_name}: {status.error or 'not found'}") + + return VitisCheckResult( + system=system, + vitis_path=vitis_path, + tools=tools, + is_ready=len(errors) == 0, + errors=errors, + ) + + +def _check_tool( + tool_name: str, vitis_path: str, system: str +) -> ToolStatus: + """Check if a single tool is available. + + Args: + tool_name: Name of the tool to check. + vitis_path: Vitis installation path to search. + system: Platform identifier (linux, windows, darwin). + + Returns: + ToolStatus with detection results. + """ + # Search in vitis_path/bin first, then system PATH + search_dirs: list[str] = [] + if vitis_path: + bin_dir = Path(vitis_path) / "bin" + if bin_dir.exists(): + search_dirs.append(str(bin_dir)) + + # Add system PATH components + path_env = os.environ.get("PATH", "") + search_dirs.extend(path_env.split(os.pathsep)) + + exec_name = f"{tool_name}.exe" if system == "windows" else tool_name + + for search_dir in search_dirs: + candidate = Path(search_dir) / exec_name + if candidate.exists() and candidate.is_file(): + version = _get_tool_version(str(candidate), tool_name, system) + return ToolStatus( + name=tool_name, + found=True, + path=str(candidate), + version=version, + ) + + return ToolStatus( + name=tool_name, + found=False, + error=f"Could not locate {exec_name} in PATH or {vitis_path}/bin", + ) + + +def _get_tool_version(tool_path: str, tool_name: str, system: str) -> str: + """Get version string from a tool executable. + + Args: + tool_path: Absolute path to the tool executable. + tool_name: Name of the tool. + system: Platform identifier. + + Returns: + Version string or empty string if unavailable. + """ + try: + flag = "-h" if tool_name == "impact" else "-h" + result = subprocess.run( + [tool_path, flag], + capture_output=True, + text=True, + timeout=10, + ) + output = (result.stdout + result.stderr).strip() + if output: + first_line = output.split("\n")[0] + return first_line[:120] + except (subprocess.TimeoutExpired, OSError, FileNotFoundError): + pass + return "" + + +def get_tool_status_dict(result: VitisCheckResult) -> dict[str, Any]: + """Convert a VitisCheckResult to a serializable dictionary. + + Args: + result: VitisCheckResult from check_vitis(). + + Returns: + Dictionary suitable for JSON serialization or GUI display. + """ + return { + "system": result.system, + "vitis_path": result.vitis_path, + "is_ready": result.is_ready, + "tools": { + name: { + "found": status.found, + "path": status.path, + "version": status.version, + "error": status.error, + } + for name, status in result.tools.items() + }, + "errors": result.errors, + }