Compare commits
14
Commits
c9636b2fcd
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
97a116b36f | ||
|
|
fce940926f | ||
|
|
c0f7219d30 | ||
|
|
4da03e97d8 | ||
|
|
c9eae220f0 | ||
|
|
e09f16a3aa | ||
|
|
47ea614ba2 | ||
|
|
13ed0d89f8 | ||
|
|
0d66f65ef2 | ||
|
|
4c01acdf69 | ||
|
|
c0cac5b2c5 | ||
|
|
5ed195e2d6 | ||
|
|
639f2e3ce6 | ||
|
|
9502bba636 |
+3
-1
@@ -6,6 +6,7 @@ dist/
|
|||||||
build/
|
build/
|
||||||
*.egg-info/
|
*.egg-info/
|
||||||
*.egg
|
*.egg
|
||||||
|
*.exe
|
||||||
|
|
||||||
# ── Python ───────────────────────────────────────────────────
|
# ── Python ───────────────────────────────────────────────────
|
||||||
__pycache__/
|
__pycache__/
|
||||||
@@ -33,7 +34,8 @@ htmlcov/
|
|||||||
# ── Opencode (internal tooling) ──────────────────────────────
|
# ── Opencode (internal tooling) ──────────────────────────────
|
||||||
.opencode/
|
.opencode/
|
||||||
|
|
||||||
# ── Config overrides (user-specific) ─────────────────────────
|
# ── Config (user-specific, auto-generated if missing) ─────────
|
||||||
|
config.yaml
|
||||||
config/zynq_flasher.yaml
|
config/zynq_flasher.yaml
|
||||||
|
|
||||||
# ── Logs ─────────────────────────────────────────────────────
|
# ── Logs ─────────────────────────────────────────────────────
|
||||||
|
|||||||
@@ -0,0 +1,63 @@
|
|||||||
|
# -*- mode: python ; coding: utf-8 -*-
|
||||||
|
"""PyInstaller spec for Zynq Flasher — Windows x64 single-exe build.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
pyinstaller --clean --noconfirm ZynqFlasher.spec
|
||||||
|
→ output: dist/ZynqFlasher.exe
|
||||||
|
"""
|
||||||
|
|
||||||
|
from PyInstaller.utils.hooks import collect_data_files, collect_submodules
|
||||||
|
|
||||||
|
block_cipher = None
|
||||||
|
|
||||||
|
a = Analysis(
|
||||||
|
['app.py'],
|
||||||
|
pathex=['src'],
|
||||||
|
binaries=[],
|
||||||
|
datas=[
|
||||||
|
# CustomTkinter themes, fonts, icons — auto-collected
|
||||||
|
*collect_data_files('customtkinter'),
|
||||||
|
# Default config shipped as template
|
||||||
|
('config/default_config.yaml', 'config'),
|
||||||
|
],
|
||||||
|
hiddenimports=[
|
||||||
|
*collect_submodules('customtkinter'),
|
||||||
|
'darkdetect',
|
||||||
|
'yaml',
|
||||||
|
'serial',
|
||||||
|
'serial.tools',
|
||||||
|
'serial.tools.list_ports',
|
||||||
|
],
|
||||||
|
hookspath=[],
|
||||||
|
hooksconfig={},
|
||||||
|
runtime_hooks=[],
|
||||||
|
excludes=[],
|
||||||
|
win_no_prefer_redirects=False,
|
||||||
|
win_private_assemblies=False,
|
||||||
|
cipher=block_cipher,
|
||||||
|
noarchive=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
pyz = PYZ(a.pure)
|
||||||
|
|
||||||
|
exe = EXE(
|
||||||
|
pyz,
|
||||||
|
a.scripts,
|
||||||
|
a.binaries,
|
||||||
|
a.zipfiles,
|
||||||
|
a.datas,
|
||||||
|
[],
|
||||||
|
name='ZynqFlasher',
|
||||||
|
debug=False,
|
||||||
|
bootloader_ignore_signals=False,
|
||||||
|
strip=False,
|
||||||
|
upx=True,
|
||||||
|
upx_exclude=[],
|
||||||
|
runtime_tmpdir=None,
|
||||||
|
console=True, # show console for terminal log output
|
||||||
|
disable_windowed_traceback=False,
|
||||||
|
argv_emulation=False,
|
||||||
|
target_arch=None,
|
||||||
|
codesign_identity=None,
|
||||||
|
entitlements_file=None,
|
||||||
|
)
|
||||||
@@ -0,0 +1,47 @@
|
|||||||
|
@echo off
|
||||||
|
setlocal
|
||||||
|
|
||||||
|
echo ============================================================
|
||||||
|
echo Zynq Flasher — Windows x64 Build Script
|
||||||
|
echo ============================================================
|
||||||
|
echo.
|
||||||
|
|
||||||
|
REM ── 1. Check Python ─────────────────────────────────────────
|
||||||
|
python --version >nul 2>&1
|
||||||
|
if %errorlevel% neq 0 (
|
||||||
|
echo [ERROR] Python not found. Install Python 3.9+ first.
|
||||||
|
pause
|
||||||
|
exit /b 1
|
||||||
|
)
|
||||||
|
echo [1/4] Python: OK
|
||||||
|
|
||||||
|
REM ── 2. Install dependencies ─────────────────────────────────
|
||||||
|
echo [2/4] Installing Python packages...
|
||||||
|
pip install -r requirements.txt --quiet
|
||||||
|
pip install pyinstaller --quiet
|
||||||
|
echo Done.
|
||||||
|
|
||||||
|
REM ── 3. Clean previous builds ────────────────────────────────
|
||||||
|
echo [3/4] Cleaning previous builds...
|
||||||
|
if exist build rmdir /s /q build
|
||||||
|
if exist dist\ZynqFlasher.exe del /q dist\ZynqFlasher.exe
|
||||||
|
if not exist dist mkdir dist
|
||||||
|
|
||||||
|
REM ── 4. Build ────────────────────────────────────────────────
|
||||||
|
echo [4/4] Building ZynqFlasher.exe...
|
||||||
|
pyinstaller --clean --noconfirm ZynqFlasher.spec
|
||||||
|
|
||||||
|
echo.
|
||||||
|
if exist dist\ZynqFlasher.exe (
|
||||||
|
echo ============================================================
|
||||||
|
echo Build successful!
|
||||||
|
echo Output: dist\ZynqFlasher.exe
|
||||||
|
echo ============================================================
|
||||||
|
) else (
|
||||||
|
echo [ERROR] Build failed — dist\ZynqFlasher.exe not found.
|
||||||
|
pause
|
||||||
|
exit /b 1
|
||||||
|
)
|
||||||
|
|
||||||
|
endlocal
|
||||||
|
pause
|
||||||
-27
@@ -1,27 +0,0 @@
|
|||||||
boot_wait_delay: 10
|
|
||||||
bootloader_bin_path: /home/ly0kos/work/Verify/FW/tftp_app/bootloader/bootloader_z100_800M.bin
|
|
||||||
bootloader_bit_path: /home/ly0kos/work/Verify/FW/tftp_app/bootloader/pl_arm_wrapper_hw_platform_0/pl_arm_wrapper.bit
|
|
||||||
bootloader_elf_path: /home/ly0kos/work/Verify/FW/tftp_app/tftp_app/tftp.elf
|
|
||||||
erase_all: false
|
|
||||||
firmware_bin_path: /home/ly0kos/work/Verify/FW/V1.3.1.3.3_06_01_pmu_cpld_check/V1.3.1.3.3_06_01_pmu_cpld_check.bin
|
|
||||||
flash_model: s25fl256s1
|
|
||||||
flash_type: qspi-x4-single
|
|
||||||
fsbl_elf_path: /home/ly0kos/work/Verify/FPGA/Xilinx/Z100/fsbl_qspi_bypass.elf
|
|
||||||
inter_step_delay: 2
|
|
||||||
ping_count: 3
|
|
||||||
ping_timeout: 5
|
|
||||||
reboot_timeout: 30
|
|
||||||
serial_baudrate: 115200
|
|
||||||
serial_port: /dev/ttyUSB0
|
|
||||||
step_timeouts:
|
|
||||||
bitstream: 60
|
|
||||||
check_env: 30
|
|
||||||
elf_download: 60
|
|
||||||
flash_erase: 120
|
|
||||||
flash_program: 600
|
|
||||||
tftp_reboot: 120
|
|
||||||
tftp_upload_name: z7bin
|
|
||||||
uart_delay: 3
|
|
||||||
xilinx_path: /data/xilinx
|
|
||||||
zynq_ip: 192.168.100.11
|
|
||||||
zynq_part: XC7Z100
|
|
||||||
@@ -271,12 +271,14 @@ def _program_with_vivado(
|
|||||||
callback("progress", "Using Vivado Hardware Manager...")
|
callback("progress", "Using Vivado Hardware Manager...")
|
||||||
|
|
||||||
tcl_content = f"""
|
tcl_content = f"""
|
||||||
open_hw_manager
|
open_hw
|
||||||
connect_hw_server
|
connect_hw_server
|
||||||
open_hw_target
|
open_hw_target
|
||||||
current_hw_device [lindex [get_hw_devices] 0]
|
current_hw_device [lindex [get_hw_devices] 0]
|
||||||
program_hw_device -file "{bit_path}"
|
program_hw_device -file "{bit_path}"
|
||||||
refresh_hw_device
|
refresh_hw_device
|
||||||
|
close_hw
|
||||||
|
disconnect_hw_server
|
||||||
quit
|
quit
|
||||||
"""
|
"""
|
||||||
tcl_path = bit_path.parent / "temp_program.tcl"
|
tcl_path = bit_path.parent / "temp_program.tcl"
|
||||||
|
|||||||
+39
-4
@@ -28,13 +28,14 @@ DEFAULT_CONFIG: dict[str, Any] = {
|
|||||||
"flash_type": "qspi-x4-single",
|
"flash_type": "qspi-x4-single",
|
||||||
"flash_model": "s25fl256s1",
|
"flash_model": "s25fl256s1",
|
||||||
"erase_all": False,
|
"erase_all": False,
|
||||||
|
"download_verify": True,
|
||||||
"firmware_bin_path": "",
|
"firmware_bin_path": "",
|
||||||
"reboot_timeout": 30,
|
"reboot_timeout": 30,
|
||||||
"ping_timeout": 5,
|
"ping_timeout": 5,
|
||||||
"ping_count": 3,
|
"ping_count": 3,
|
||||||
"uart_delay": 3,
|
"uart_delay": 3,
|
||||||
"inter_step_delay": 2,
|
"inter_step_delay": 2,
|
||||||
"boot_wait_delay": 10,
|
"boot_wait_delay": 30,
|
||||||
"step_timeouts": {
|
"step_timeouts": {
|
||||||
"check_env": 120,
|
"check_env": 120,
|
||||||
"flash_erase": 120,
|
"flash_erase": 120,
|
||||||
@@ -67,13 +68,14 @@ class Config:
|
|||||||
flash_type: str = "qspi-x4-single"
|
flash_type: str = "qspi-x4-single"
|
||||||
flash_model: str = "s25fl256s1"
|
flash_model: str = "s25fl256s1"
|
||||||
erase_all: bool = False
|
erase_all: bool = False
|
||||||
|
download_verify: bool = True
|
||||||
firmware_bin_path: str = ""
|
firmware_bin_path: str = ""
|
||||||
reboot_timeout: int = 30
|
reboot_timeout: int = 30
|
||||||
ping_timeout: int = 5
|
ping_timeout: int = 5
|
||||||
ping_count: int = 3
|
ping_count: int = 3
|
||||||
uart_delay: int = 3
|
uart_delay: int = 3
|
||||||
inter_step_delay: int = 2
|
inter_step_delay: int = 2
|
||||||
boot_wait_delay: int = 10
|
boot_wait_delay: int = 30
|
||||||
step_timeouts: dict[str, int] = field(default_factory=lambda: {})
|
step_timeouts: dict[str, int] = field(default_factory=lambda: {})
|
||||||
|
|
||||||
# Internal: path to the config file on disk
|
# Internal: path to the config file on disk
|
||||||
@@ -176,6 +178,7 @@ class Config:
|
|||||||
"flash_type": self.flash_type,
|
"flash_type": self.flash_type,
|
||||||
"flash_model": self.flash_model,
|
"flash_model": self.flash_model,
|
||||||
"erase_all": self.erase_all,
|
"erase_all": self.erase_all,
|
||||||
|
"download_verify": self.download_verify,
|
||||||
"firmware_bin_path": self._relative_path(self.firmware_bin_path),
|
"firmware_bin_path": self._relative_path(self.firmware_bin_path),
|
||||||
"reboot_timeout": self.reboot_timeout,
|
"reboot_timeout": self.reboot_timeout,
|
||||||
"ping_timeout": self.ping_timeout,
|
"ping_timeout": self.ping_timeout,
|
||||||
@@ -190,8 +193,11 @@ class Config:
|
|||||||
def _relative_path(self, path: str) -> str:
|
def _relative_path(self, path: str) -> str:
|
||||||
"""Convert an absolute path to a relative path from the config file.
|
"""Convert an absolute path to a relative path from the config file.
|
||||||
|
|
||||||
|
Uses :func:`os.path.relpath` which works across filesystem
|
||||||
|
boundaries (unlike :meth:`Path.relative_to`).
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
path: Absolute path string.
|
path: Absolute or relative path string.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Relative path string, or the original if conversion fails.
|
Relative path string, or the original if conversion fails.
|
||||||
@@ -201,7 +207,8 @@ class Config:
|
|||||||
try:
|
try:
|
||||||
target = Path(path)
|
target = Path(path)
|
||||||
if target.is_absolute():
|
if target.is_absolute():
|
||||||
return str(target.relative_to(self._config_path.parent))
|
base = str(self._config_path.parent)
|
||||||
|
return os.path.relpath(str(target), base)
|
||||||
except (ValueError, OSError):
|
except (ValueError, OSError):
|
||||||
pass
|
pass
|
||||||
return path
|
return path
|
||||||
@@ -219,6 +226,34 @@ class Config:
|
|||||||
return Path()
|
return Path()
|
||||||
return (self._config_path.parent / relative_path).resolve()
|
return (self._config_path.parent / relative_path).resolve()
|
||||||
|
|
||||||
|
def validate_paths(self) -> list[str]:
|
||||||
|
"""Validate file paths in config and clear those that don't exist.
|
||||||
|
|
||||||
|
Checks each file path field, resolves it against the config
|
||||||
|
directory, and clears it if the target file does not exist
|
||||||
|
on the current filesystem.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
List of field names that were cleared.
|
||||||
|
"""
|
||||||
|
cleared: list[str] = []
|
||||||
|
path_fields = [
|
||||||
|
"bootloader_bit_path",
|
||||||
|
"bootloader_elf_path",
|
||||||
|
"bootloader_bin_path",
|
||||||
|
"fsbl_elf_path",
|
||||||
|
"firmware_bin_path",
|
||||||
|
]
|
||||||
|
for attr in path_fields:
|
||||||
|
val = getattr(self, attr, "")
|
||||||
|
if not val:
|
||||||
|
continue
|
||||||
|
resolved = self.resolve_path(val)
|
||||||
|
if not resolved.is_file():
|
||||||
|
setattr(self, attr, "")
|
||||||
|
cleared.append(attr)
|
||||||
|
return cleared
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def config_path(self) -> Path | None:
|
def config_path(self) -> Path | None:
|
||||||
"""Return the path to the config file on disk."""
|
"""Return the path to the config file on disk."""
|
||||||
|
|||||||
@@ -0,0 +1,82 @@
|
|||||||
|
"""File metadata utilities — CRC32, human-readable size, timestamps."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import datetime
|
||||||
|
import os
|
||||||
|
import zlib
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class FileInfo:
|
||||||
|
"""Computed metadata for a single file."""
|
||||||
|
|
||||||
|
path: str
|
||||||
|
exists: bool = False
|
||||||
|
size: int = 0
|
||||||
|
crc32: int = 0
|
||||||
|
mtime: float = 0.0
|
||||||
|
|
||||||
|
@property
|
||||||
|
def size_human(self) -> str:
|
||||||
|
"""Human-readable file size (KiB / MiB / GiB)."""
|
||||||
|
if self.size >= 1_073_741_824:
|
||||||
|
return f"{self.size / 1_073_741_824:.2f} GiB"
|
||||||
|
if self.size >= 1_048_576:
|
||||||
|
return f"{self.size / 1_048_576:.2f} MiB"
|
||||||
|
if self.size >= 1_024:
|
||||||
|
return f"{self.size / 1_024:.2f} KiB"
|
||||||
|
return f"{self.size} B"
|
||||||
|
|
||||||
|
@property
|
||||||
|
def mtime_iso(self) -> str:
|
||||||
|
"""ISO-formatted modification time, or empty string."""
|
||||||
|
if not self.mtime:
|
||||||
|
return ""
|
||||||
|
return datetime.datetime.fromtimestamp(self.mtime).strftime(
|
||||||
|
"%Y-%m-%d %H:%M:%S"
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def crc32_hex(self) -> str:
|
||||||
|
"""8-character hex CRC32 representation."""
|
||||||
|
return f"{self.crc32 & 0xFFFFFFFF:08X}"
|
||||||
|
|
||||||
|
|
||||||
|
def compute_file_info(file_path: str | Path) -> FileInfo:
|
||||||
|
"""Compute size, CRC32, and modification time for a file.
|
||||||
|
|
||||||
|
If the file does not exist or cannot be read, the returned
|
||||||
|
``FileInfo`` will have ``exists=False`` and all numeric fields
|
||||||
|
set to zero.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
file_path: Absolute or relative path to the file.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
FileInfo dataclass with computed metadata.
|
||||||
|
"""
|
||||||
|
path = Path(file_path)
|
||||||
|
info = FileInfo(path=str(path))
|
||||||
|
if not path.is_file():
|
||||||
|
return info
|
||||||
|
info.exists = True
|
||||||
|
try:
|
||||||
|
stat = path.stat()
|
||||||
|
info.size = stat.st_size
|
||||||
|
info.mtime = stat.st_mtime
|
||||||
|
except OSError:
|
||||||
|
return info
|
||||||
|
|
||||||
|
# CRC32
|
||||||
|
try:
|
||||||
|
value = 0
|
||||||
|
with open(path, "rb") as f:
|
||||||
|
while chunk := f.read(1_048_576): # 1 MiB chunks
|
||||||
|
value = zlib.crc32(chunk, value)
|
||||||
|
info.crc32 = value
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
return info
|
||||||
+320
-47
@@ -7,6 +7,7 @@ built with CustomTkinter.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
import os
|
||||||
|
import sys
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
import tkinter as tk
|
import tkinter as tk
|
||||||
@@ -25,6 +26,7 @@ from tftp_manager import tftp_upload, tftp_download, tftp_upload_verify, TftpRes
|
|||||||
from reboot_manager import reboot_zynq, RebootResult
|
from reboot_manager import reboot_zynq, RebootResult
|
||||||
from boot_verifier import verify_boot, verify_zynq_status, BootVerificationResult
|
from boot_verifier import verify_boot, verify_zynq_status, BootVerificationResult
|
||||||
from serial_monitor import detect_serial_ports, parse_boot_output, check_uart_available, UartMonitor
|
from serial_monitor import detect_serial_ports, parse_boot_output, check_uart_available, UartMonitor
|
||||||
|
from file_utils import compute_file_info, FileInfo
|
||||||
from gui.styles import (
|
from gui.styles import (
|
||||||
WINDOW_WIDTH,
|
WINDOW_WIDTH,
|
||||||
WINDOW_HEIGHT,
|
WINDOW_HEIGHT,
|
||||||
@@ -90,8 +92,42 @@ class MainWindow(ctk.CTk):
|
|||||||
# StringVar references for config sync
|
# StringVar references for config sync
|
||||||
self._ip_string_var: ctk.StringVar | None = None
|
self._ip_string_var: ctk.StringVar | None = None
|
||||||
|
|
||||||
|
# SD boot mode detection
|
||||||
|
self._hw_server_pid: int | None = None
|
||||||
|
self._sd_detected: bool = False
|
||||||
|
self._sd_timer_id: str | None = None
|
||||||
|
|
||||||
self._load_config()
|
self._load_config()
|
||||||
self._build_ui()
|
self._build_ui()
|
||||||
|
|
||||||
|
# Validate file paths — clear any that don't exist on this machine
|
||||||
|
if self._config and self._config._config_path:
|
||||||
|
try:
|
||||||
|
cleared = self._config.validate_paths()
|
||||||
|
if cleared:
|
||||||
|
joined = ", ".join(cleared)
|
||||||
|
self._log_message(f" ⚠ Cleared stale paths from config: {joined}")
|
||||||
|
# Also clear the UI selectors so stale paths don't
|
||||||
|
# get re-saved by subsequent _auto_save_config calls
|
||||||
|
_selector_map = {
|
||||||
|
"bootloader_bit_path": self._bit_selector,
|
||||||
|
"bootloader_elf_path": self._elf_selector,
|
||||||
|
"bootloader_bin_path": self._bootloader_bin_selector,
|
||||||
|
"fsbl_elf_path": self._fsbl_selector,
|
||||||
|
"firmware_bin_path": self._firmware_bin_selector,
|
||||||
|
}
|
||||||
|
for attr in cleared:
|
||||||
|
sel = _selector_map.get(attr)
|
||||||
|
if sel:
|
||||||
|
sel.set_path("")
|
||||||
|
self._config.save()
|
||||||
|
self._log_message(f" ✓ Config saved after clearing stale paths")
|
||||||
|
except Exception as e:
|
||||||
|
self._log_message(f" ✗ Config validation failed: {e}")
|
||||||
|
|
||||||
|
# Log file metadata for all configured paths
|
||||||
|
self._log_all_file_info()
|
||||||
|
|
||||||
self._update_step_dependencies()
|
self._update_step_dependencies()
|
||||||
|
|
||||||
# Handle window close
|
# Handle window close
|
||||||
@@ -109,6 +145,18 @@ class MainWindow(ctk.CTk):
|
|||||||
|
|
||||||
# ── Config ─────────────────────────────────────────────────
|
# ── Config ─────────────────────────────────────────────────
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _get_app_dir() -> Path:
|
||||||
|
"""Return the application directory (where config.yaml lives).
|
||||||
|
|
||||||
|
When running as a PyInstaller bundle (frozen), uses the
|
||||||
|
directory of the .exe file. Otherwise navigates from
|
||||||
|
``__file__`` up to the project root.
|
||||||
|
"""
|
||||||
|
if getattr(sys, 'frozen', False):
|
||||||
|
return Path(sys.executable).parent
|
||||||
|
return Path(__file__).parent.parent.parent
|
||||||
|
|
||||||
def _find_config(self) -> Path | None:
|
def _find_config(self) -> Path | None:
|
||||||
"""Find the user configuration file.
|
"""Find the user configuration file.
|
||||||
|
|
||||||
@@ -120,8 +168,7 @@ class MainWindow(ctk.CTk):
|
|||||||
Returns:
|
Returns:
|
||||||
Path to user config file, or None if not found.
|
Path to user config file, or None if not found.
|
||||||
"""
|
"""
|
||||||
# Navigate from src/gui/main_window.py -> project root
|
app_dir = self._get_app_dir()
|
||||||
app_dir = Path(__file__).parent.parent.parent
|
|
||||||
|
|
||||||
# 1. Check for user config at project root
|
# 1. Check for user config at project root
|
||||||
user_config = app_dir / "config.yaml"
|
user_config = app_dir / "config.yaml"
|
||||||
@@ -144,33 +191,39 @@ class MainWindow(ctk.CTk):
|
|||||||
def _get_user_config_path(self) -> Path:
|
def _get_user_config_path(self) -> Path:
|
||||||
"""Get the path to the user config file (config.yaml).
|
"""Get the path to the user config file (config.yaml).
|
||||||
|
|
||||||
Creates config.yaml in project root if it doesn't exist.
|
Creates config.yaml in app directory if it doesn't exist.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Path to user config file.
|
Path to user config file.
|
||||||
"""
|
"""
|
||||||
# Navigate from src/gui/main_window.py -> project root
|
app_dir = self._get_app_dir()
|
||||||
app_dir = Path(__file__).parent.parent.parent
|
|
||||||
config_path = app_dir / "config.yaml"
|
config_path = app_dir / "config.yaml"
|
||||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
return config_path
|
return config_path
|
||||||
|
|
||||||
def _load_config(self) -> None:
|
def _load_config(self) -> None:
|
||||||
"""Load configuration from file or create default."""
|
"""Load configuration from file or create a blank one on disk.
|
||||||
|
|
||||||
|
If config.yaml does not exist or cannot be read, a new one
|
||||||
|
is created with default values and written to disk immediately.
|
||||||
|
"""
|
||||||
|
user_config_path = self._get_user_config_path()
|
||||||
if self._config_path:
|
if self._config_path:
|
||||||
try:
|
try:
|
||||||
self._config = Config.from_file(self._config_path)
|
self._config = Config.from_file(self._config_path)
|
||||||
if self._config_path.name == "default_config.yaml":
|
if self._config_path.name == "default_config.yaml":
|
||||||
user_config_path = self._get_user_config_path()
|
|
||||||
self._config._config_path = user_config_path
|
self._config._config_path = user_config_path
|
||||||
|
return
|
||||||
except Exception:
|
except Exception:
|
||||||
|
pass # Fall through and create default
|
||||||
|
|
||||||
|
# Config file missing or unreadable — create a blank one
|
||||||
self._config = Config.from_default()
|
self._config = Config.from_default()
|
||||||
user_config_path = self._get_user_config_path()
|
|
||||||
self._config._config_path = user_config_path
|
|
||||||
else:
|
|
||||||
self._config = Config.from_default()
|
|
||||||
user_config_path = self._get_user_config_path()
|
|
||||||
self._config._config_path = user_config_path
|
self._config._config_path = user_config_path
|
||||||
|
try:
|
||||||
|
self._config.save()
|
||||||
|
except Exception:
|
||||||
|
pass # Will be saved later by _auto_save_config
|
||||||
|
|
||||||
def _reload_config(self) -> None:
|
def _reload_config(self) -> None:
|
||||||
"""Re-read config.yaml and update UI selectors with latest paths.
|
"""Re-read config.yaml and update UI selectors with latest paths.
|
||||||
@@ -185,8 +238,15 @@ class MainWindow(ctk.CTk):
|
|||||||
try:
|
try:
|
||||||
fresh = Config.from_file(user_path)
|
fresh = Config.from_file(user_path)
|
||||||
self._config = fresh
|
self._config = fresh
|
||||||
# Update UI selectors with resolved (absolute) paths — selectors
|
# Update UI selectors — suppress callbacks to avoid
|
||||||
# display only the filename internally.
|
# redundant file-info logging and status-panel rewrites.
|
||||||
|
_sels = [
|
||||||
|
self._bit_selector, self._elf_selector,
|
||||||
|
self._bootloader_bin_selector, self._fsbl_selector,
|
||||||
|
self._firmware_bin_selector,
|
||||||
|
]
|
||||||
|
for s in _sels:
|
||||||
|
s._suppress_callback = True
|
||||||
for attr, selector in [
|
for attr, selector in [
|
||||||
("bootloader_bit_path", self._bit_selector),
|
("bootloader_bit_path", self._bit_selector),
|
||||||
("bootloader_elf_path", self._elf_selector),
|
("bootloader_elf_path", self._elf_selector),
|
||||||
@@ -199,6 +259,8 @@ class MainWindow(ctk.CTk):
|
|||||||
resolved = fresh.resolve_path(val)
|
resolved = fresh.resolve_path(val)
|
||||||
selector.set_relative_path(val)
|
selector.set_relative_path(val)
|
||||||
selector.set_path(str(resolved))
|
selector.set_path(str(resolved))
|
||||||
|
for s in _sels:
|
||||||
|
s._suppress_callback = False
|
||||||
# Update IP
|
# Update IP
|
||||||
if self._ip_string_var:
|
if self._ip_string_var:
|
||||||
self._ip_string_var.set(fresh.zynq_ip)
|
self._ip_string_var.set(fresh.zynq_ip)
|
||||||
@@ -234,7 +296,8 @@ class MainWindow(ctk.CTk):
|
|||||||
if hasattr(self, '_port_var'):
|
if hasattr(self, '_port_var'):
|
||||||
self._config.serial_port = self._port_var.get()
|
self._config.serial_port = self._port_var.get()
|
||||||
# Sync file paths from FileSelectors
|
# Sync file paths from FileSelectors
|
||||||
files = self._get_selected_files()
|
# Always derive relative from absolute — the _relative_path
|
||||||
|
# stored in FileSelector can be stale after Browse.
|
||||||
for attr, selector in [
|
for attr, selector in [
|
||||||
("bootloader_bit_path", self._bit_selector),
|
("bootloader_bit_path", self._bit_selector),
|
||||||
("bootloader_elf_path", self._elf_selector),
|
("bootloader_elf_path", self._elf_selector),
|
||||||
@@ -242,17 +305,17 @@ class MainWindow(ctk.CTk):
|
|||||||
("fsbl_elf_path", self._fsbl_selector),
|
("fsbl_elf_path", self._fsbl_selector),
|
||||||
("firmware_bin_path", self._firmware_bin_selector),
|
("firmware_bin_path", self._firmware_bin_selector),
|
||||||
]:
|
]:
|
||||||
rel = selector.get_relative_path()
|
|
||||||
if rel:
|
|
||||||
setattr(self._config, attr, rel)
|
|
||||||
else:
|
|
||||||
abs_path = selector.get_path()
|
abs_path = selector.get_path()
|
||||||
if abs_path:
|
if abs_path:
|
||||||
self._config._config_path # ensure _config_path is set
|
|
||||||
setattr(self._config, attr, self._config._relative_path(abs_path))
|
setattr(self._config, attr, self._config._relative_path(abs_path))
|
||||||
|
else:
|
||||||
|
setattr(self._config, attr, "")
|
||||||
# Sync erase checkbox (may not exist yet)
|
# Sync erase checkbox (may not exist yet)
|
||||||
if hasattr(self, '_erase_cb_var'):
|
if hasattr(self, '_erase_cb_var'):
|
||||||
self._config.erase_all = self._erase_cb_var.get()
|
self._config.erase_all = self._erase_cb_var.get()
|
||||||
|
# Sync download verify checkbox
|
||||||
|
if hasattr(self, '_download_verify_var'):
|
||||||
|
self._config.download_verify = self._download_verify_var.get()
|
||||||
|
|
||||||
def _auto_save_config(self) -> None:
|
def _auto_save_config(self) -> None:
|
||||||
"""Auto-save config to disk after UI changes.
|
"""Auto-save config to disk after UI changes.
|
||||||
@@ -297,6 +360,11 @@ class MainWindow(ctk.CTk):
|
|||||||
if self._config:
|
if self._config:
|
||||||
self._config.erase_all = self._erase_cb_var.get()
|
self._config.erase_all = self._erase_cb_var.get()
|
||||||
|
|
||||||
|
def _on_download_verify_toggle(self) -> None:
|
||||||
|
"""Update config.download_verify immediately when checkbox toggles."""
|
||||||
|
if self._config:
|
||||||
|
self._config.download_verify = self._download_verify_var.get()
|
||||||
|
|
||||||
def _browse_xilinx_path(self) -> None:
|
def _browse_xilinx_path(self) -> None:
|
||||||
"""Open directory dialog to select Xilinx root folder."""
|
"""Open directory dialog to select Xilinx root folder."""
|
||||||
from tkinter import filedialog
|
from tkinter import filedialog
|
||||||
@@ -318,13 +386,58 @@ class MainWindow(ctk.CTk):
|
|||||||
self._config.zynq_ip = self._ip_string_var.get()
|
self._config.zynq_ip = self._ip_string_var.get()
|
||||||
|
|
||||||
def _on_file_selected(self, path: str) -> None:
|
def _on_file_selected(self, path: str) -> None:
|
||||||
"""Handle file selection — sync and auto-save to disk.
|
"""Handle file selection — sync, auto-save, and log file info."""
|
||||||
|
from file_utils import compute_file_info
|
||||||
When the user selects a file via the browse dialog, immediately
|
|
||||||
sync the path to the Config object and persist it to YAML so
|
|
||||||
that subsequent step executions pick up the new value.
|
|
||||||
"""
|
|
||||||
self._auto_save_config()
|
self._auto_save_config()
|
||||||
|
if path:
|
||||||
|
info = compute_file_info(path)
|
||||||
|
self._log_file_info(info)
|
||||||
|
# Update status info panel
|
||||||
|
self._log_all_file_info()
|
||||||
|
|
||||||
|
def _log_file_info(self, info: FileInfo) -> None:
|
||||||
|
"""Log file metadata in the log window.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
info: FileInfo dataclass with computed metadata.
|
||||||
|
"""
|
||||||
|
filename = os.path.basename(info.path) if info.path else "(unknown)"
|
||||||
|
if not info.exists:
|
||||||
|
self._log_message(f" ⚠ {filename}: file not found")
|
||||||
|
return
|
||||||
|
self._log_message(
|
||||||
|
f" ✓ {filename}: "
|
||||||
|
f"{info.size_human} | CRC {info.crc32_hex} | {info.mtime_iso}"
|
||||||
|
)
|
||||||
|
|
||||||
|
def _log_all_file_info(self) -> None:
|
||||||
|
"""Update file metadata in the status info panel (preserves other info).
|
||||||
|
|
||||||
|
Each file line is updated in-place via ``set_info()`` — existing
|
||||||
|
Zynq info, tool versions, etc. are left untouched.
|
||||||
|
"""
|
||||||
|
if not self._config:
|
||||||
|
return
|
||||||
|
paths = [
|
||||||
|
("FSBL ELF", self._config.fsbl_elf_path),
|
||||||
|
("Bitstream", self._config.bootloader_bit_path),
|
||||||
|
("Bootloader ELF", self._config.bootloader_elf_path),
|
||||||
|
("Bootloader BIN", self._config.bootloader_bin_path),
|
||||||
|
("Firmware BIN", self._config.firmware_bin_path),
|
||||||
|
]
|
||||||
|
for label, path in paths:
|
||||||
|
if not path:
|
||||||
|
self._status.set_info(label, "(not set)")
|
||||||
|
continue
|
||||||
|
resolved = self._config.resolve_path(path)
|
||||||
|
info = compute_file_info(resolved)
|
||||||
|
if info.exists:
|
||||||
|
self._status.set_info(
|
||||||
|
label,
|
||||||
|
f"{info.size_human} CRC {info.crc32_hex} {info.mtime_iso}"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
self._status.set_info(label, "(not found)")
|
||||||
|
|
||||||
# ── UI Construction ────────────────────────────────────────
|
# ── UI Construction ────────────────────────────────────────
|
||||||
|
|
||||||
@@ -334,7 +447,8 @@ class MainWindow(ctk.CTk):
|
|||||||
self.main_frame = ctk.CTkScrollableFrame(self)
|
self.main_frame = ctk.CTkScrollableFrame(self)
|
||||||
self.main_frame.pack(fill="both", expand=True, padx=PADDING_LARGE, pady=PADDING_LARGE)
|
self.main_frame.pack(fill="both", expand=True, padx=PADDING_LARGE, pady=PADDING_LARGE)
|
||||||
self.main_frame.grid_rowconfigure(1, weight=1)
|
self.main_frame.grid_rowconfigure(1, weight=1)
|
||||||
self.main_frame.grid_columnconfigure(0, weight=1)
|
self.main_frame.grid_columnconfigure(0, weight=7, uniform="main_col")
|
||||||
|
self.main_frame.grid_columnconfigure(1, weight=3, uniform="main_col")
|
||||||
|
|
||||||
# ── Header ──
|
# ── Header ──
|
||||||
self._build_header(self.main_frame)
|
self._build_header(self.main_frame)
|
||||||
@@ -351,7 +465,7 @@ class MainWindow(ctk.CTk):
|
|||||||
# ── Right Panel: Configuration ──
|
# ── Right Panel: Configuration ──
|
||||||
right_frame = ctk.CTkFrame(self.main_frame)
|
right_frame = ctk.CTkFrame(self.main_frame)
|
||||||
right_frame.grid(row=1, column=1, sticky="nsew", padx=(PADDING, 0))
|
right_frame.grid(row=1, column=1, sticky="nsew", padx=(PADDING, 0))
|
||||||
right_frame.grid_rowconfigure(3, weight=1)
|
right_frame.grid_rowconfigure(3, weight=3) # status panel — main info area
|
||||||
right_frame.grid_columnconfigure(0, weight=1)
|
right_frame.grid_columnconfigure(0, weight=1)
|
||||||
|
|
||||||
self._build_config_panel(right_frame)
|
self._build_config_panel(right_frame)
|
||||||
@@ -550,6 +664,7 @@ class MainWindow(ctk.CTk):
|
|||||||
padx=PADDING, pady=PADDING,
|
padx=PADDING, pady=PADDING,
|
||||||
)
|
)
|
||||||
panel.grid_rowconfigure(9, weight=1)
|
panel.grid_rowconfigure(9, weight=1)
|
||||||
|
panel.grid_columnconfigure(0, weight=1) # let content stretch
|
||||||
|
|
||||||
title = ctk.CTkLabel(
|
title = ctk.CTkLabel(
|
||||||
panel,
|
panel,
|
||||||
@@ -664,6 +779,17 @@ class MainWindow(ctk.CTk):
|
|||||||
)
|
)
|
||||||
flash_model_label.grid(row=1, column=0, sticky="w", padx=PADDING_SMALL, pady=(2, 0))
|
flash_model_label.grid(row=1, column=0, sticky="w", padx=PADDING_SMALL, pady=(2, 0))
|
||||||
|
|
||||||
|
# Download verify checkbox (Step 4.2)
|
||||||
|
self._download_verify_var = ctk.BooleanVar(value=self._config.download_verify)
|
||||||
|
self._download_verify_cb = ctk.CTkCheckBox(
|
||||||
|
flash_frame,
|
||||||
|
text="下载校验 + CRC (Step 4.2)",
|
||||||
|
variable=self._download_verify_var,
|
||||||
|
font=FONT_SMALL,
|
||||||
|
command=self._on_download_verify_toggle,
|
||||||
|
)
|
||||||
|
self._download_verify_cb.grid(row=2, column=0, sticky="w", padx=PADDING_SMALL, pady=(2, 0))
|
||||||
|
|
||||||
# Firmware BIN path (TFTP upload)
|
# Firmware BIN path (TFTP upload)
|
||||||
self._firmware_bin_selector = FileSelector(
|
self._firmware_bin_selector = FileSelector(
|
||||||
panel,
|
panel,
|
||||||
@@ -812,24 +938,29 @@ class MainWindow(ctk.CTk):
|
|||||||
self._status = StatusDisplay(panel)
|
self._status = StatusDisplay(panel)
|
||||||
self._status.grid(row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING)
|
self._status.grid(row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING)
|
||||||
|
|
||||||
# UART warning label (persistent, not overwritten by other status updates)
|
# UART warning — multi-line textbox, wraps long messages instead
|
||||||
self._uart_warning_label = ctk.CTkLabel(
|
# of expanding the right panel horizontally
|
||||||
|
self._uart_warning = ctk.CTkTextbox(
|
||||||
panel,
|
panel,
|
||||||
text="",
|
|
||||||
font=FONT_BODY,
|
font=FONT_BODY,
|
||||||
anchor="w",
|
height=40,
|
||||||
text_color=WARNING_COLOR,
|
corner_radius=4,
|
||||||
|
fg_color="transparent",
|
||||||
|
state="disabled",
|
||||||
|
wrap="word",
|
||||||
)
|
)
|
||||||
self._uart_warning_label.grid(
|
self._uart_warning.grid(
|
||||||
row=1, column=0, sticky="ew", padx=PADDING, pady=(0, PADDING_SMALL)
|
row=1, column=0, sticky="ew", padx=PADDING, pady=(0, PADDING_SMALL)
|
||||||
)
|
)
|
||||||
self._uart_warning_label.grid_remove() # Hide by default
|
self._uart_warning.grid_remove() # Hide by default
|
||||||
|
|
||||||
# ── Workflow Steps ─────────────────────────────────────────
|
# ── Workflow Steps ─────────────────────────────────────────
|
||||||
|
|
||||||
def _log_message(self, message: str) -> None:
|
def _log_message(self, message: str) -> None:
|
||||||
"""Append a message to the log display (thread-safe)."""
|
"""Append a message to the log display and terminal (thread-safe)."""
|
||||||
|
print(message, flush=True)
|
||||||
self.after(0, lambda: self._log_display.append(message))
|
self.after(0, lambda: self._log_display.append(message))
|
||||||
|
self._check_sd_boot(message)
|
||||||
|
|
||||||
def _set_step_status(self, index: int, status: str) -> None:
|
def _set_step_status(self, index: int, status: str) -> None:
|
||||||
"""Set the status of a workflow step.
|
"""Set the status of a workflow step.
|
||||||
@@ -885,6 +1016,10 @@ class MainWindow(ctk.CTk):
|
|||||||
if result.is_ready:
|
if result.is_ready:
|
||||||
self._vitis_status.configure(text="Vitis: Ready", text_color=SUCCESS_COLOR)
|
self._vitis_status.configure(text="Vitis: Ready", text_color=SUCCESS_COLOR)
|
||||||
self._log_message(" Vitis/Vivado tools available")
|
self._log_message(" Vitis/Vivado tools available")
|
||||||
|
# Show tool versions in status display
|
||||||
|
for tool_name, tool_info in status_dict["tools"].items():
|
||||||
|
if tool_info["found"] and tool_info.get("version"):
|
||||||
|
self._status.set_info(tool_name, f"v{tool_info['version']}")
|
||||||
# Report all found versions for diagnostics
|
# Report all found versions for diagnostics
|
||||||
if self._config.xilinx_path:
|
if self._config.xilinx_path:
|
||||||
try:
|
try:
|
||||||
@@ -904,6 +1039,8 @@ class MainWindow(ctk.CTk):
|
|||||||
self._log_message(" Checking Zynq JTAG presence...")
|
self._log_message(" Checking Zynq JTAG presence...")
|
||||||
jtag_result = check_zynq_jtag(self._config, self._jtag_callback)
|
jtag_result = check_zynq_jtag(self._config, self._jtag_callback)
|
||||||
jtag_dict = get_zynq_status_dict(jtag_result)
|
jtag_dict = get_zynq_status_dict(jtag_result)
|
||||||
|
if jtag_result.hw_server_pid:
|
||||||
|
self._hw_server_pid = jtag_result.hw_server_pid
|
||||||
|
|
||||||
# Log JTAG chain information
|
# Log JTAG chain information
|
||||||
if jtag_result.jtag_chain:
|
if jtag_result.jtag_chain:
|
||||||
@@ -938,6 +1075,7 @@ class MainWindow(ctk.CTk):
|
|||||||
|
|
||||||
self._log_message(f" ✓ Zynq {part} detected on JTAG chain")
|
self._log_message(f" ✓ Zynq {part} detected on JTAG chain")
|
||||||
self._jtag_status.configure(text=f"JTAG: {part} ✓", text_color=SUCCESS_COLOR)
|
self._jtag_status.configure(text=f"JTAG: {part} ✓", text_color=SUCCESS_COLOR)
|
||||||
|
self._status.set_info("Zynq", part)
|
||||||
else:
|
else:
|
||||||
self._zynq_jtag_present = False
|
self._zynq_jtag_present = False
|
||||||
self._zynq_jtag_info = None
|
self._zynq_jtag_info = None
|
||||||
@@ -949,7 +1087,8 @@ class MainWindow(ctk.CTk):
|
|||||||
self._log_message(" Checking UART serial port...")
|
self._log_message(" Checking UART serial port...")
|
||||||
port = self._config.serial_port
|
port = self._config.serial_port
|
||||||
if port:
|
if port:
|
||||||
available, reason = check_uart_available(port, self._config.serial_baudrate)
|
available, reason = check_uart_available(port, self._config.serial_baudrate,
|
||||||
|
monitor=self._uart_monitor)
|
||||||
self._uart_available = available
|
self._uart_available = available
|
||||||
self._uart_message = reason
|
self._uart_message = reason
|
||||||
if available:
|
if available:
|
||||||
@@ -977,17 +1116,26 @@ class MainWindow(ctk.CTk):
|
|||||||
self.after(0, lambda: self._show_uart_warning(msg))
|
self.after(0, lambda: self._show_uart_warning(msg))
|
||||||
|
|
||||||
def _show_uart_warning(self, msg: str) -> None:
|
def _show_uart_warning(self, msg: str) -> None:
|
||||||
"""Display the persistent UART warning message.
|
"""Display the persistent UART warning message (multi-line).
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
msg: Warning message to display.
|
msg: Warning message to display.
|
||||||
"""
|
"""
|
||||||
self._uart_warning_label.configure(text=msg)
|
self._uart_warning.configure(state="normal")
|
||||||
self._uart_warning_label.grid()
|
self._uart_warning.delete("1.0", "end")
|
||||||
|
self._uart_warning.insert("1.0", msg)
|
||||||
|
self._uart_warning.tag_config("warn", foreground=WARNING_COLOR)
|
||||||
|
self._uart_warning.tag_add("warn", "1.0", "end")
|
||||||
|
self._uart_warning.configure(state="disabled")
|
||||||
|
self._uart_warning.grid()
|
||||||
|
|
||||||
|
# Auto-size height based on line count
|
||||||
|
line_count = msg.count("\n") + 1
|
||||||
|
self._uart_warning.configure(height=max(2, line_count))
|
||||||
|
|
||||||
def _hide_uart_warning(self) -> None:
|
def _hide_uart_warning(self) -> None:
|
||||||
"""Hide the persistent UART warning label."""
|
"""Hide the persistent UART warning."""
|
||||||
self._uart_warning_label.grid_remove()
|
self._uart_warning.grid_remove()
|
||||||
|
|
||||||
def _on_window_resize(self, event) -> None:
|
def _on_window_resize(self, event) -> None:
|
||||||
"""Handle window resize / maximize events.
|
"""Handle window resize / maximize events.
|
||||||
@@ -1285,7 +1433,19 @@ class MainWindow(ctk.CTk):
|
|||||||
self._log_message(" ⏳ Waiting 2s before download...")
|
self._log_message(" ⏳ Waiting 2s before download...")
|
||||||
time.sleep(2)
|
time.sleep(2)
|
||||||
|
|
||||||
# 4b: TFTP Download Verify
|
# 4b+4c: TFTP Download Verify + CRC Check
|
||||||
|
if not self._config.download_verify:
|
||||||
|
self._log_message("Step 4b+4c: Download verify & CRC — skipped (disabled in config)")
|
||||||
|
all_results.append(True) # 4b
|
||||||
|
all_results.append(True) # 4c
|
||||||
|
if hasattr(self, '_tftp_sub_step_frame'):
|
||||||
|
self._tftp_sub_step_frame.set_expanded(True)
|
||||||
|
self._tftp_sub_step_frame.set_step_skipped("Download Verify")
|
||||||
|
self._tftp_sub_step_frame.set_step_skipped("CRC Check")
|
||||||
|
self._log_message(" ◌ Download Verify: Skip")
|
||||||
|
self._log_message(" ◌ CRC Check: Skip")
|
||||||
|
else:
|
||||||
|
# 4b: Download Verify
|
||||||
self._log_message("Step 4b: TFTP download verify...")
|
self._log_message("Step 4b: TFTP download verify...")
|
||||||
self._tftp_phase = "Download Verify"
|
self._tftp_phase = "Download Verify"
|
||||||
if hasattr(self, '_tftp_sub_step_frame'):
|
if hasattr(self, '_tftp_sub_step_frame'):
|
||||||
@@ -1309,8 +1469,7 @@ class MainWindow(ctk.CTk):
|
|||||||
|
|
||||||
if not all_results[-1]:
|
if not all_results[-1]:
|
||||||
self._log_message(" Skipping CRC check (download failed)")
|
self._log_message(" Skipping CRC check (download failed)")
|
||||||
return False
|
else:
|
||||||
|
|
||||||
# 4c: CRC Check
|
# 4c: CRC Check
|
||||||
self._log_message("Step 4c: CRC check...")
|
self._log_message("Step 4c: CRC check...")
|
||||||
if hasattr(self, '_tftp_sub_step_frame'):
|
if hasattr(self, '_tftp_sub_step_frame'):
|
||||||
@@ -1332,7 +1491,7 @@ class MainWindow(ctk.CTk):
|
|||||||
all_results.append(False)
|
all_results.append(False)
|
||||||
|
|
||||||
if not all_results[-1]:
|
if not all_results[-1]:
|
||||||
self._log_message(" Skipping reboot (CRC mismatch)")
|
self._log_message(" Skipping reboot (download/CRC failed)")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# 4d: Reboot
|
# 4d: Reboot
|
||||||
@@ -1429,11 +1588,33 @@ class MainWindow(ctk.CTk):
|
|||||||
discovered_ip = parsed.ip_address
|
discovered_ip = parsed.ip_address
|
||||||
self._log_message(f" IP from UART buffer: {discovered_ip}")
|
self._log_message(f" IP from UART buffer: {discovered_ip}")
|
||||||
|
|
||||||
# Discover actual IP after reboot (may have changed via DHCP)
|
# Fallback: if no IP from UART, try ping → UDP scan, then extra delay
|
||||||
if not discovered_ip:
|
if not discovered_ip:
|
||||||
self._log_message(" ⏳ Network stack — waiting 3s...")
|
self._log_message(" ⏳ Network stack — waiting 3s...")
|
||||||
time.sleep(3)
|
time.sleep(3)
|
||||||
|
|
||||||
|
# Try ping configured IP first (fast, reliable)
|
||||||
|
self._log_message(f" Pinging {self._config.zynq_ip}...")
|
||||||
|
ok = ping_ip(self._config.zynq_ip, timeout=2)
|
||||||
|
if ok:
|
||||||
|
discovered_ip = self._config.zynq_ip
|
||||||
|
self._log_message(f" ✓ Zynq responds to ping at {discovered_ip}")
|
||||||
|
else:
|
||||||
|
# Fallback: UDP broadcast scan
|
||||||
self._log_message(" Scanning for Zynq IP (192.168.100.11–30)...")
|
self._log_message(" Scanning for Zynq IP (192.168.100.11–30)...")
|
||||||
|
discovered_ip = self._discover_zynq_ip()
|
||||||
|
if discovered_ip:
|
||||||
|
self._log_message(f" ✓ Found Zynq at {discovered_ip}")
|
||||||
|
else:
|
||||||
|
self._log_message(" ⚠ Zynq not found — extra boot delay...")
|
||||||
|
self._log_message(f" ⏳ Waiting {boot_delay}s for late boot...")
|
||||||
|
time.sleep(boot_delay)
|
||||||
|
# Retry ping + scan after extra delay
|
||||||
|
ok2 = ping_ip(self._config.zynq_ip, timeout=2)
|
||||||
|
if ok2:
|
||||||
|
discovered_ip = self._config.zynq_ip
|
||||||
|
self._log_message(f" ✓ Zynq responds to ping at {discovered_ip}")
|
||||||
|
else:
|
||||||
discovered_ip = self._discover_zynq_ip()
|
discovered_ip = self._discover_zynq_ip()
|
||||||
if discovered_ip:
|
if discovered_ip:
|
||||||
self._log_message(f" ✓ Found Zynq at {discovered_ip}")
|
self._log_message(f" ✓ Found Zynq at {discovered_ip}")
|
||||||
@@ -1640,6 +1821,13 @@ class MainWindow(ctk.CTk):
|
|||||||
self._progress.set_value(step_idx / total)
|
self._progress.set_value(step_idx / total)
|
||||||
t_step_start = time.time()
|
t_step_start = time.time()
|
||||||
|
|
||||||
|
# Abort if SD boot timeout fired
|
||||||
|
if not self._is_running:
|
||||||
|
self._log_message(f" ⚠ Step {i+1} aborted (SD boot mode — switch to QSPI)")
|
||||||
|
self._set_step_status(i, "error")
|
||||||
|
results.append(False)
|
||||||
|
break
|
||||||
|
|
||||||
try:
|
try:
|
||||||
success = step_funcs[i]()
|
success = step_funcs[i]()
|
||||||
results.append(success)
|
results.append(success)
|
||||||
@@ -1647,6 +1835,13 @@ class MainWindow(ctk.CTk):
|
|||||||
elapsed = time.time() - t_step_start
|
elapsed = time.time() - t_step_start
|
||||||
self._log_message(f" ⏱ Step {i+1} completed in {elapsed:.0f}s")
|
self._log_message(f" ⏱ Step {i+1} completed in {elapsed:.0f}s")
|
||||||
|
|
||||||
|
# Step 3: After bootloader load, fix IP to 192.168.100.11
|
||||||
|
if i == 2 and success:
|
||||||
|
self._config.zynq_ip = "192.168.100.11"
|
||||||
|
if self._ip_string_var:
|
||||||
|
self._ip_string_var.set("192.168.100.11")
|
||||||
|
self._log_message(" IP set to 192.168.100.11")
|
||||||
|
|
||||||
# Step 2: mark sub-steps based on result
|
# Step 2: mark sub-steps based on result
|
||||||
if i == 1 and hasattr(self, '_sub_step_frame'):
|
if i == 1 and hasattr(self, '_sub_step_frame'):
|
||||||
if success:
|
if success:
|
||||||
@@ -1663,6 +1858,11 @@ class MainWindow(ctk.CTk):
|
|||||||
results.append(False)
|
results.append(False)
|
||||||
self._set_step_status(i, "error")
|
self._set_step_status(i, "error")
|
||||||
self._log_message(f" Exception in step {i+1}: {e}")
|
self._log_message(f" Exception in step {i+1}: {e}")
|
||||||
|
# Mark sub-step frames as error too
|
||||||
|
if i == 1 and hasattr(self, '_sub_step_frame'):
|
||||||
|
self._sub_step_frame.set_phase("error")
|
||||||
|
if i == 3 and hasattr(self, '_tftp_sub_step_frame'):
|
||||||
|
self._tftp_sub_step_frame.set_step_error("Boot Verify")
|
||||||
|
|
||||||
self._progress.set_value((step_idx + 1) / total)
|
self._progress.set_value((step_idx + 1) / total)
|
||||||
|
|
||||||
@@ -1689,6 +1889,7 @@ class MainWindow(ctk.CTk):
|
|||||||
|
|
||||||
self._progress.set_value(1.0)
|
self._progress.set_value(1.0)
|
||||||
self._is_running = False
|
self._is_running = False
|
||||||
|
self._cancel_sd_timer()
|
||||||
self._run_btn.configure(state="normal", text="▶ Run All Steps")
|
self._run_btn.configure(state="normal", text="▶ Run All Steps")
|
||||||
|
|
||||||
# Update dependency statuses
|
# Update dependency statuses
|
||||||
@@ -1814,6 +2015,7 @@ class MainWindow(ctk.CTk):
|
|||||||
success, message, version = test_serial_version(
|
success, message, version = test_serial_version(
|
||||||
port,
|
port,
|
||||||
self._config.serial_baudrate if self._config else 115200,
|
self._config.serial_baudrate if self._config else 115200,
|
||||||
|
monitor=self._uart_monitor,
|
||||||
)
|
)
|
||||||
|
|
||||||
if success:
|
if success:
|
||||||
@@ -1958,6 +2160,77 @@ class MainWindow(ctk.CTk):
|
|||||||
continue
|
continue
|
||||||
return ""
|
return ""
|
||||||
|
|
||||||
|
# ── SD Boot Detection ──────────────────────────────────────
|
||||||
|
|
||||||
|
def _check_sd_boot(self, line: str) -> None:
|
||||||
|
"""Scan a log line for SD boot mode; start 30s timer if detected.
|
||||||
|
|
||||||
|
New log output after detection resets the timer — recovery
|
||||||
|
may indicate the DIP switch was changed.
|
||||||
|
"""
|
||||||
|
import re
|
||||||
|
for pattern in [
|
||||||
|
re.compile(r"[Bb]oot\s*[Mm]ode\s*(?:is)?\s*[:\[=\s]*(\w+)"),
|
||||||
|
re.compile(r"BootMode\s*[:\[=\s]*(\w+)", re.IGNORECASE),
|
||||||
|
]:
|
||||||
|
m = pattern.search(line)
|
||||||
|
if m and m.group(1).upper() == "SD":
|
||||||
|
if not self._sd_detected:
|
||||||
|
self._sd_detected = True
|
||||||
|
self._log_message(" ⚠ SD boot mode detected — waiting 30s for QSPI switch...")
|
||||||
|
else:
|
||||||
|
self._log_message(" ⚠ SD boot mode still present — resetting 30s timer")
|
||||||
|
self.after_cancel(self._sd_timer_id)
|
||||||
|
self._sd_timer_id = self.after(30000, self._on_sd_timeout)
|
||||||
|
return
|
||||||
|
# Any other log output resets the timer (boot mode may have changed)
|
||||||
|
if self._sd_detected:
|
||||||
|
self._log_message(" ✓ New output received — cancelling SD timer")
|
||||||
|
self._cancel_sd_timer()
|
||||||
|
|
||||||
|
def _cancel_sd_timer(self) -> None:
|
||||||
|
"""Cancel the SD boot timeout if still pending."""
|
||||||
|
if self._sd_timer_id:
|
||||||
|
self.after_cancel(self._sd_timer_id)
|
||||||
|
self._sd_timer_id = None
|
||||||
|
self._sd_detected = False
|
||||||
|
|
||||||
|
def _on_sd_timeout(self) -> None:
|
||||||
|
"""SD boot mode persisted 30s — kill hw_server, alert user, stop workflow."""
|
||||||
|
self._sd_timer_id = None
|
||||||
|
self._sd_detected = False
|
||||||
|
self._log_message(" ✗ SD boot mode persisted >30s — switch DIP to QSPI mode!")
|
||||||
|
self._kill_hw_server()
|
||||||
|
from tkinter import messagebox
|
||||||
|
messagebox.showerror(
|
||||||
|
"启动拨码开关位置错误!",
|
||||||
|
"检测到 Boot mode 为 SD,请将拨码开关切换到 QSPI 模式后重试。",
|
||||||
|
)
|
||||||
|
# Stop current workflow execution
|
||||||
|
self._is_running = False
|
||||||
|
|
||||||
|
def _kill_hw_server(self) -> None:
|
||||||
|
"""Terminate the hw_server process started during Step 1."""
|
||||||
|
if not self._hw_server_pid:
|
||||||
|
return
|
||||||
|
import platform
|
||||||
|
import subprocess
|
||||||
|
import os
|
||||||
|
try:
|
||||||
|
if platform.system().lower() == "windows":
|
||||||
|
subprocess.run(
|
||||||
|
["taskkill", "/F", "/PID", str(self._hw_server_pid)],
|
||||||
|
capture_output=True,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
import signal
|
||||||
|
os.kill(self._hw_server_pid, signal.SIGTERM)
|
||||||
|
self._log_message(f" ⚠ hw_server (PID {self._hw_server_pid}) terminated")
|
||||||
|
except Exception as e:
|
||||||
|
self._log_message(f" ⚠ Failed to kill hw_server: {e}")
|
||||||
|
finally:
|
||||||
|
self._hw_server_pid = None
|
||||||
|
|
||||||
# ── Vitis Check ────────────────────────────────────────────
|
# ── Vitis Check ────────────────────────────────────────────
|
||||||
|
|
||||||
def _check_vitis(self) -> None:
|
def _check_vitis(self) -> None:
|
||||||
|
|||||||
+68
-32
@@ -437,63 +437,82 @@ class ProgressIndicator(ctk.CTkFrame):
|
|||||||
|
|
||||||
|
|
||||||
class StatusDisplay(ctk.CTkFrame):
|
class StatusDisplay(ctk.CTkFrame):
|
||||||
"""Status display widget for operation results.
|
"""Info display widget — persistent key-value file/device info.
|
||||||
|
|
||||||
Shows status messages in a multi-line, expandable text area that
|
Shows aligned ``key : value`` lines for file metadata, Zynq
|
||||||
uses the full available space. Messages are color-coded and
|
part info, tool versions, etc.
|
||||||
automatically wrapped.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, master, **kwargs):
|
def __init__(self, master, **kwargs):
|
||||||
"""Initialize the status display.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
master: Parent widget.
|
|
||||||
"""
|
|
||||||
super().__init__(master, **kwargs)
|
super().__init__(master, **kwargs)
|
||||||
|
self._info_items: list[tuple[str, str]] = []
|
||||||
self._create_widgets()
|
self._create_widgets()
|
||||||
|
|
||||||
def _create_widgets(self) -> None:
|
def _create_widgets(self) -> None:
|
||||||
"""Create the status display UI — multi-line expandable area."""
|
|
||||||
self.grid_rowconfigure(0, weight=1)
|
self.grid_rowconfigure(0, weight=1)
|
||||||
self.grid_columnconfigure(0, weight=1)
|
self.grid_columnconfigure(0, weight=1)
|
||||||
|
|
||||||
self._status_text = ctk.CTkTextbox(
|
self._info_text = ctk.CTkTextbox(
|
||||||
self,
|
self,
|
||||||
font=FONT_BODY,
|
font=FONT_BODY,
|
||||||
corner_radius=CORNER_RADIUS,
|
corner_radius=CORNER_RADIUS,
|
||||||
state="disabled",
|
state="disabled",
|
||||||
wrap="word",
|
wrap="word",
|
||||||
)
|
)
|
||||||
self._status_text.grid(
|
self._info_text.grid(
|
||||||
row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL
|
row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Pre-configure tags
|
||||||
|
for name, color in [
|
||||||
|
("info_label", PRIMARY_COLOR),
|
||||||
|
("info_value", DESC_TEXT_COLOR if _is_dark() else DESC_TEXT_COLOR_LIGHT),
|
||||||
|
]:
|
||||||
|
self._info_text.tag_config(name, foreground=color)
|
||||||
|
|
||||||
def set_status(self, message: str, status: str = "info") -> None:
|
def set_status(self, message: str, status: str = "info") -> None:
|
||||||
"""Update the status message and color.
|
"""Compatibility stub — status messages go to log now."""
|
||||||
|
|
||||||
|
def set_info(self, key: str, value: str) -> None:
|
||||||
|
"""Set or update a persistent info line.
|
||||||
|
|
||||||
|
Displays as ``key : value`` in the info block.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
message: Status message text.
|
key: Label for the info line.
|
||||||
status: Status type for coloring ('info', 'success', 'error', 'warning').
|
value: Value string.
|
||||||
"""
|
"""
|
||||||
self._status_text.configure(state="normal")
|
for i, (k, _) in enumerate(self._info_items):
|
||||||
self._status_text.delete("1.0", "end")
|
if k == key:
|
||||||
|
self._info_items[i] = (key, value)
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
self._info_items.append((key, value))
|
||||||
|
self._refresh_info()
|
||||||
|
|
||||||
colors = {
|
def clear_info(self) -> None:
|
||||||
"info": INFO_COLOR,
|
"""Remove all persistent info lines."""
|
||||||
"success": SUCCESS_COLOR,
|
self._info_items.clear()
|
||||||
"error": DANGER_COLOR,
|
self._info_text.configure(state="normal")
|
||||||
"warning": WARNING_COLOR,
|
self._info_text.delete("1.0", "end")
|
||||||
}
|
self._info_text.configure(state="disabled")
|
||||||
color = colors.get(status, INFO_COLOR)
|
|
||||||
|
|
||||||
self._status_text.insert("1.0", message)
|
def _refresh_info(self) -> None:
|
||||||
# Tag the entire text with the color (CTkTextbox uses tag_config)
|
"""Rebuild the info text block from current items."""
|
||||||
self._status_text.tag_config("status", foreground=color)
|
self._info_text.configure(state="normal")
|
||||||
self._status_text.tag_add("status", "1.0", "end")
|
self._info_text.delete("1.0", "end")
|
||||||
|
if not self._info_items:
|
||||||
self._status_text.configure(state="disabled")
|
self._info_text.configure(state="disabled")
|
||||||
|
return
|
||||||
|
max_key = max(len(k) for k, _ in self._info_items)
|
||||||
|
for key, value in self._info_items:
|
||||||
|
line = f"{key.ljust(max_key + 2)} {value}\n"
|
||||||
|
start = self._info_text.index("end-1c")
|
||||||
|
self._info_text.insert("end", line)
|
||||||
|
end = self._info_text.index("end-1c")
|
||||||
|
self._info_text.tag_add("info_label", start, f"{start}+{len(key)}c")
|
||||||
|
self._info_text.tag_add("info_value", f"{start}+{len(key)}c", end)
|
||||||
|
self._info_text.configure(state="disabled")
|
||||||
|
|
||||||
|
|
||||||
class FileSelector(ctk.CTkFrame):
|
class FileSelector(ctk.CTkFrame):
|
||||||
@@ -525,6 +544,7 @@ class FileSelector(ctk.CTkFrame):
|
|||||||
|
|
||||||
self._callback = callback
|
self._callback = callback
|
||||||
self._file_types = file_types or [("All files", "*")]
|
self._file_types = file_types or [("All files", "*")]
|
||||||
|
self._hint_text = label_text.rstrip(":") # e.g. "FSBL ELF (.elf)"
|
||||||
self._full_path: str = "" # absolute path (internal)
|
self._full_path: str = "" # absolute path (internal)
|
||||||
self._relative_path: str = "" # relative path (synced to Config)
|
self._relative_path: str = "" # relative path (synced to Config)
|
||||||
self._display_text = ctk.StringVar(value="")
|
self._display_text = ctk.StringVar(value="")
|
||||||
@@ -574,7 +594,7 @@ class FileSelector(ctk.CTkFrame):
|
|||||||
from tkinter import filedialog
|
from tkinter import filedialog
|
||||||
|
|
||||||
file_path = filedialog.askopenfilename(
|
file_path = filedialog.askopenfilename(
|
||||||
title="Select file",
|
title=f"Select {self._hint_text}",
|
||||||
filetypes=self._file_types,
|
filetypes=self._file_types,
|
||||||
)
|
)
|
||||||
if file_path:
|
if file_path:
|
||||||
@@ -851,6 +871,7 @@ class TftpSubStepFrame(ctk.CTkFrame):
|
|||||||
|
|
||||||
self._collapsed = True
|
self._collapsed = True
|
||||||
self._pulse_job: str | None = None
|
self._pulse_job: str | None = None
|
||||||
|
self._skipped: set[int] = set()
|
||||||
self._substep_names = substep_names or ["Upload", "Download Verify", "CRC Check", "Reboot", "Boot Verify"]
|
self._substep_names = substep_names or ["Upload", "Download Verify", "CRC Check", "Reboot", "Boot Verify"]
|
||||||
self._num_steps = len(self._substep_names)
|
self._num_steps = len(self._substep_names)
|
||||||
|
|
||||||
@@ -916,6 +937,8 @@ class TftpSubStepFrame(ctk.CTkFrame):
|
|||||||
|
|
||||||
pct_text = f"{pct}%" if pct >= 0 else ""
|
pct_text = f"{pct}%" if pct >= 0 else ""
|
||||||
for i, item in enumerate(self._sub_items):
|
for i, item in enumerate(self._sub_items):
|
||||||
|
if i in self._skipped:
|
||||||
|
continue # leave skipped items untouched
|
||||||
if i < idx:
|
if i < idx:
|
||||||
item["dot"].configure(text="●", text_color=_C_DONE)
|
item["dot"].configure(text="●", text_color=_C_DONE)
|
||||||
item["label"].configure(text_color=_C_DONE)
|
item["label"].configure(text_color=_C_DONE)
|
||||||
@@ -955,6 +978,18 @@ class TftpSubStepFrame(ctk.CTkFrame):
|
|||||||
item["label"].configure(text_color=_C_ERROR)
|
item["label"].configure(text_color=_C_ERROR)
|
||||||
self._stop_pulse()
|
self._stop_pulse()
|
||||||
|
|
||||||
|
def set_step_skipped(self, phase: str) -> None:
|
||||||
|
"""Mark a specific step as skipped (warning color, Skip text)."""
|
||||||
|
idx = self._substep_names.index(phase) if phase in self._substep_names else -1
|
||||||
|
if idx < 0:
|
||||||
|
return
|
||||||
|
self._skipped.add(idx)
|
||||||
|
item = self._sub_items[idx]
|
||||||
|
item["dot"].configure(text="◌", text_color=WARNING_COLOR)
|
||||||
|
item["label"].configure(text_color=WARNING_COLOR)
|
||||||
|
item["pct"].configure(text="Skip", text_color=WARNING_COLOR)
|
||||||
|
self._stop_pulse()
|
||||||
|
|
||||||
def _start_pulse(self, idx: int) -> None:
|
def _start_pulse(self, idx: int) -> None:
|
||||||
"""Start pulsing the running sub-step."""
|
"""Start pulsing the running sub-step."""
|
||||||
_LIGHT = RUNNING_FADE_LIGHT
|
_LIGHT = RUNNING_FADE_LIGHT
|
||||||
@@ -979,6 +1014,7 @@ class TftpSubStepFrame(ctk.CTkFrame):
|
|||||||
|
|
||||||
def reset(self) -> None:
|
def reset(self) -> None:
|
||||||
self._stop_pulse()
|
self._stop_pulse()
|
||||||
|
self._skipped.clear()
|
||||||
_C = ACCENT_PENDING
|
_C = ACCENT_PENDING
|
||||||
for i, item in enumerate(self._sub_items):
|
for i, item in enumerate(self._sub_items):
|
||||||
item["dot"].configure(text="○", text_color=_C)
|
item["dot"].configure(text="○", text_color=_C)
|
||||||
|
|||||||
+22
-10
@@ -137,6 +137,7 @@ def check_uart_available(
|
|||||||
port: str,
|
port: str,
|
||||||
baudrate: int = 115200,
|
baudrate: int = 115200,
|
||||||
timeout: float = 3.0,
|
timeout: float = 3.0,
|
||||||
|
monitor: UartMonitor | None = None,
|
||||||
) -> tuple[bool, str]:
|
) -> tuple[bool, str]:
|
||||||
"""Check whether the UART serial port is available and readable.
|
"""Check whether the UART serial port is available and readable.
|
||||||
|
|
||||||
@@ -145,19 +146,24 @@ def check_uart_available(
|
|||||||
considered available (the device may simply not be outputting yet).
|
considered available (the device may simply not be outputting yet).
|
||||||
If the port cannot be opened at all, it is unavailable.
|
If the port cannot be opened at all, it is unavailable.
|
||||||
|
|
||||||
|
If *monitor* is provided and currently running, the function
|
||||||
|
skips the port-open test (which would fail on Windows with
|
||||||
|
Error 13 due to exclusive access) and returns True immediately.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
||||||
baudrate: Baud rate for serial communication.
|
baudrate: Baud rate for serial communication.
|
||||||
timeout: Seconds to wait for data after opening.
|
timeout: Seconds to wait for data after opening.
|
||||||
|
monitor: Optional running UartMonitor — if active, reuse it.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Tuple of (is_available, reason_string).
|
Tuple of (is_available, reason_string).
|
||||||
- (True, "Port open") — port is usable
|
|
||||||
- (True, "Port open, no data yet") — port open, no data currently
|
|
||||||
- (False, "No serial ports detected") — system has no serial ports
|
|
||||||
- (False, "Port not in detected ports") — configured port missing
|
|
||||||
- (False, "Failed to open: <error>") — port open error
|
|
||||||
"""
|
"""
|
||||||
|
# Reuse existing monitor when available (avoids Error 13 on Windows)
|
||||||
|
if monitor is not None and monitor.is_running():
|
||||||
|
return True, "Already monitoring"
|
||||||
|
|
||||||
|
# 1. Quick port list check
|
||||||
# 1. Quick port list check
|
# 1. Quick port list check
|
||||||
ports = detect_serial_ports()
|
ports = detect_serial_ports()
|
||||||
if not ports:
|
if not ports:
|
||||||
@@ -223,25 +229,31 @@ def test_serial_version(
|
|||||||
port: str,
|
port: str,
|
||||||
baudrate: int = 115200,
|
baudrate: int = 115200,
|
||||||
timeout: float = 5.0,
|
timeout: float = 5.0,
|
||||||
|
monitor: UartMonitor | None = None,
|
||||||
) -> tuple[bool, str, str]:
|
) -> tuple[bool, str, str]:
|
||||||
"""Test serial port by sending 'ver()' command and parsing response.
|
"""Test serial port by sending 'ver()' command and parsing response.
|
||||||
|
|
||||||
Sends 'ver\\r\\n' to the serial port and reads the response.
|
If *monitor* is running, skips opening a new connection and
|
||||||
Attempts to parse version string from the response.
|
returns the latest parsed version from the monitor instead.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
||||||
baudrate: Baud rate for serial communication.
|
baudrate: Baud rate for serial communication.
|
||||||
timeout: Read timeout in seconds.
|
timeout: Read timeout in seconds.
|
||||||
|
monitor: Optional running UartMonitor — if active, reuse it.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Tuple of (success, message, version_string).
|
Tuple of (success, message, version_string).
|
||||||
- (True, "Version: x.y.z", "x.y.z") — version detected
|
|
||||||
- (True, "Port responded", "") — port responded but no version
|
|
||||||
- (False, "Error message", "") — error occurred
|
|
||||||
"""
|
"""
|
||||||
import serial
|
import serial
|
||||||
|
|
||||||
|
# Reuse existing monitor when available (avoids Error 13 on Windows)
|
||||||
|
if monitor is not None and monitor.is_running():
|
||||||
|
info = monitor.latest_info
|
||||||
|
if info and info.version:
|
||||||
|
return True, f"Version: {info.version}", info.version
|
||||||
|
return True, "Port monitored, no version parsed yet", ""
|
||||||
|
|
||||||
try:
|
try:
|
||||||
with serial.Serial(port, baudrate, timeout=timeout) as ser:
|
with serial.Serial(port, baudrate, timeout=timeout) as ser:
|
||||||
ser.reset_input_buffer()
|
ser.reset_input_buffer()
|
||||||
|
|||||||
+14
-7
@@ -60,7 +60,8 @@ TOOL_ALIASES: dict[str, list[str]] = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
# Product subdirectories under xilinx_root to scan (in preference order)
|
# Product subdirectories under xilinx_root to scan (in preference order)
|
||||||
_PRODUCT_DIRS = ("Vitis", "Vivado")
|
# SDK covers pre-2019 installations where xsct lived in SDK/*/bin/
|
||||||
|
_PRODUCT_DIRS = ("Vitis", "Vivado", "SDK")
|
||||||
|
|
||||||
|
|
||||||
# ── Version helpers ──────────────────────────────────────────────────
|
# ── Version helpers ──────────────────────────────────────────────────
|
||||||
@@ -163,9 +164,9 @@ def _find_executable(
|
|||||||
"""Find an executable by name using PATH, Vitis, and Xilinx root.
|
"""Find an executable by name using PATH, Vitis, and Xilinx root.
|
||||||
|
|
||||||
Search order:
|
Search order:
|
||||||
1. System PATH (shutil.which)
|
1. System PATH (shutil.which) — tries bare name, .exe, .bat on Windows
|
||||||
2. vitis_path/bin (legacy config)
|
2. vitis_path/bin (legacy config)
|
||||||
3. xilinx_root/Vitis/*/bin and Vivado/*/bin — pick NEWEST version
|
3. xilinx_root/Vitis/*/bin, Vivado/*/bin, SDK/*/bin — pick NEWEST version
|
||||||
4. Common install paths (cross-platform)
|
4. Common install paths (cross-platform)
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
@@ -179,14 +180,20 @@ def _find_executable(
|
|||||||
"""
|
"""
|
||||||
exec_suffix = ".exe" if system == "windows" else ""
|
exec_suffix = ".exe" if system == "windows" else ""
|
||||||
|
|
||||||
# 1. System PATH
|
# 1. System PATH — search without explicit extension first
|
||||||
|
# so Windows PATHEXT resolves .bat/.exe/.cmd automatically
|
||||||
|
found = shutil.which(exec_name)
|
||||||
|
if not found and exec_suffix:
|
||||||
found = shutil.which(f"{exec_name}{exec_suffix}")
|
found = shutil.which(f"{exec_name}{exec_suffix}")
|
||||||
|
if not found:
|
||||||
|
found = shutil.which(f"{exec_name}.bat")
|
||||||
if found:
|
if found:
|
||||||
return found
|
return found
|
||||||
|
|
||||||
# 2. Legacy vitis_path/bin
|
# 2. Legacy vitis_path/bin
|
||||||
if vitis_path:
|
if vitis_path:
|
||||||
candidate = Path(vitis_path) / "bin" / f"{exec_name}{exec_suffix}"
|
for sfx in (".bat", ".exe") if system == "windows" else (exec_suffix,):
|
||||||
|
candidate = Path(vitis_path) / "bin" / f"{exec_name}{sfx}"
|
||||||
if candidate.is_file():
|
if candidate.is_file():
|
||||||
return str(candidate)
|
return str(candidate)
|
||||||
|
|
||||||
@@ -203,8 +210,8 @@ def _find_executable(
|
|||||||
if system == "windows":
|
if system == "windows":
|
||||||
common: list[str] = []
|
common: list[str] = []
|
||||||
for sfx in (".bat", ".exe"):
|
for sfx in (".bat", ".exe"):
|
||||||
for ver in ("2023.2", "2022.2"):
|
for ver in ("2023.2", "2022.2", "2018.3"):
|
||||||
for prod in ("Vitis", "Vivado"):
|
for prod in ("Vitis", "Vivado", "SDK"):
|
||||||
common.append(f"C:/Xilinx/{prod}/{ver}/bin/{exec_name}{sfx}")
|
common.append(f"C:/Xilinx/{prod}/{ver}/bin/{exec_name}{sfx}")
|
||||||
else:
|
else:
|
||||||
common = [
|
common = [
|
||||||
|
|||||||
+29
-25
@@ -53,6 +53,7 @@ class ZynqCheckResult:
|
|||||||
pl_detected: bool
|
pl_detected: bool
|
||||||
hw_server_url: str
|
hw_server_url: str
|
||||||
error: str = ""
|
error: str = ""
|
||||||
|
hw_server_pid: int | None = None
|
||||||
|
|
||||||
|
|
||||||
# ── Patterns ──────────────────────────────────────────────────────────
|
# ── Patterns ──────────────────────────────────────────────────────────
|
||||||
@@ -104,12 +105,14 @@ def check_zynq_jtag(
|
|||||||
|
|
||||||
# ── 2. Start hw_server ─────────────────────────────────────
|
# ── 2. Start hw_server ─────────────────────────────────────
|
||||||
hw_proc: subprocess.Popen | None = None
|
hw_proc: subprocess.Popen | None = None
|
||||||
|
hw_pid: int | None = None
|
||||||
try:
|
try:
|
||||||
hw_proc = subprocess.Popen(
|
hw_proc = subprocess.Popen(
|
||||||
build_tool_command(hw_server_path),
|
build_tool_command(hw_server_path),
|
||||||
stdout=subprocess.DEVNULL,
|
stdout=subprocess.DEVNULL,
|
||||||
stderr=subprocess.DEVNULL,
|
stderr=subprocess.DEVNULL,
|
||||||
)
|
)
|
||||||
|
hw_pid = hw_proc.pid
|
||||||
# Wait up to 10s for hw_server to bind
|
# Wait up to 10s for hw_server to bind
|
||||||
started = False
|
started = False
|
||||||
for _ in range(100):
|
for _ in range(100):
|
||||||
@@ -118,14 +121,14 @@ def check_zynq_jtag(
|
|||||||
started = True
|
started = True
|
||||||
break
|
break
|
||||||
if not started:
|
if not started:
|
||||||
return _fail("hw_server did not bind port 3121 within 10s", hw_server_url)
|
return _fail("hw_server did not bind port 3121 within 10s", hw_server_url, pid=hw_pid)
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
return _fail(f"Failed to start hw_server: {e}", hw_server_url)
|
return _fail(f"Failed to start hw_server: {e}", hw_server_url)
|
||||||
|
|
||||||
if callback:
|
if callback:
|
||||||
callback("progress", "Scanning JTAG chain...")
|
callback("progress", "Scanning JTAG chain...")
|
||||||
|
|
||||||
# ── 3. Run Vivado TCL (stream output to callback) ──────────
|
# ── 3. Run Vivado TCL ──────────────────────────────────────
|
||||||
tcl = _build_jtag_query_tcl()
|
tcl = _build_jtag_query_tcl()
|
||||||
python_timeout = config.step_timeouts.get("check_env", 120)
|
python_timeout = config.step_timeouts.get("check_env", 120)
|
||||||
|
|
||||||
@@ -141,39 +144,35 @@ def check_zynq_jtag(
|
|||||||
|
|
||||||
output_lines: list[str] = []
|
output_lines: list[str] = []
|
||||||
try:
|
try:
|
||||||
proc = subprocess.Popen(
|
cmd = build_tool_command(vivado_path, *vivado_args)
|
||||||
build_tool_command(vivado_path, *vivado_args),
|
# Use run() with communicate() under the hood — reliably
|
||||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
# drains pipes on Windows even when .bat wrappers are involved
|
||||||
|
result = subprocess.run(
|
||||||
|
cmd,
|
||||||
|
capture_output=True,
|
||||||
text=True,
|
text=True,
|
||||||
|
timeout=python_timeout,
|
||||||
)
|
)
|
||||||
# Read line by line so user sees real-time progress
|
output = result.stdout + result.stderr
|
||||||
deadline = time.monotonic() + python_timeout
|
for line in output.splitlines():
|
||||||
while True:
|
stripped = line.strip()
|
||||||
line = proc.stdout.readline() if proc.stdout else ""
|
if stripped:
|
||||||
if not line:
|
output_lines.append(stripped)
|
||||||
if proc.poll() is not None:
|
if callback:
|
||||||
break
|
callback("progress", stripped)
|
||||||
if time.monotonic() > deadline:
|
except subprocess.TimeoutExpired:
|
||||||
proc.kill()
|
|
||||||
proc.wait()
|
|
||||||
return _fail(
|
return _fail(
|
||||||
f"JTAG scan timed out after {python_timeout}s",
|
f"JTAG scan timed out after {python_timeout}s",
|
||||||
hw_server_url,
|
hw_server_url,
|
||||||
|
pid=hw_pid,
|
||||||
)
|
)
|
||||||
time.sleep(0.1)
|
|
||||||
continue
|
|
||||||
line = line.rstrip("\n\r")
|
|
||||||
output_lines.append(line)
|
|
||||||
if callback and line.strip():
|
|
||||||
callback("progress", line)
|
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
return _fail(f"Failed to run Vivado: {e}", hw_server_url)
|
return _fail(f"Failed to run Vivado: {e}", hw_server_url, pid=hw_pid)
|
||||||
finally:
|
finally:
|
||||||
try:
|
try:
|
||||||
Path(tcl).unlink(missing_ok=True)
|
Path(tcl).unlink(missing_ok=True)
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
# Clean up Vivado temp files we redirected
|
|
||||||
for f in ("vivado_jtag.jou", "vivado_jtag.log"):
|
for f in ("vivado_jtag.jou", "vivado_jtag.log"):
|
||||||
try:
|
try:
|
||||||
(Path(tmp) / f).unlink(missing_ok=True)
|
(Path(tmp) / f).unlink(missing_ok=True)
|
||||||
@@ -206,6 +205,7 @@ def check_zynq_jtag(
|
|||||||
ps_detected=False, pl_detected=False,
|
ps_detected=False, pl_detected=False,
|
||||||
hw_server_url=hw_server_url,
|
hw_server_url=hw_server_url,
|
||||||
error=msg,
|
error=msg,
|
||||||
|
hw_server_pid=hw_pid,
|
||||||
)
|
)
|
||||||
|
|
||||||
ps_detected = any(d.is_arm_dap for d in jtag_chain)
|
ps_detected = any(d.is_arm_dap for d in jtag_chain)
|
||||||
@@ -217,16 +217,18 @@ def check_zynq_jtag(
|
|||||||
ps_detected=ps_detected, pl_detected=pl_detected,
|
ps_detected=ps_detected, pl_detected=pl_detected,
|
||||||
hw_server_url=hw_server_url,
|
hw_server_url=hw_server_url,
|
||||||
error="",
|
error="",
|
||||||
|
hw_server_pid=hw_pid,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
# ── Helpers ───────────────────────────────────────────────────────────
|
# ── Helpers ───────────────────────────────────────────────────────────
|
||||||
|
|
||||||
def _fail(error: str, url: str) -> ZynqCheckResult:
|
def _fail(error: str, url: str, pid: int | None = None) -> ZynqCheckResult:
|
||||||
return ZynqCheckResult(
|
return ZynqCheckResult(
|
||||||
devices=[], jtag_chain=[], is_present=False,
|
devices=[], jtag_chain=[], is_present=False,
|
||||||
ps_detected=False, pl_detected=False,
|
ps_detected=False, pl_detected=False,
|
||||||
hw_server_url=url, error=error,
|
hw_server_url=url, error=error,
|
||||||
|
hw_server_pid=pid,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -257,7 +259,7 @@ def _extract_relevant(output: str) -> str:
|
|||||||
def _build_jtag_query_tcl() -> str:
|
def _build_jtag_query_tcl() -> str:
|
||||||
"""Build Vivado TCL script file for JTAG device enumeration."""
|
"""Build Vivado TCL script file for JTAG device enumeration."""
|
||||||
content = """\
|
content = """\
|
||||||
open_hw_manager
|
open_hw
|
||||||
connect_hw_server
|
connect_hw_server
|
||||||
open_hw_target
|
open_hw_target
|
||||||
puts "===JTAG_START==="
|
puts "===JTAG_START==="
|
||||||
@@ -267,6 +269,8 @@ foreach dev [get_hw_devices] {
|
|||||||
puts "DEVICE:$name IDCODE:$idcode"
|
puts "DEVICE:$name IDCODE:$idcode"
|
||||||
}
|
}
|
||||||
puts "===JTAG_END==="
|
puts "===JTAG_END==="
|
||||||
|
close_hw
|
||||||
|
disconnect_hw_server
|
||||||
quit
|
quit
|
||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
|
|||||||
Reference in New Issue
Block a user