Compare commits
41
Commits
5ce544cd3c
...
v0.0.1
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c0f7219d30 | ||
|
|
4da03e97d8 | ||
|
|
c9eae220f0 | ||
|
|
e09f16a3aa | ||
|
|
47ea614ba2 | ||
|
|
13ed0d89f8 | ||
|
|
0d66f65ef2 | ||
|
|
4c01acdf69 | ||
|
|
c0cac5b2c5 | ||
|
|
5ed195e2d6 | ||
|
|
639f2e3ce6 | ||
|
|
9502bba636 | ||
|
|
9238cbc63e | ||
|
|
f330f8c276 | ||
|
|
04ad61e694 | ||
|
|
4ca15c1190 | ||
|
|
4453fc8e12 | ||
|
|
799da5b2f3 | ||
|
|
7e8427d522 | ||
|
|
66c04da5e0 | ||
|
|
954aed7c0d | ||
|
|
474e3f122b | ||
|
|
dd35ab0846 | ||
|
|
86d65641cc | ||
|
|
7716171aeb | ||
|
|
91417f609e | ||
|
|
72d0eceb8d | ||
|
|
1619dc194f | ||
|
|
267cae93b6 | ||
|
|
17c8f2d68b | ||
|
|
61f0e6cd2b | ||
|
|
6d686964e2 | ||
|
|
da94cc6cdc | ||
|
|
70c88815cc | ||
|
|
32b201cf8e | ||
|
|
21acc16c8c | ||
|
|
475042c85c | ||
|
|
f6cf5de96f | ||
|
|
9b07a4d5ca | ||
|
|
84142bb229 | ||
|
|
f04fbe2130 |
+15
-1
@@ -33,8 +33,22 @@ htmlcov/
|
||||
# ── Opencode (internal tooling) ──────────────────────────────
|
||||
.opencode/
|
||||
|
||||
# ── Config overrides (user-specific) ─────────────────────────
|
||||
# ── Config (user-specific, auto-generated if missing) ─────────
|
||||
config.yaml
|
||||
config/zynq_flasher.yaml
|
||||
|
||||
# ── Logs ─────────────────────────────────────────────────────
|
||||
*.log
|
||||
|
||||
# ── Vivado / Xilinx temp files ────────────────────────────────
|
||||
vivado.jou
|
||||
vivado.log
|
||||
vivado_*.backup.jou
|
||||
vivado_*.backup.log
|
||||
vivado_*.str
|
||||
hs_err_pid*.log
|
||||
webtalk*.jou
|
||||
webtalk*.log
|
||||
usage_statistics_*
|
||||
xicom_zynq_bin*
|
||||
*.pcapng
|
||||
|
||||
@@ -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
|
||||
-25
@@ -1,25 +0,0 @@
|
||||
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
|
||||
+22
-44
@@ -11,14 +11,13 @@ Compatible with Xilinx 2018.3 — 2023.2.
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
from config_manager import Config
|
||||
from vitis_checker import get_xsct_path
|
||||
from vitis_checker import get_xsct_path, get_vivado_path
|
||||
|
||||
# ── Types ───────────────────────────────────────────────────────────
|
||||
|
||||
@@ -38,33 +37,13 @@ ProgressCallback = Callable[[str, str], None] | None
|
||||
# ── Helpers ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _get_vivado_path(vitis_path: str) -> str | None:
|
||||
"""Find the Vivado executable.
|
||||
def _tcl_path(path: str | Path) -> str:
|
||||
"""Convert a filesystem path to TCL-safe format.
|
||||
|
||||
Uses shutil.which() for PATH resolution, then common install paths.
|
||||
|
||||
Args:
|
||||
vitis_path: Optional Vitis installation directory.
|
||||
|
||||
Returns:
|
||||
Path to vivado executable, or None.
|
||||
TCL interprets backslashes as escape sequences (\\f, \\t, \\n).
|
||||
Forward slashes work on all platforms including Windows.
|
||||
"""
|
||||
found = shutil.which("vivado")
|
||||
if found:
|
||||
return found
|
||||
if vitis_path:
|
||||
candidate = Path(vitis_path) / "bin" / "vivado"
|
||||
if candidate.is_file():
|
||||
return str(candidate)
|
||||
common = [
|
||||
"/data/xilinx/Vivado/2023.2/bin/vivado",
|
||||
"/opt/xilinx/bin/vivado",
|
||||
os.path.expanduser("~/Xilinx/Vivado/2023.2/bin/vivado"),
|
||||
]
|
||||
for p in common:
|
||||
if os.path.isfile(p):
|
||||
return p
|
||||
return None
|
||||
return str(path).replace("\\", "/")
|
||||
|
||||
|
||||
def _find_ps7_init_tcl(config: Config) -> str:
|
||||
@@ -167,16 +146,15 @@ def program_bitstream(
|
||||
|
||||
# Try xsct first (preferred)
|
||||
xsct = get_xsct_path(
|
||||
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "",
|
||||
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
|
||||
xilinx_root=getattr(config, "xilinx_path", ""),
|
||||
)
|
||||
if xsct:
|
||||
timeout = config.step_timeouts.get("bitstream", 60)
|
||||
return _program_with_xsct(xsct, bit_path, callback, ps7_tcl, timeout)
|
||||
|
||||
# Fallback to Vivado
|
||||
vivado = _get_vivado_path(
|
||||
config.vitis_path if hasattr(config, 'vitis_path') else ""
|
||||
vivado = get_vivado_path(
|
||||
xilinx_root=getattr(config, "xilinx_path", ""),
|
||||
)
|
||||
if vivado:
|
||||
return _program_with_vivado(vivado, bit_path, callback)
|
||||
@@ -218,11 +196,11 @@ def _program_with_xsct(
|
||||
lines = ["connect"]
|
||||
if ps7_init_tcl and Path(ps7_init_tcl).exists():
|
||||
lines.append("targets 1")
|
||||
lines.append(f'source "{ps7_init_tcl}"')
|
||||
lines.append(f'source "{_tcl_path(ps7_init_tcl)}"')
|
||||
lines.append("ps7_init")
|
||||
if callback:
|
||||
callback("progress", "PS initialized (clocks, DDR, MIO)")
|
||||
lines.append(f'fpga -file "{bit_path}"')
|
||||
lines.append(f'fpga -file {{{_tcl_path(bit_path)}}}')
|
||||
if ps7_init_tcl and Path(ps7_init_tcl).exists():
|
||||
lines.append("ps7_post_config")
|
||||
if callback:
|
||||
@@ -293,12 +271,14 @@ def _program_with_vivado(
|
||||
callback("progress", "Using Vivado Hardware Manager...")
|
||||
|
||||
tcl_content = f"""
|
||||
open_hw_manager
|
||||
open_hw
|
||||
connect_hw_server
|
||||
open_hw_target
|
||||
current_hw_device [lindex [get_hw_devices] 0]
|
||||
program_hw_device -file "{bit_path}"
|
||||
refresh_hw_device
|
||||
close_hw
|
||||
disconnect_hw_server
|
||||
quit
|
||||
"""
|
||||
tcl_path = bit_path.parent / "temp_program.tcl"
|
||||
@@ -364,8 +344,7 @@ def run_elf(
|
||||
callback("start", f"Running ELF: {elf_path.name}...")
|
||||
|
||||
xsct = get_xsct_path(
|
||||
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "",
|
||||
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
|
||||
xilinx_root=getattr(config, "xilinx_path", ""),
|
||||
)
|
||||
if not xsct:
|
||||
return BitstreamResult(
|
||||
@@ -413,13 +392,13 @@ def _run_elf_via_fsbl(
|
||||
tcl = "\n".join([
|
||||
"connect",
|
||||
"targets 2",
|
||||
"dow \"%s\"" % fsbl_path,
|
||||
"dow {%s}" % _tcl_path(fsbl_path),
|
||||
"puts \"FSBL downloaded\"",
|
||||
"con",
|
||||
"after 3000",
|
||||
"stop",
|
||||
"puts \"FSBL done, CPU halted\"",
|
||||
"dow \"%s\"" % elf_path,
|
||||
"dow {%s}" % _tcl_path(elf_path),
|
||||
"mwr 0xF800025C 0x00000001",
|
||||
"con",
|
||||
"puts \"App running\"",
|
||||
@@ -493,7 +472,7 @@ def _run_elf_direct(
|
||||
tcl = "\n".join([
|
||||
"connect",
|
||||
"targets 2",
|
||||
"dow \"%s\"" % elf_path,
|
||||
"dow {%s}" % _tcl_path(elf_path),
|
||||
"mwr 0xF800025C 0x00000001",
|
||||
"con",
|
||||
"exit",
|
||||
@@ -579,8 +558,7 @@ def full_bitstream_program(
|
||||
message=f"FSBL not found: {fsbl_path or '(not configured)'}")]
|
||||
|
||||
xsct = get_xsct_path(
|
||||
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "",
|
||||
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
|
||||
xilinx_root=getattr(config, "xilinx_path", ""),
|
||||
)
|
||||
if not xsct:
|
||||
return [BitstreamResult(step="bitstream", success=False,
|
||||
@@ -653,18 +631,18 @@ def full_bitstream_program(
|
||||
"rst -processor",
|
||||
"after 500",
|
||||
'puts "INFO: Downloading FSBL..."',
|
||||
'dow "%s"' % fsbl_path,
|
||||
'dow {%s}' % _tcl_path(fsbl_path),
|
||||
'puts "INFO: Running FSBL for hardware initialization..."',
|
||||
"con",
|
||||
"after 2000",
|
||||
"stop",
|
||||
'puts "INFO: Hardware initialization completed."',
|
||||
'puts "INFO: Downloading Bitstream..."',
|
||||
'fpga -f "%s"' % bit_path,
|
||||
'fpga -f {%s}' % _tcl_path(bit_path),
|
||||
"after 1000",
|
||||
'targets -set -nocase -filter {name =~ "arm*#0"}',
|
||||
'puts "INFO: Downloading Application ELF..."',
|
||||
'dow "%s"' % elf_path,
|
||||
'dow {%s}' % _tcl_path(elf_path),
|
||||
'puts "INFO: Executing Application..."',
|
||||
"con",
|
||||
'puts "SUCCESS: Zynq target is running the specified application."',
|
||||
|
||||
+43
-3
@@ -17,6 +17,7 @@ import yaml
|
||||
DEFAULT_CONFIG: dict[str, Any] = {
|
||||
"xilinx_path": "",
|
||||
"zynq_ip": "192.168.100.11",
|
||||
"zynq_part": "XC7Z100",
|
||||
"tftp_upload_name": "z7bin",
|
||||
"serial_port": "",
|
||||
"serial_baudrate": 115200,
|
||||
@@ -27,14 +28,16 @@ DEFAULT_CONFIG: dict[str, Any] = {
|
||||
"flash_type": "qspi-x4-single",
|
||||
"flash_model": "s25fl256s1",
|
||||
"erase_all": False,
|
||||
"download_verify": True,
|
||||
"firmware_bin_path": "",
|
||||
"reboot_timeout": 30,
|
||||
"ping_timeout": 5,
|
||||
"ping_count": 3,
|
||||
"uart_delay": 3,
|
||||
"inter_step_delay": 2,
|
||||
"boot_wait_delay": 30,
|
||||
"step_timeouts": {
|
||||
"check_env": 30,
|
||||
"check_env": 120,
|
||||
"flash_erase": 120,
|
||||
"flash_program": 600,
|
||||
"bitstream": 60,
|
||||
@@ -54,6 +57,7 @@ class Config:
|
||||
|
||||
xilinx_path: str = ""
|
||||
zynq_ip: str = "192.168.100.11"
|
||||
zynq_part: str = "XC7Z100"
|
||||
tftp_upload_name: str = "z7bin"
|
||||
serial_port: str = ""
|
||||
serial_baudrate: int = 115200
|
||||
@@ -64,12 +68,14 @@ class Config:
|
||||
flash_type: str = "qspi-x4-single"
|
||||
flash_model: str = "s25fl256s1"
|
||||
erase_all: bool = False
|
||||
download_verify: bool = True
|
||||
firmware_bin_path: str = ""
|
||||
reboot_timeout: int = 30
|
||||
ping_timeout: int = 5
|
||||
ping_count: int = 3
|
||||
uart_delay: int = 3
|
||||
inter_step_delay: int = 2
|
||||
boot_wait_delay: int = 30
|
||||
step_timeouts: dict[str, int] = field(default_factory=lambda: {})
|
||||
|
||||
# Internal: path to the config file on disk
|
||||
@@ -172,12 +178,14 @@ class Config:
|
||||
"flash_type": self.flash_type,
|
||||
"flash_model": self.flash_model,
|
||||
"erase_all": self.erase_all,
|
||||
"download_verify": self.download_verify,
|
||||
"firmware_bin_path": self._relative_path(self.firmware_bin_path),
|
||||
"reboot_timeout": self.reboot_timeout,
|
||||
"ping_timeout": self.ping_timeout,
|
||||
"ping_count": self.ping_count,
|
||||
"uart_delay": self.uart_delay,
|
||||
"inter_step_delay": self.inter_step_delay,
|
||||
"boot_wait_delay": self.boot_wait_delay,
|
||||
"step_timeouts": self.step_timeouts,
|
||||
}
|
||||
return data
|
||||
@@ -185,8 +193,11 @@ class Config:
|
||||
def _relative_path(self, path: str) -> str:
|
||||
"""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:
|
||||
path: Absolute path string.
|
||||
path: Absolute or relative path string.
|
||||
|
||||
Returns:
|
||||
Relative path string, or the original if conversion fails.
|
||||
@@ -196,7 +207,8 @@ class Config:
|
||||
try:
|
||||
target = Path(path)
|
||||
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):
|
||||
pass
|
||||
return path
|
||||
@@ -214,6 +226,34 @@ class Config:
|
||||
return Path()
|
||||
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
|
||||
def config_path(self) -> Path | None:
|
||||
"""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
|
||||
+62
-12
@@ -10,6 +10,7 @@ Compatible with Xilinx 2018.3 — 2023.2.
|
||||
from __future__ import annotations
|
||||
|
||||
import binascii
|
||||
import re
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
@@ -35,6 +36,27 @@ ProgressCallback = Callable[[str, str], None] | None
|
||||
|
||||
# ── Helpers ─────────────────────────────────────────────────────────
|
||||
|
||||
# Keywords in program_flash output that indicate failure even with rc=0
|
||||
_FAIL_KEYWORDS = [
|
||||
"failed", "cannot", "abort",
|
||||
"invalid", "unsupported", "timeout",
|
||||
]
|
||||
|
||||
|
||||
def _output_has_error(output: str) -> str:
|
||||
"""Check combined stdout+stderr for known failure keywords.
|
||||
|
||||
Returns the first matching line, or empty string if clean.
|
||||
"""
|
||||
lowered = output.lower()
|
||||
for kw in _FAIL_KEYWORDS:
|
||||
if kw in lowered:
|
||||
# Find the actual line
|
||||
for line in output.splitlines():
|
||||
if kw in line.lower():
|
||||
return line.strip()[:200]
|
||||
return ""
|
||||
|
||||
|
||||
def _run_command(
|
||||
cmd: list[str],
|
||||
@@ -113,8 +135,7 @@ def wipe_flash(
|
||||
FlashResult with status.
|
||||
"""
|
||||
flash_tool = get_program_flash_path(
|
||||
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "",
|
||||
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
|
||||
xilinx_root=getattr(config, "xilinx_path", ""),
|
||||
)
|
||||
if not flash_tool:
|
||||
return FlashResult(
|
||||
@@ -168,12 +189,27 @@ def wipe_flash(
|
||||
callback=callback,
|
||||
)
|
||||
|
||||
success = result.returncode == 0
|
||||
combined = result.stdout + result.stderr
|
||||
err_line = _output_has_error(combined)
|
||||
if err_line:
|
||||
return FlashResult(
|
||||
step="wipe",
|
||||
success=success,
|
||||
message="Flash erased successfully" if success else "Flash erase failed",
|
||||
output=(result.stdout + result.stderr)[:4000],
|
||||
success=False,
|
||||
message=f"Flash erase failed: {err_line}",
|
||||
output=combined[:4000],
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return FlashResult(
|
||||
step="wipe",
|
||||
success=False,
|
||||
message=f"Flash erase failed (exit {result.returncode})",
|
||||
output=combined[:4000],
|
||||
)
|
||||
return FlashResult(
|
||||
step="wipe",
|
||||
success=True,
|
||||
message="Flash erased successfully",
|
||||
output=combined[:4000],
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
return FlashResult(step="wipe", success=False, message="Flash erase timed out")
|
||||
@@ -229,8 +265,7 @@ def program_flash_bin(
|
||||
)
|
||||
|
||||
flash_tool = get_program_flash_path(
|
||||
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "",
|
||||
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
|
||||
xilinx_root=getattr(config, "xilinx_path", ""),
|
||||
)
|
||||
if not flash_tool:
|
||||
return FlashResult(
|
||||
@@ -254,12 +289,27 @@ def program_flash_bin(
|
||||
|
||||
try:
|
||||
result = _run_command(cmd, timeout=config.step_timeouts.get("flash_program", 600), callback=callback)
|
||||
success = result.returncode == 0
|
||||
combined = result.stdout + result.stderr
|
||||
err_line = _output_has_error(combined)
|
||||
if err_line:
|
||||
return FlashResult(
|
||||
step="program",
|
||||
success=success,
|
||||
message="Flash programmed successfully" if success else "Flash programming failed",
|
||||
output=(result.stdout + result.stderr)[:4000],
|
||||
success=False,
|
||||
message=f"Flash programming failed: {err_line}",
|
||||
output=combined[:4000],
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return FlashResult(
|
||||
step="program",
|
||||
success=False,
|
||||
message=f"Flash programming failed (exit {result.returncode})",
|
||||
output=combined[:4000],
|
||||
)
|
||||
return FlashResult(
|
||||
step="program",
|
||||
success=True,
|
||||
message="Flash programmed successfully",
|
||||
output=combined[:4000],
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
return FlashResult(step="program", success=False, message="Flash programming timed out (600s)")
|
||||
|
||||
+929
-102
File diff suppressed because it is too large
Load Diff
+318
-55
@@ -6,6 +6,7 @@ consistent styling from gui/styles.py.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import threading
|
||||
|
||||
import customtkinter as ctk
|
||||
@@ -392,26 +393,32 @@ class ProgressIndicator(ctk.CTkFrame):
|
||||
|
||||
def _create_widgets(self) -> None:
|
||||
"""Create the progress indicator UI."""
|
||||
self.grid_rowconfigure(0, weight=1)
|
||||
self.grid_rowconfigure(0, weight=0)
|
||||
self.grid_rowconfigure(1, weight=1)
|
||||
self.grid_columnconfigure(0, weight=1)
|
||||
|
||||
self._progress_bar = ctk.CTkProgressBar(
|
||||
self,
|
||||
variable=self._progress_var,
|
||||
corner_radius=CORNER_RADIUS,
|
||||
height=8,
|
||||
)
|
||||
self._progress_bar.grid(
|
||||
row=0, column=0, sticky="nsew", padx=PADDING
|
||||
)
|
||||
|
||||
# Percentage label on top
|
||||
self._percent_label = ctk.CTkLabel(
|
||||
self,
|
||||
text="0%",
|
||||
font=FONT_SMALL,
|
||||
font=(FONT_BODY[0], 13, "bold"),
|
||||
text_color=PRIMARY_COLOR,
|
||||
)
|
||||
self._percent_label.grid(
|
||||
row=0, column=0, sticky="ne", padx=PADDING, pady=PADDING_SMALL
|
||||
row=0, column=0, sticky="e", padx=PADDING, pady=(0, 6)
|
||||
)
|
||||
|
||||
# Progress bar below
|
||||
self._progress_bar = ctk.CTkProgressBar(
|
||||
self,
|
||||
variable=self._progress_var,
|
||||
corner_radius=6,
|
||||
height=10,
|
||||
progress_color=PRIMARY_COLOR,
|
||||
border_width=0,
|
||||
)
|
||||
self._progress_bar.grid(
|
||||
row=1, column=0, sticky="ew", padx=PADDING, pady=2
|
||||
)
|
||||
|
||||
def set_value(self, value: float) -> None:
|
||||
@@ -430,54 +437,91 @@ class ProgressIndicator(ctk.CTkFrame):
|
||||
|
||||
|
||||
class StatusDisplay(ctk.CTkFrame):
|
||||
"""Status display widget for operation results.
|
||||
"""Info display widget — persistent key-value file/device info.
|
||||
|
||||
Shows a status message with color-coded feedback.
|
||||
Shows aligned ``key : value`` lines for file metadata, Zynq
|
||||
part info, tool versions, etc.
|
||||
"""
|
||||
|
||||
def __init__(self, master, **kwargs):
|
||||
"""Initialize the status display.
|
||||
|
||||
Args:
|
||||
master: Parent widget.
|
||||
"""
|
||||
super().__init__(master, **kwargs)
|
||||
|
||||
self._info_items: list[tuple[str, str]] = []
|
||||
self._create_widgets()
|
||||
|
||||
def _create_widgets(self) -> None:
|
||||
"""Create the status display UI."""
|
||||
self._status_label = ctk.CTkLabel(
|
||||
self.grid_rowconfigure(0, weight=1)
|
||||
self.grid_columnconfigure(0, weight=1)
|
||||
|
||||
self._info_text = ctk.CTkTextbox(
|
||||
self,
|
||||
text="Ready",
|
||||
font=FONT_BODY,
|
||||
anchor="w",
|
||||
corner_radius=CORNER_RADIUS,
|
||||
state="disabled",
|
||||
wrap="word",
|
||||
)
|
||||
self._status_label.pack(
|
||||
fill="x", padx=PADDING, pady=PADDING_SMALL
|
||||
self._info_text.grid(
|
||||
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:
|
||||
"""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:
|
||||
message: Status message text.
|
||||
status: Status type for coloring ('info', 'success', 'error', 'warning').
|
||||
key: Label for the info line.
|
||||
value: Value string.
|
||||
"""
|
||||
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))
|
||||
for i, (k, _) in enumerate(self._info_items):
|
||||
if k == key:
|
||||
self._info_items[i] = (key, value)
|
||||
break
|
||||
else:
|
||||
self._info_items.append((key, value))
|
||||
self._refresh_info()
|
||||
|
||||
def clear_info(self) -> None:
|
||||
"""Remove all persistent info lines."""
|
||||
self._info_items.clear()
|
||||
self._info_text.configure(state="normal")
|
||||
self._info_text.delete("1.0", "end")
|
||||
self._info_text.configure(state="disabled")
|
||||
|
||||
def _refresh_info(self) -> None:
|
||||
"""Rebuild the info text block from current items."""
|
||||
self._info_text.configure(state="normal")
|
||||
self._info_text.delete("1.0", "end")
|
||||
if not self._info_items:
|
||||
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):
|
||||
"""File selector widget with browse button and path display.
|
||||
|
||||
Allows users to browse for files and displays the selected path.
|
||||
Stores the full absolute path internally but displays only the
|
||||
filename in the entry field. Supports relative paths: the
|
||||
``set_relative_path()`` / ``get_relative_path()`` methods work
|
||||
with paths relative to a given base directory.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@@ -500,7 +544,11 @@ class FileSelector(ctk.CTkFrame):
|
||||
|
||||
self._callback = callback
|
||||
self._file_types = file_types or [("All files", "*")]
|
||||
self._selected_path = ctk.StringVar(value="")
|
||||
self._hint_text = label_text.rstrip(":") # e.g. "FSBL ELF (.elf)"
|
||||
self._full_path: str = "" # absolute path (internal)
|
||||
self._relative_path: str = "" # relative path (synced to Config)
|
||||
self._display_text = ctk.StringVar(value="")
|
||||
self._suppress_callback: bool = False # True during construction
|
||||
|
||||
self._create_widgets(label_text)
|
||||
|
||||
@@ -518,10 +566,10 @@ class FileSelector(ctk.CTkFrame):
|
||||
)
|
||||
label.grid(row=0, column=0, sticky="w", padx=(PADDING, PADDING_SMALL))
|
||||
|
||||
# Path entry
|
||||
# Path entry — shows only the filename
|
||||
self._entry = ctk.CTkEntry(
|
||||
self,
|
||||
textvariable=self._selected_path,
|
||||
textvariable=self._display_text,
|
||||
font=FONT_MONO,
|
||||
state="readonly",
|
||||
)
|
||||
@@ -543,33 +591,60 @@ class FileSelector(ctk.CTkFrame):
|
||||
|
||||
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'}",
|
||||
title=f"Select {self._hint_text}",
|
||||
filetypes=self._file_types,
|
||||
)
|
||||
if file_path:
|
||||
self.set_path(file_path)
|
||||
|
||||
def set_path(self, path: str) -> None:
|
||||
"""Set the selected file path.
|
||||
"""Set the selected file path (absolute or relative).
|
||||
|
||||
Internally stores the absolute path and displays only the
|
||||
filename.
|
||||
|
||||
Args:
|
||||
path: Absolute path to the selected file.
|
||||
path: Absolute or relative path to the selected file.
|
||||
"""
|
||||
self._selected_path.set(path)
|
||||
if self._callback:
|
||||
self._full_path = path
|
||||
self._display_text.set(os.path.basename(path) if path else "")
|
||||
if self._callback and not self._suppress_callback:
|
||||
self._callback(path)
|
||||
|
||||
def set_relative_path(self, path: str) -> None:
|
||||
"""Set a relative path (from config).
|
||||
|
||||
Resolves to absolute for internal storage, shows only filename.
|
||||
|
||||
Args:
|
||||
path: Relative path string.
|
||||
"""
|
||||
self._relative_path = path
|
||||
# Resolve to absolute for internal use
|
||||
if path:
|
||||
self._full_path = path # caller should resolve before calling
|
||||
self._display_text.set(os.path.basename(path) if path else "")
|
||||
if self._callback and not self._suppress_callback:
|
||||
self._callback(path)
|
||||
|
||||
def get_path(self) -> str:
|
||||
"""Get the currently selected file path.
|
||||
"""Get the currently selected file path (absolute).
|
||||
|
||||
Returns:
|
||||
Selected file path string.
|
||||
"""
|
||||
return self._selected_path.get()
|
||||
return self._full_path
|
||||
|
||||
def get_relative_path(self) -> str:
|
||||
"""Get the currently selected relative path.
|
||||
|
||||
Returns:
|
||||
Relative path string, or empty string.
|
||||
"""
|
||||
return self._relative_path
|
||||
|
||||
|
||||
class LogDisplay(ctk.CTkFrame):
|
||||
@@ -613,6 +688,7 @@ class LogDisplay(ctk.CTkFrame):
|
||||
"""
|
||||
self._text_widget.configure(state="normal")
|
||||
self._text_widget.insert("end", message + "\n")
|
||||
self._text_widget.update_idletasks()
|
||||
self._text_widget.see("end")
|
||||
self._text_widget.configure(state="disabled")
|
||||
|
||||
@@ -643,15 +719,15 @@ class SubStepFrame(ctk.CTkFrame):
|
||||
"""Collapsible frame showing flash operation sub-steps with status.
|
||||
|
||||
Colors match StepHeader: gray pending, blue running (+pulse),
|
||||
green complete, red error. Default expanded.
|
||||
green complete, red error. Default collapsed.
|
||||
"""
|
||||
|
||||
def __init__(self, parent: ctk.CTkFrame) -> None:
|
||||
super().__init__(parent, fg_color="transparent")
|
||||
|
||||
self._collapsed = False
|
||||
self._collapsed = True
|
||||
self._toggle_btn = ctk.CTkButton(
|
||||
self, text="▼", width=22, height=22,
|
||||
self, text="▶", width=22, height=22,
|
||||
font=(FONT_SMALL[0], 9), command=self._toggle,
|
||||
)
|
||||
self._toggle_btn.grid(row=0, column=0, padx=(4, 4), pady=(2, 2), sticky="w")
|
||||
@@ -676,6 +752,16 @@ class SubStepFrame(ctk.CTkFrame):
|
||||
self.grid_columnconfigure(1, weight=1)
|
||||
self._pulse_job: str | None = None
|
||||
|
||||
# Apply initial collapsed state (hide sub-items)
|
||||
self._hide()
|
||||
|
||||
def _hide(self) -> None:
|
||||
"""Hide sub-items without toggling state."""
|
||||
for item in self._sub_items:
|
||||
item["dot"].grid_remove()
|
||||
item["label"].grid_remove()
|
||||
item["pct"].grid_remove()
|
||||
|
||||
def _toggle(self) -> None:
|
||||
self._collapsed = not self._collapsed
|
||||
self._toggle_btn.configure(text="▶" if self._collapsed else "▼")
|
||||
@@ -690,7 +776,7 @@ class SubStepFrame(ctk.CTkFrame):
|
||||
item["pct"].grid()
|
||||
|
||||
def set_expanded(self, expanded: bool) -> None:
|
||||
if expanded != self._collapsed:
|
||||
if expanded == self._collapsed:
|
||||
self._toggle()
|
||||
|
||||
def set_phase(self, phase: str, pct: int = -1) -> None:
|
||||
@@ -765,8 +851,179 @@ class SubStepFrame(ctk.CTkFrame):
|
||||
if not self._collapsed:
|
||||
self._toggle()
|
||||
self._toggle() # collapse then expand to reset
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════════
|
||||
# TftpSubStepFrame — collapsible sub-step progress for Step 4
|
||||
# ════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TftpSubStepFrame(ctk.CTkFrame):
|
||||
"""Collapsible frame showing TFTP workflow sub-steps with status.
|
||||
|
||||
Sub-steps: Upload, Download Verify, CRC Check, Reboot, Boot Verify.
|
||||
Colors match StepHeader: gray pending, blue running (+pulse),
|
||||
green complete, red error. Default collapsed.
|
||||
"""
|
||||
|
||||
def __init__(self, parent: ctk.CTkFrame, substep_names: list[str] | None = None) -> None:
|
||||
super().__init__(parent, fg_color="transparent")
|
||||
|
||||
self._collapsed = True
|
||||
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._num_steps = len(self._substep_names)
|
||||
|
||||
self._toggle_btn = ctk.CTkButton(
|
||||
self, text="▶", width=22, height=22,
|
||||
font=(FONT_SMALL[0], 9), command=self._toggle,
|
||||
)
|
||||
self._toggle_btn.grid(row=0, column=0, padx=(4, 4), pady=(2, 2), sticky="w")
|
||||
|
||||
self._header_label = ctk.CTkLabel(
|
||||
self, text=" → ".join(self._substep_names),
|
||||
font=FONT_SMALL,
|
||||
)
|
||||
self._header_label.grid(row=0, column=1, padx=(0, 4), pady=(2, 2), sticky="w")
|
||||
|
||||
self._sub_items: list[dict] = []
|
||||
for i, name in enumerate(self._substep_names):
|
||||
dot = ctk.CTkLabel(self, text="○", font=FONT_BODY, text_color=ACCENT_PENDING, anchor="w")
|
||||
label = ctk.CTkLabel(self, text=name, font=FONT_BODY, text_color=ACCENT_PENDING, anchor="w")
|
||||
pct = ctk.CTkLabel(self, text="", font=FONT_BODY, text_color=ACCENT_PENDING, anchor="e")
|
||||
dot.grid(row=i + 1, column=0, padx=(16, 2), pady=(2, 2), sticky="w")
|
||||
label.grid(row=i + 1, column=1, padx=(2, 4), pady=(2, 2), sticky="w")
|
||||
pct.grid(row=i + 1, column=2, padx=(4, 8), pady=(2, 2), sticky="e")
|
||||
self._sub_items.append({"dot": dot, "label": label, "pct": pct})
|
||||
|
||||
self.grid_columnconfigure(1, weight=1)
|
||||
|
||||
# Apply initial collapsed state (hide sub-items)
|
||||
self._hide()
|
||||
|
||||
def _hide(self) -> None:
|
||||
"""Hide sub-items without toggling state."""
|
||||
for item in self._sub_items:
|
||||
item["dot"].grid_remove()
|
||||
item["label"].grid_remove()
|
||||
item["pct"].grid_remove()
|
||||
|
||||
def _toggle(self) -> None:
|
||||
self._collapsed = not self._collapsed
|
||||
self._toggle_btn.configure(text="▶" if self._collapsed else "▼")
|
||||
for item in self._sub_items:
|
||||
if self._collapsed:
|
||||
item["dot"].grid_remove()
|
||||
item["label"].grid_remove()
|
||||
item["pct"].grid_remove()
|
||||
else:
|
||||
item["dot"].grid()
|
||||
item["label"].grid()
|
||||
item["pct"].grid()
|
||||
|
||||
def set_expanded(self, expanded: bool) -> None:
|
||||
if expanded == self._collapsed:
|
||||
self._toggle()
|
||||
|
||||
def set_phase(self, phase: str, pct: int = -1) -> None:
|
||||
_C_PENDING = ACCENT_PENDING
|
||||
_C_RUNNING = CIRCLE_RUNNING
|
||||
_C_DONE = ACCENT_COMPLETE
|
||||
|
||||
idx = self._substep_names.index(phase) if phase in self._substep_names else -1
|
||||
if idx < 0:
|
||||
return
|
||||
|
||||
pct_text = f"{pct}%" if pct >= 0 else ""
|
||||
for i, item in enumerate(self._sub_items):
|
||||
if i in self._skipped:
|
||||
continue # leave skipped items untouched
|
||||
if i < idx:
|
||||
item["dot"].configure(text="●", text_color=_C_DONE)
|
||||
item["label"].configure(text_color=_C_DONE)
|
||||
item["pct"].configure(text="OK", text_color=_C_DONE)
|
||||
elif i == idx:
|
||||
item["dot"].configure(text="◉", text_color=_C_RUNNING)
|
||||
item["label"].configure(text_color=_C_RUNNING)
|
||||
item["pct"].configure(text=pct_text, text_color=_C_RUNNING)
|
||||
else:
|
||||
item["dot"].configure(text="○", text_color=_C_PENDING)
|
||||
item["label"].configure(text_color=_C_PENDING)
|
||||
item["pct"].configure(text="", text_color=_C_PENDING)
|
||||
|
||||
if phase in self._substep_names and not self._pulse_job:
|
||||
self._start_pulse(idx)
|
||||
|
||||
def set_step_complete(self, phase: str) -> None:
|
||||
"""Mark a specific step as complete."""
|
||||
_C_DONE = ACCENT_COMPLETE
|
||||
idx = self._substep_names.index(phase) if phase in self._substep_names else -1
|
||||
if idx < 0:
|
||||
return
|
||||
item = self._sub_items[idx]
|
||||
item["dot"].configure(text="●", text_color=_C_DONE)
|
||||
item["label"].configure(text_color=_C_DONE)
|
||||
item["pct"].configure(text="OK", text_color=_C_DONE)
|
||||
self._stop_pulse()
|
||||
|
||||
def set_step_error(self, phase: str) -> None:
|
||||
"""Mark a specific step as error."""
|
||||
_C_ERROR = ACCENT_ERROR
|
||||
idx = self._substep_names.index(phase) if phase in self._substep_names else -1
|
||||
if idx < 0:
|
||||
return
|
||||
item = self._sub_items[idx]
|
||||
item["dot"].configure(text="✗", text_color=_C_ERROR)
|
||||
item["label"].configure(text_color=_C_ERROR)
|
||||
self._stop_pulse()
|
||||
|
||||
def 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:
|
||||
"""Start pulsing the running sub-step."""
|
||||
_LIGHT = RUNNING_FADE_LIGHT
|
||||
_DARK = RUNNING_FADE_DARK
|
||||
|
||||
def _pulse(state=[True]):
|
||||
if idx >= len(self._sub_items):
|
||||
return
|
||||
item = self._sub_items[idx]
|
||||
c = _LIGHT if state[0] else _DARK
|
||||
item["dot"].configure(text_color=c)
|
||||
item["label"].configure(text_color=c)
|
||||
state[0] = not state[0]
|
||||
self._pulse_job = self.after(PULSE_DURATION, lambda: _pulse(state))
|
||||
|
||||
_pulse()
|
||||
|
||||
def _stop_pulse(self) -> None:
|
||||
if self._pulse_job:
|
||||
self.after_cancel(self._pulse_job)
|
||||
self._pulse_job = None
|
||||
|
||||
def reset(self) -> None:
|
||||
self._stop_pulse()
|
||||
self._skipped.clear()
|
||||
_C = ACCENT_PENDING
|
||||
for i, item in enumerate(self._sub_items):
|
||||
item["dot"].configure(text="○", text_color=_C)
|
||||
item["label"].configure(text_color=_C)
|
||||
item["pct"].configure(text="", text_color=_C)
|
||||
if not self._collapsed:
|
||||
self._toggle()
|
||||
self._toggle() # collapse then expand to reset
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════════
|
||||
# UartMonitorWindow — standalone UART monitoring window
|
||||
@@ -927,7 +1184,7 @@ class UartMonitorWindow(ctk.CTkToplevel):
|
||||
ser.close()
|
||||
self.after(0, self._on_open_success)
|
||||
except Exception as e:
|
||||
self.after(0, lambda: self._show_unavailable(str(e)))
|
||||
self.after(0, lambda e=e: self._show_unavailable(str(e)))
|
||||
|
||||
threading.Thread(target=_bg_open, daemon=True).start()
|
||||
|
||||
@@ -1038,6 +1295,12 @@ class UartMonitorWindow(ctk.CTkToplevel):
|
||||
"""True if the UART monitor thread is currently running."""
|
||||
return self._monitor is not None and self._monitor.is_running()
|
||||
|
||||
def get_latest_info(self):
|
||||
"""Return latest parsed BootInfo from the UART monitor, or None."""
|
||||
if self._monitor:
|
||||
return self._monitor.latest_info
|
||||
return None
|
||||
|
||||
def _on_close(self) -> None:
|
||||
"""Cleanup UartMonitor and destroy window."""
|
||||
if self._monitor:
|
||||
|
||||
+270
-83
@@ -1,14 +1,15 @@
|
||||
"""Reboot manager for Zynq Flasher GUI.
|
||||
|
||||
Handles rebooting the Zynq device via SSH or serial commands.
|
||||
Handles rebooting the Zynq device via JTAG, UDP, or serial.
|
||||
Also provides UDP-based version query.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import platform
|
||||
import shutil
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable
|
||||
@@ -26,85 +27,75 @@ class RebootResult:
|
||||
output: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class UdpVersionResult:
|
||||
"""Result of a UDP version query."""
|
||||
|
||||
success: bool
|
||||
message: str
|
||||
version: str = ""
|
||||
|
||||
|
||||
ProgressCallback = Callable[[str, str], None] | None
|
||||
|
||||
# ── UDP helpers ──────────────────────────────────────────────────────
|
||||
|
||||
def reboot_via_ssh(
|
||||
config: Config,
|
||||
timeout: int | None = None,
|
||||
callback: ProgressCallback = None,
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device via SSH.
|
||||
|
||||
Sends a 'reboot' command over SSH to the Zynq device.
|
||||
def _udp_port_from_ip(ip: str) -> int:
|
||||
"""Derive UDP command port from the last octet of the IP address.
|
||||
|
||||
Port = last_octet - 3 (e.g., 192.168.100.11 → 8).
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
timeout: Seconds to wait for reboot acknowledgment.
|
||||
callback: Optional progress callback.
|
||||
ip: Zynq IP address string.
|
||||
|
||||
Returns:
|
||||
RebootResult with status.
|
||||
UDP port number, or -1 on parse failure.
|
||||
"""
|
||||
timeout = timeout or config.reboot_timeout
|
||||
ip = config.zynq_ip
|
||||
|
||||
if callback:
|
||||
callback("start", f"Sending reboot via SSH to {ip}...")
|
||||
|
||||
# Check ssh availability
|
||||
if not shutil.which("ssh"):
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="ssh command not found in PATH",
|
||||
method="ssh",
|
||||
)
|
||||
|
||||
# Try SSH reboot
|
||||
cmd = ["ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=no",
|
||||
f"root@{ip}", "reboot"]
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
)
|
||||
if callback:
|
||||
if result.returncode == 0:
|
||||
callback("progress", f"Reboot command sent to {ip}")
|
||||
else:
|
||||
callback("error", f"SSH reboot failed: {result.stderr.strip()}")
|
||||
octets = ip.strip().split(".")
|
||||
last = int(octets[3])
|
||||
return last - 3
|
||||
except (ValueError, IndexError):
|
||||
return -1
|
||||
|
||||
# Wait for device to come back
|
||||
if result.returncode == 0 and callback:
|
||||
callback("progress", f"Waiting {timeout}s for device to reboot...")
|
||||
|
||||
return RebootResult(
|
||||
success=result.returncode == 0,
|
||||
message="Reboot command sent" if result.returncode == 0 else f"SSH reboot failed: {result.stderr.strip()}",
|
||||
method="ssh",
|
||||
output=result.stdout,
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="SSH reboot timed out",
|
||||
method="ssh",
|
||||
)
|
||||
except FileNotFoundError:
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="ssh command not found",
|
||||
method="ssh",
|
||||
)
|
||||
def _send_udp_command(
|
||||
ip: str,
|
||||
command: bytes,
|
||||
timeout: float = 5.0,
|
||||
buffer_size: int = 4096,
|
||||
) -> tuple[bool, str]:
|
||||
"""Send a UDP command to the Zynq and receive the response.
|
||||
|
||||
Args:
|
||||
ip: Zynq IP address.
|
||||
command: Command bytes to send (e.g., b'reboot()').
|
||||
timeout: Receive timeout in seconds.
|
||||
buffer_size: Max receive buffer size.
|
||||
|
||||
Returns:
|
||||
(success, response_text).
|
||||
"""
|
||||
port = _udp_port_from_ip(ip)
|
||||
if port <= 0:
|
||||
return False, f"Invalid IP for UDP port derivation: {ip}"
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.settimeout(timeout)
|
||||
try:
|
||||
sock.sendto(command, (ip, port))
|
||||
data, addr = sock.recvfrom(buffer_size)
|
||||
return True, data.decode("utf-8", errors="replace").strip()
|
||||
except socket.timeout:
|
||||
return False, f"UDP timeout ({timeout}s)"
|
||||
except OSError as e:
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message=f"OS error: {e}",
|
||||
method="ssh",
|
||||
)
|
||||
return False, f"UDP error: {e}"
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
|
||||
# ── Reboot methods ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
def reboot_via_serial(
|
||||
@@ -113,7 +104,7 @@ def reboot_via_serial(
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device via serial port.
|
||||
|
||||
Sends a reboot command through the serial console.
|
||||
Sends reboot() through the serial console.
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
@@ -144,16 +135,16 @@ def reboot_via_serial(
|
||||
# Send Ctrl+C to break out of any running process
|
||||
ser.write(b"\x03")
|
||||
time.sleep(0.5)
|
||||
# Send reboot command
|
||||
ser.write(b"reboot\r\n")
|
||||
# Send reboot command as function call
|
||||
ser.write(b"reboot()\r\n")
|
||||
ser.reset_input_buffer()
|
||||
|
||||
if callback:
|
||||
callback("progress", "Reboot command sent via serial")
|
||||
callback("progress", "reboot() sent via serial")
|
||||
|
||||
return RebootResult(
|
||||
success=True,
|
||||
message="Reboot command sent via serial",
|
||||
message="reboot() sent via serial",
|
||||
method="serial",
|
||||
)
|
||||
except ImportError:
|
||||
@@ -176,6 +167,194 @@ def reboot_via_serial(
|
||||
)
|
||||
|
||||
|
||||
def reboot_via_jtag(
|
||||
config: Config,
|
||||
callback: ProgressCallback = None,
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device via JTAG system reset.
|
||||
|
||||
Connects to the Zynq over JTAG through hw_server and issues
|
||||
a full system reset (rst -system), then resumes ARM execution.
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
callback: Optional progress callback.
|
||||
|
||||
Returns:
|
||||
RebootResult with status.
|
||||
"""
|
||||
from vitis_checker import get_xsct_path
|
||||
|
||||
xsct = get_xsct_path(
|
||||
xilinx_root=getattr(config, "xilinx_path", ""),
|
||||
)
|
||||
if not xsct:
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="xsct not found — Xilinx tools not available",
|
||||
method="jtag",
|
||||
)
|
||||
|
||||
if callback:
|
||||
callback("start", f"JTAG reset via xsct ({xsct})...")
|
||||
|
||||
# Build TCL: system reset then resume ARM execution (no stop)
|
||||
tcl_script = """\
|
||||
puts "INFO: Connecting to hw_server..."
|
||||
connect
|
||||
after 500
|
||||
|
||||
puts "INFO: Selecting ARM target..."
|
||||
targets -set -nocase -filter {name =~ "arm*#0"}
|
||||
after 200
|
||||
|
||||
puts "INFO: Issuing system reset..."
|
||||
rst -system
|
||||
after 1500
|
||||
|
||||
puts "INFO: Resuming ARM execution..."
|
||||
catch { con }
|
||||
|
||||
puts "INFO: System reset complete — disconnecting..."
|
||||
disconnect
|
||||
exit
|
||||
"""
|
||||
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".tcl", delete=False,
|
||||
) as f:
|
||||
f.write(tcl_script)
|
||||
tcl_path = f.name
|
||||
|
||||
if callback:
|
||||
callback("progress", "Running JTAG system reset...")
|
||||
|
||||
from vitis_checker import build_tool_command
|
||||
result = subprocess.run(
|
||||
build_tool_command(xsct, tcl_path),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
os.unlink(tcl_path)
|
||||
|
||||
success = result.returncode == 0
|
||||
output = (result.stdout + result.stderr).strip()
|
||||
|
||||
if callback:
|
||||
callback(
|
||||
"complete" if success else "error",
|
||||
"JTAG reset " + ("succeeded" if success else "failed"),
|
||||
)
|
||||
|
||||
return RebootResult(
|
||||
success=success,
|
||||
message=f"JTAG system reset {'OK' if success else 'failed'}",
|
||||
method="jtag",
|
||||
output=output[:2000],
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
try:
|
||||
os.unlink(tcl_path)
|
||||
except OSError:
|
||||
pass
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message="JTAG reset timed out (30s)",
|
||||
method="jtag",
|
||||
)
|
||||
except OSError as e:
|
||||
try:
|
||||
os.unlink(tcl_path)
|
||||
except OSError:
|
||||
pass
|
||||
return RebootResult(
|
||||
success=False,
|
||||
message=f"JTAG reset error: {e}",
|
||||
method="jtag",
|
||||
)
|
||||
|
||||
|
||||
def reboot_via_udp(
|
||||
config: Config,
|
||||
callback: ProgressCallback = None,
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device via UDP command.
|
||||
|
||||
Sends reboot() to the Zynq's derived UDP port (last-octet - 3).
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
callback: Optional progress callback.
|
||||
|
||||
Returns:
|
||||
RebootResult with status.
|
||||
"""
|
||||
ip = config.zynq_ip
|
||||
port = _udp_port_from_ip(ip)
|
||||
|
||||
if callback:
|
||||
callback("start", f"UDP reboot → {ip}:{port}")
|
||||
|
||||
ok, resp = _send_udp_command(ip, b"reboot()")
|
||||
|
||||
if callback:
|
||||
if ok:
|
||||
callback("complete", f"UDP reboot OK: {resp}")
|
||||
else:
|
||||
callback("error", f"UDP reboot failed: {resp}")
|
||||
|
||||
return RebootResult(
|
||||
success=ok,
|
||||
message=f"reboot() via UDP: {resp}" if ok else resp,
|
||||
method="udp",
|
||||
output=resp,
|
||||
)
|
||||
|
||||
|
||||
# ── Version query ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def get_version_via_udp(
|
||||
config: Config,
|
||||
callback: ProgressCallback = None,
|
||||
) -> UdpVersionResult:
|
||||
"""Query Zynq firmware version via UDP.
|
||||
|
||||
Sends ver() to the Zynq's derived UDP port (last-octet - 3).
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
callback: Optional progress callback.
|
||||
|
||||
Returns:
|
||||
UdpVersionResult with version string.
|
||||
"""
|
||||
ip = config.zynq_ip
|
||||
port = _udp_port_from_ip(ip)
|
||||
|
||||
if callback:
|
||||
callback("start", f"UDP version query → {ip}:{port}")
|
||||
|
||||
ok, resp = _send_udp_command(ip, b"ver()")
|
||||
|
||||
if callback:
|
||||
if ok:
|
||||
callback("complete", f"Version: {resp}")
|
||||
else:
|
||||
callback("error", f"Version query failed: {resp}")
|
||||
|
||||
return UdpVersionResult(
|
||||
success=ok,
|
||||
message=f"Version: {resp}" if ok else resp,
|
||||
version=resp if ok else "",
|
||||
)
|
||||
|
||||
|
||||
# ── Top-level dispatcher ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def reboot_zynq(
|
||||
config: Config,
|
||||
method: str = "auto",
|
||||
@@ -184,12 +363,12 @@ def reboot_zynq(
|
||||
) -> RebootResult:
|
||||
"""Reboot the Zynq device using the best available method.
|
||||
|
||||
Tries SSH first, falls back to serial if SSH is unavailable.
|
||||
Tries JTAG first, falls back to UDP then serial.
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
method: Reboot method: 'auto', 'ssh', or 'serial'.
|
||||
timeout: Seconds to wait for reboot.
|
||||
method: Reboot method: 'auto', 'jtag', 'udp', or 'serial'.
|
||||
timeout: Seconds to wait for reboot (ssh-only, kept for compat).
|
||||
callback: Optional progress callback.
|
||||
|
||||
Returns:
|
||||
@@ -197,17 +376,25 @@ def reboot_zynq(
|
||||
"""
|
||||
timeout = timeout or config.reboot_timeout
|
||||
|
||||
if method == "ssh":
|
||||
return reboot_via_ssh(config, timeout, callback)
|
||||
if method == "jtag":
|
||||
return reboot_via_jtag(config, callback)
|
||||
elif method == "udp":
|
||||
return reboot_via_udp(config, callback)
|
||||
elif method == "serial":
|
||||
return reboot_via_serial(config, callback)
|
||||
|
||||
# Auto: try SSH first, then serial
|
||||
result = reboot_via_ssh(config, timeout, callback)
|
||||
# Auto: try JTAG first, then UDP, then serial
|
||||
result = reboot_via_jtag(config, callback)
|
||||
if result.success:
|
||||
return result
|
||||
|
||||
if callback:
|
||||
callback("progress", "SSH failed, trying serial...")
|
||||
callback("progress", "JTAG failed, trying UDP...")
|
||||
result = reboot_via_udp(config, callback)
|
||||
if result.success:
|
||||
return result
|
||||
|
||||
if callback:
|
||||
callback("progress", "UDP failed, trying serial...")
|
||||
|
||||
return reboot_via_serial(config, callback)
|
||||
|
||||
+23
-11
@@ -9,6 +9,7 @@ from __future__ import annotations
|
||||
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable
|
||||
|
||||
@@ -136,6 +137,7 @@ def check_uart_available(
|
||||
port: str,
|
||||
baudrate: int = 115200,
|
||||
timeout: float = 3.0,
|
||||
monitor: UartMonitor | None = None,
|
||||
) -> tuple[bool, str]:
|
||||
"""Check whether the UART serial port is available and readable.
|
||||
|
||||
@@ -144,19 +146,24 @@ def check_uart_available(
|
||||
considered available (the device may simply not be outputting yet).
|
||||
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:
|
||||
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
||||
baudrate: Baud rate for serial communication.
|
||||
timeout: Seconds to wait for data after opening.
|
||||
monitor: Optional running UartMonitor — if active, reuse it.
|
||||
|
||||
Returns:
|
||||
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
|
||||
ports = detect_serial_ports()
|
||||
if not ports:
|
||||
@@ -222,28 +229,33 @@ def test_serial_version(
|
||||
port: str,
|
||||
baudrate: int = 115200,
|
||||
timeout: float = 5.0,
|
||||
monitor: UartMonitor | None = None,
|
||||
) -> tuple[bool, str, str]:
|
||||
"""Test serial port by sending 'ver()' command and parsing response.
|
||||
|
||||
Sends 'ver\\r\\n' to the serial port and reads the response.
|
||||
Attempts to parse version string from the response.
|
||||
If *monitor* is running, skips opening a new connection and
|
||||
returns the latest parsed version from the monitor instead.
|
||||
|
||||
Args:
|
||||
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
||||
baudrate: Baud rate for serial communication.
|
||||
timeout: Read timeout in seconds.
|
||||
monitor: Optional running UartMonitor — if active, reuse it.
|
||||
|
||||
Returns:
|
||||
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
|
||||
|
||||
# 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:
|
||||
with serial.Serial(port, baudrate, timeout=timeout) as ser:
|
||||
# Clear buffers
|
||||
ser.reset_input_buffer()
|
||||
ser.reset_output_buffer()
|
||||
|
||||
|
||||
+163
-18
@@ -9,7 +9,6 @@ from __future__ import annotations
|
||||
import binascii
|
||||
import socket
|
||||
import struct
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
@@ -55,7 +54,8 @@ def _compute_crc32(file_path: str | Path) -> int:
|
||||
|
||||
|
||||
def _tftp_put(host: str, file_path: str, remote_name: str,
|
||||
timeout: float = 10.0) -> tuple[bool, str]:
|
||||
timeout: float = 10.0,
|
||||
callback: Callable[[str, str], None] | None = None) -> tuple[bool, str]:
|
||||
"""Upload a file to a TFTP server using pure Python.
|
||||
|
||||
Args:
|
||||
@@ -63,24 +63,28 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
|
||||
file_path: Local file path.
|
||||
remote_name: Remote filename.
|
||||
timeout: Socket timeout in seconds.
|
||||
callback: Optional progress callback (status, message).
|
||||
|
||||
Returns:
|
||||
(success, message).
|
||||
"""
|
||||
data = Path(file_path).read_bytes()
|
||||
total = len(data)
|
||||
block_size = _TFTP_BLOCK_SIZE
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.settimeout(timeout)
|
||||
# Allow reuse of local port (prevents "Address already in use")
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
|
||||
try:
|
||||
# ── Send WRQ ──
|
||||
# ── Send WRQ to port 69 ──
|
||||
wrq = struct.pack("!H", _TFTP_WRQ)
|
||||
wrq += remote_name.encode("ascii") + b"\x00"
|
||||
wrq += b"octet\x00"
|
||||
sock.sendto(wrq, (host, 69))
|
||||
|
||||
# ── Wait for ACK(0) to get server's data port ──
|
||||
# ── Wait for ACK(0) — discover server's data port ──
|
||||
ack_data, server_addr = sock.recvfrom(516)
|
||||
if len(ack_data) < 4:
|
||||
return False, "Invalid ACK from server"
|
||||
@@ -89,24 +93,36 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
|
||||
err_msg = ack_data[4:].decode("ascii", errors="replace").rstrip("\x00")
|
||||
return False, f"TFTP error: {err_msg}"
|
||||
if opcode != _TFTP_ACK:
|
||||
return False, f"Expected ACK, got opcode {opcode}"
|
||||
return False, f"Expected ACK(0), got opcode {opcode} from {server_addr}"
|
||||
|
||||
srv_ip, srv_port = server_addr
|
||||
import sys
|
||||
print(f"[TFTP] ACK(0) received from {srv_ip}:{srv_port} "
|
||||
f"(file={total}B, blocks={-(-total // block_size)})", file=sys.stderr)
|
||||
|
||||
# ── Send data blocks ──
|
||||
block = 1
|
||||
offset = 0
|
||||
while offset < total:
|
||||
chunk = data[offset:offset + _TFTP_BLOCK_SIZE]
|
||||
chunk = data[offset:offset + block_size]
|
||||
offset += len(chunk)
|
||||
data_pkt = struct.pack("!HH", _TFTP_DATA, block) + chunk
|
||||
sock.sendto(data_pkt, server_addr)
|
||||
|
||||
if block % 100 == 0 or block == -(-total // block_size):
|
||||
pct = offset * 100 // total
|
||||
print(f"[TFTP] Block {block}/{-(-total // block_size)} ({pct}%) sent to {srv_ip}:{srv_port}",
|
||||
file=sys.stderr)
|
||||
if callback:
|
||||
callback("progress", f"{pct}%")
|
||||
|
||||
# Wait for ACK
|
||||
try:
|
||||
ack_data, _ = sock.recvfrom(516)
|
||||
ack_op = struct.unpack("!H", ack_data[:2])[0]
|
||||
ack_block = struct.unpack("!H", ack_data[2:4])[0]
|
||||
if ack_op != _TFTP_ACK or ack_block != block:
|
||||
return False, f"Bad ACK: op={ack_op} block={ack_block}"
|
||||
return False, f"Bad ACK: op={ack_op} block={ack_block} (expected {block})"
|
||||
except socket.timeout:
|
||||
return False, f"Timeout waiting for ACK block {block}"
|
||||
block += 1
|
||||
@@ -116,6 +132,105 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
|
||||
sock.close()
|
||||
|
||||
|
||||
def _tftp_get(host: str, remote_name: str, local_dir: str | Path,
|
||||
timeout: float = 30.0,
|
||||
callback: Callable[[str, str], None] | None = None,
|
||||
expected_size: int = 0) -> tuple[bool, str, Path]:
|
||||
"""Download a file from a TFTP server using pure Python.
|
||||
|
||||
Args:
|
||||
host: TFTP server IP address.
|
||||
remote_name: Remote filename.
|
||||
local_dir: Local directory to save the downloaded file.
|
||||
timeout: Socket timeout in seconds.
|
||||
callback: Optional progress callback (status, message).
|
||||
expected_size: Expected file size in bytes (for progress percentage).
|
||||
|
||||
Returns:
|
||||
(success, message, local_path).
|
||||
"""
|
||||
local_dir = Path(local_dir)
|
||||
local_dir.mkdir(parents=True, exist_ok=True)
|
||||
local_path = local_dir / remote_name
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.settimeout(timeout)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
|
||||
try:
|
||||
# ── Send RRQ to port 69 ──
|
||||
rrq = struct.pack("!H", _TFTP_RRQ)
|
||||
rrq += remote_name.encode("ascii") + b"\x00"
|
||||
rrq += b"octet\x00"
|
||||
sock.sendto(rrq, (host, 69))
|
||||
|
||||
# ── Wait for first DATA block — discover server's data port ──
|
||||
data_pkt, server_addr = sock.recvfrom(516)
|
||||
if len(data_pkt) < 4:
|
||||
return False, "Invalid DATA from server", local_path
|
||||
opcode = struct.unpack("!H", data_pkt[:2])[0]
|
||||
if opcode == _TFTP_ERROR:
|
||||
err_msg = data_pkt[4:].decode("ascii", errors="replace").rstrip("\x00")
|
||||
return False, f"TFTP error: {err_msg}", local_path
|
||||
if opcode != _TFTP_DATA:
|
||||
return False, f"Expected DATA, got opcode {opcode} from {server_addr}", local_path
|
||||
|
||||
srv_ip, srv_port = server_addr
|
||||
block = struct.unpack("!H", data_pkt[2:4])[0]
|
||||
chunk = data_pkt[4:]
|
||||
import sys
|
||||
print(f"[TFTP] DATA block {block} received from {srv_ip}:{srv_port} "
|
||||
f"({len(chunk)} bytes)", file=sys.stderr)
|
||||
|
||||
# ── ACK this block with the actual block number ──
|
||||
file_data = bytearray(chunk)
|
||||
ack = struct.pack("!HH", _TFTP_ACK, block)
|
||||
sock.sendto(ack, server_addr)
|
||||
|
||||
if callback:
|
||||
pct = len(file_data) * 100 // expected_size if expected_size > 0 else -1
|
||||
callback("progress", f"{pct}%")
|
||||
|
||||
# ── Receive subsequent blocks while last chunk was full-size ──
|
||||
while len(chunk) == _TFTP_BLOCK_SIZE:
|
||||
try:
|
||||
data_pkt, _ = sock.recvfrom(516)
|
||||
if len(data_pkt) < 4:
|
||||
return False, "Incomplete DATA packet", local_path
|
||||
opcode = struct.unpack("!H", data_pkt[:2])[0]
|
||||
if opcode == _TFTP_ERROR:
|
||||
err_msg = data_pkt[4:].decode("ascii", errors="replace").rstrip("\x00")
|
||||
return False, f"TFTP error: {err_msg}", local_path
|
||||
if opcode != _TFTP_DATA:
|
||||
return False, f"Expected DATA, got opcode {opcode}", local_path
|
||||
|
||||
new_block = struct.unpack("!H", data_pkt[2:4])[0]
|
||||
chunk = data_pkt[4:]
|
||||
file_data += chunk
|
||||
block = new_block
|
||||
|
||||
if block % 100 == 0:
|
||||
print(f"[TFTP] Block {block} ({len(file_data)} bytes) "
|
||||
f"received from {srv_ip}:{srv_port}", file=sys.stderr)
|
||||
|
||||
# ACK this block
|
||||
ack = struct.pack("!HH", _TFTP_ACK, block)
|
||||
sock.sendto(ack, server_addr)
|
||||
|
||||
if callback:
|
||||
pct = len(file_data) * 100 // expected_size if expected_size > 0 else -1
|
||||
callback("progress", f"{pct}%")
|
||||
except socket.timeout:
|
||||
return False, f"Timeout waiting for DATA block {block + 1}", local_path
|
||||
|
||||
# ── Write file ──
|
||||
total = len(file_data)
|
||||
local_path.write_bytes(bytes(file_data))
|
||||
return True, f"Downloaded {total} bytes in {block} blocks to {local_path}", local_path
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
|
||||
def tftp_upload(
|
||||
config: Config,
|
||||
file_path: str | Path,
|
||||
@@ -156,7 +271,8 @@ def tftp_upload(
|
||||
)
|
||||
|
||||
try:
|
||||
ok, msg = _tftp_put(config.zynq_ip, str(file_path), remote_name)
|
||||
ok, msg = _tftp_put(config.zynq_ip, str(file_path), remote_name,
|
||||
callback=callback)
|
||||
except OSError as e:
|
||||
return TftpResult(
|
||||
step="upload", success=False,
|
||||
@@ -181,30 +297,59 @@ def tftp_download(
|
||||
remote_name: str,
|
||||
local_dir: str | Path | None = None,
|
||||
callback: ProgressCallback = None,
|
||||
expected_size: int = 0,
|
||||
) -> TftpResult:
|
||||
"""Download a file from the Zynq device via TFTP (pure Python).
|
||||
|
||||
Note: This requires the Zynq to act as a TFTP client (U-Boot tftpboot),
|
||||
which typically means we need a TFTP server on the PC. For simplicity,
|
||||
this operation is not natively supported — use ping-based verification
|
||||
or UART log analysis instead.
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
remote_name: Remote filename on the TFTP server.
|
||||
local_dir: Local directory to save the file.
|
||||
local_dir: Local directory to save the file (default: project .tmp/tftp/).
|
||||
callback: Optional progress callback.
|
||||
expected_size: Expected file size in bytes (for progress percentage).
|
||||
|
||||
Returns:
|
||||
TftpResult with status (always not-implemented).
|
||||
TftpResult with download status and CRC.
|
||||
"""
|
||||
if local_dir is None:
|
||||
# Use project-local temp dir for cross-platform compatibility
|
||||
local_dir = Path(__file__).resolve().parent.parent / ".tmp" / "tftp"
|
||||
local_dir = Path(local_dir)
|
||||
|
||||
if callback:
|
||||
callback("start", f"Downloading {remote_name} from {config.zynq_ip}")
|
||||
|
||||
# Verify device is reachable
|
||||
if not ping_ip(config.zynq_ip, config.ping_count, config.ping_timeout):
|
||||
return TftpResult(
|
||||
step="download",
|
||||
success=False,
|
||||
message="TFTP download requires PC-side server (not implemented)",
|
||||
step="download", success=False,
|
||||
message=f"Device {config.zynq_ip} not reachable",
|
||||
remote_path=remote_name,
|
||||
)
|
||||
|
||||
try:
|
||||
ok, msg, local_path = _tftp_get(config.zynq_ip, remote_name, local_dir,
|
||||
callback=callback, expected_size=expected_size)
|
||||
except OSError as e:
|
||||
return TftpResult(
|
||||
step="download", success=False,
|
||||
message=f"Socket error: {e}",
|
||||
remote_path=remote_name,
|
||||
)
|
||||
|
||||
crc = _compute_crc32(local_path) if ok else 0
|
||||
|
||||
if callback:
|
||||
callback("complete" if ok else "error",
|
||||
"Download " + ("succeeded" if ok else "failed"))
|
||||
|
||||
return TftpResult(
|
||||
step="download", success=ok,
|
||||
message=msg,
|
||||
local_path=str(local_path), remote_path=remote_name,
|
||||
crc_local=crc,
|
||||
)
|
||||
|
||||
|
||||
def tftp_upload_verify(
|
||||
config: Config,
|
||||
|
||||
+303
-62
@@ -5,6 +5,12 @@ Checks for required Xilinx tools:
|
||||
- program_flash — QSPI flash programming
|
||||
- bootgen — BOOT.BIN generation (image creation only, not hardware)
|
||||
|
||||
Version selection strategy:
|
||||
- Scans all Vitis/Vivado version directories under xilinx_root
|
||||
- Picks the NEWEST version (by semantic version comparison)
|
||||
- Logs all found versions so the user can verify the choice
|
||||
- If a specific version is needed, set xilinx_path directly
|
||||
|
||||
Provides shared helpers used by flash_programmer and bitstream_programmer.
|
||||
"""
|
||||
|
||||
@@ -12,6 +18,7 @@ from __future__ import annotations
|
||||
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
@@ -49,8 +56,104 @@ TOOL_ALIASES: dict[str, list[str]] = {
|
||||
"xsct": ["xsct"],
|
||||
"program_flash": ["program_flash"],
|
||||
"bootgen": ["bootgen"],
|
||||
"hw_server": ["hw_server"],
|
||||
}
|
||||
|
||||
# Product subdirectories under xilinx_root to scan (in preference order)
|
||||
# SDK covers pre-2019 installations where xsct lived in SDK/*/bin/
|
||||
_PRODUCT_DIRS = ("Vitis", "Vivado", "SDK")
|
||||
|
||||
|
||||
# ── Version helpers ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _parse_version_tuple(dirname: str) -> tuple[int, ...]:
|
||||
"""Parse a semantic-like version from a directory name.
|
||||
|
||||
E.g., '2023.2' → (2023, 2), '2018.3' → (2018, 3).
|
||||
Returns (0,) for unparseable names.
|
||||
|
||||
Args:
|
||||
dirname: Directory name (e.g., '2023.2', '2018.3.1').
|
||||
|
||||
Returns:
|
||||
Tuple of integers for version comparison.
|
||||
"""
|
||||
m = re.match(r"(\d{4})\.(\d+)(?:\.(\d+))?", dirname)
|
||||
if m:
|
||||
return tuple(int(g) for g in m.groups() if g is not None)
|
||||
return (0,)
|
||||
|
||||
|
||||
def _version_from_path(tool_path: str) -> str:
|
||||
"""Extract version string from a tool's install path.
|
||||
|
||||
E.g., '/data/xilinx/Vitis/2023.2/bin/xsct' → '2023.2'.
|
||||
|
||||
Args:
|
||||
tool_path: Absolute path to the tool executable.
|
||||
|
||||
Returns:
|
||||
Version string or empty string.
|
||||
"""
|
||||
parts = Path(tool_path).parts
|
||||
for i, part in enumerate(parts):
|
||||
if re.match(r"\d{4}\.\d+", part) and i > 0:
|
||||
return part
|
||||
return ""
|
||||
|
||||
|
||||
# ── Executable discovery ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def _scan_xilinx_root(
|
||||
xilinx_root: str, exec_name: str, exec_suffix: str,
|
||||
) -> list[tuple[tuple[int, ...], Path, str]]:
|
||||
"""Scan xilinx_root for all installed versions of a tool.
|
||||
|
||||
Searches under Vitis/*/bin/ and Vivado/*/bin/ for the executable.
|
||||
On Windows also checks .bat wrappers in addition to .exe.
|
||||
|
||||
Args:
|
||||
xilinx_root: Xilinx root directory (e.g., /data/xilinx, C:/Xilinx).
|
||||
exec_name: Executable name without extension.
|
||||
exec_suffix: Platform suffix ('' or '.exe').
|
||||
|
||||
Returns:
|
||||
List of (version_tuple, full_path, product_name) sorted newest first.
|
||||
"""
|
||||
results: list[tuple[tuple[int, ...], Path, str]] = []
|
||||
xr = Path(xilinx_root)
|
||||
if not xr.is_dir():
|
||||
return results
|
||||
|
||||
# On Windows, also try .bat extension (Xilinx tools use .bat wrappers)
|
||||
suffixes = [exec_suffix]
|
||||
if exec_suffix == ".exe":
|
||||
suffixes.append(".bat")
|
||||
|
||||
for product in _PRODUCT_DIRS:
|
||||
pdir = xr / product
|
||||
if not pdir.is_dir():
|
||||
continue
|
||||
try:
|
||||
entries = list(pdir.iterdir())
|
||||
except OSError:
|
||||
continue
|
||||
for ver_dir in entries:
|
||||
if not ver_dir.is_dir():
|
||||
continue
|
||||
for sfx in suffixes:
|
||||
candidate = ver_dir / "bin" / f"{exec_name}{sfx}"
|
||||
if candidate.is_file():
|
||||
v = _parse_version_tuple(ver_dir.name)
|
||||
if v != (0,):
|
||||
results.append((v, candidate, product))
|
||||
break # Found for this version dir
|
||||
# Sort newest first (descending version tuple)
|
||||
results.sort(key=lambda x: x[0], reverse=True)
|
||||
return results
|
||||
|
||||
|
||||
def _find_executable(
|
||||
exec_name: str,
|
||||
@@ -61,56 +164,55 @@ def _find_executable(
|
||||
"""Find an executable by name using PATH, Vitis, and Xilinx root.
|
||||
|
||||
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)
|
||||
3. xilinx_root/Vitis/*/bin and Vivado/*/bin (multi-version scan)
|
||||
4. Common install paths
|
||||
3. xilinx_root/Vitis/*/bin, Vivado/*/bin, SDK/*/bin — pick NEWEST version
|
||||
4. Common install paths (cross-platform)
|
||||
|
||||
Args:
|
||||
exec_name: Executable name (without extension).
|
||||
vitis_path: Legacy Vitis installation directory.
|
||||
system: Platform identifier (linux, windows, darwin).
|
||||
xilinx_root: Xilinx root directory (e.g., /data/xilinx or C:/Xilinx).
|
||||
xilinx_root: Xilinx root directory (e.g., /data/xilinx, C:/Xilinx).
|
||||
|
||||
Returns:
|
||||
Absolute path to the executable, or None.
|
||||
Absolute path to the best executable, or None.
|
||||
"""
|
||||
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}")
|
||||
if not found:
|
||||
found = shutil.which(f"{exec_name}.bat")
|
||||
if found:
|
||||
return found
|
||||
|
||||
# 2. Legacy vitis_path/bin
|
||||
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():
|
||||
return str(candidate)
|
||||
|
||||
# 3. Scan xilinx_root for all versions (Vitis + Vivado)
|
||||
# 3. Scan xilinx_root — pick newest version across Vitis + Vivado
|
||||
if xilinx_root:
|
||||
xr = Path(xilinx_root)
|
||||
for product in ("Vitis", "Vivado"):
|
||||
pdir = xr / product
|
||||
if not pdir.is_dir():
|
||||
continue
|
||||
# Scan version directories, newest first
|
||||
versions = sorted(
|
||||
[d for d in pdir.iterdir() if d.is_dir()],
|
||||
reverse=True,
|
||||
)
|
||||
for ver_dir in versions:
|
||||
candidate = ver_dir / "bin" / f"{exec_name}{exec_suffix}"
|
||||
if candidate.is_file():
|
||||
return str(candidate)
|
||||
candidates = _scan_xilinx_root(xilinx_root, exec_name, exec_suffix)
|
||||
if candidates:
|
||||
# Preference: Vitis over Vivado if same version
|
||||
# candidates are already sorted by version desc;
|
||||
# within same version, Vitis appears first (scanned first)
|
||||
return str(candidates[0][1])
|
||||
|
||||
# 4. Common install paths (cross-platform)
|
||||
# 4. Common install paths (cross-platform fallback)
|
||||
if system == "windows":
|
||||
common = [
|
||||
f"C:/Xilinx/Vitis/2023.2/bin/{exec_name}.exe",
|
||||
f"C:/Xilinx/Vivado/2023.2/bin/{exec_name}.exe",
|
||||
]
|
||||
common: list[str] = []
|
||||
for sfx in (".bat", ".exe"):
|
||||
for ver in ("2023.2", "2022.2", "2018.3"):
|
||||
for prod in ("Vitis", "Vivado", "SDK"):
|
||||
common.append(f"C:/Xilinx/{prod}/{ver}/bin/{exec_name}{sfx}")
|
||||
else:
|
||||
common = [
|
||||
f"/data/xilinx/Vitis/2023.2/bin/{exec_name}",
|
||||
@@ -126,78 +228,206 @@ def _find_executable(
|
||||
return None
|
||||
|
||||
|
||||
def get_xsct_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
|
||||
"""Find the xsct (Xilinx Software Command Line Tool) executable.
|
||||
def get_all_found_versions(
|
||||
xilinx_root: str,
|
||||
exec_name: str,
|
||||
) -> list[tuple[str, str]]:
|
||||
"""Return all found versions of a tool under xilinx_root.
|
||||
|
||||
Useful for logging which versions are available.
|
||||
|
||||
Args:
|
||||
vitis_path: Legacy Vitis install path.
|
||||
xilinx_root: Xilinx root directory.
|
||||
exec_name: Executable name without extension.
|
||||
|
||||
Returns:
|
||||
Absolute path to xsct, or None.
|
||||
List of (version_string, product_name) sorted newest first.
|
||||
"""
|
||||
system = platform.system().lower()
|
||||
suffix = ".exe" if system == "windows" else ""
|
||||
candidates = _scan_xilinx_root(xilinx_root, exec_name, suffix)
|
||||
return [
|
||||
(Path(str(p)).parent.parent.name, product)
|
||||
for _, p, product in candidates
|
||||
]
|
||||
|
||||
|
||||
# ── Environment setup ────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _find_settings_script(tool_path: str) -> str | None:
|
||||
"""Find the settings64 script for a Xilinx tool's version directory.
|
||||
|
||||
Walks up from the tool's bin/ directory looking for
|
||||
settings64.sh (Linux) or settings64.bat (Windows).
|
||||
|
||||
Args:
|
||||
tool_path: Absolute path to a Xilinx tool executable.
|
||||
|
||||
Returns:
|
||||
Path to settings64 script, or None if not found.
|
||||
"""
|
||||
p = Path(tool_path).resolve()
|
||||
for parent in p.parents:
|
||||
for name in ("settings64.sh", "settings64.bat"):
|
||||
candidate = parent / name
|
||||
if candidate.is_file():
|
||||
return str(candidate)
|
||||
return None
|
||||
|
||||
|
||||
def build_tool_command(tool_path: str, *args: str) -> list[str]:
|
||||
"""Build a subprocess-safe command with Xilinx environment setup.
|
||||
|
||||
On Linux: sources settings64.sh before running the tool.
|
||||
On Windows: .bat wrappers handle environment internally.
|
||||
|
||||
Args:
|
||||
tool_path: Path to the Xilinx tool executable.
|
||||
*args: Arguments to pass to the tool.
|
||||
|
||||
Returns:
|
||||
Command list suitable for subprocess.run().
|
||||
"""
|
||||
import shlex
|
||||
|
||||
system = platform.system().lower()
|
||||
|
||||
# Windows: .bat wrappers set up their own environment
|
||||
if system == "windows":
|
||||
return [tool_path, *args]
|
||||
|
||||
# Linux: find settings64.sh and source it before running
|
||||
settings = _find_settings_script(tool_path)
|
||||
if not settings:
|
||||
return [tool_path, *args]
|
||||
|
||||
quoted = " ".join(shlex.quote(a) for a in [tool_path, *args])
|
||||
return ["bash", "-c", f"source {shlex.quote(settings)} && {quoted}"]
|
||||
|
||||
|
||||
# ── Public path getters ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def get_xsct_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
|
||||
"""Find the xsct (Xilinx Software Command Line Tool) executable."""
|
||||
return _find_executable("xsct", vitis_path, platform.system().lower(), xilinx_root)
|
||||
|
||||
|
||||
def get_program_flash_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
|
||||
"""Find the program_flash executable.
|
||||
|
||||
Args:
|
||||
vitis_path: Legacy Vitis install path.
|
||||
xilinx_root: Xilinx root directory.
|
||||
|
||||
Returns:
|
||||
Absolute path to program_flash, or None.
|
||||
"""
|
||||
"""Find the program_flash executable."""
|
||||
return _find_executable("program_flash", vitis_path, platform.system().lower(), xilinx_root)
|
||||
|
||||
|
||||
def get_bootgen_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
|
||||
"""Find the bootgen executable.
|
||||
|
||||
Args:
|
||||
vitis_path: Legacy Vitis install path.
|
||||
xilinx_root: Xilinx root directory.
|
||||
|
||||
Returns:
|
||||
Absolute path to bootgen, or None.
|
||||
"""
|
||||
"""Find the bootgen executable."""
|
||||
return _find_executable("bootgen", vitis_path, platform.system().lower(), xilinx_root)
|
||||
|
||||
|
||||
def _get_tool_version(tool_path: str, tool_name: str) -> str:
|
||||
"""Get the version/h banner from a tool executable.
|
||||
def get_vivado_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
|
||||
"""Find the Vivado executable (JTAG operations via hw_server).
|
||||
|
||||
Args:
|
||||
tool_path: Absolute path to the tool.
|
||||
tool_name: Canonical tool name.
|
||||
vitis_path: Legacy Vitis installation directory.
|
||||
xilinx_root: Xilinx root directory.
|
||||
|
||||
Returns:
|
||||
First line of help output (truncated to 120 chars), or empty string.
|
||||
Path to vivado executable, or None.
|
||||
"""
|
||||
return _find_executable("vivado", vitis_path, platform.system().lower(), xilinx_root)
|
||||
|
||||
|
||||
def get_hw_server_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
|
||||
"""Find the hw_server executable (Xilinx hardware server).
|
||||
|
||||
hw_server must be running before Vivado/xsct can connect to the
|
||||
JTAG chain. This finder searches the same Vitis/Vivado bin dirs.
|
||||
|
||||
Args:
|
||||
vitis_path: Legacy Vitis installation directory.
|
||||
xilinx_root: Xilinx root directory.
|
||||
|
||||
Returns:
|
||||
Path to hw_server executable, or None.
|
||||
"""
|
||||
return _find_executable(
|
||||
"hw_server", vitis_path, platform.system().lower(), xilinx_root
|
||||
)
|
||||
|
||||
|
||||
# ── Tool version detection ───────────────────────────────────────────
|
||||
|
||||
|
||||
def _get_tool_version(tool_path: str) -> str:
|
||||
"""Get the version string from a tool executable.
|
||||
|
||||
Tries (in order):
|
||||
1. tool -version → parse version from output
|
||||
2. Extract from path (e.g., .../Vitis/2023.2/bin/xsct → 2023.2)
|
||||
|
||||
Args:
|
||||
tool_path: Absolute path to the tool executable.
|
||||
|
||||
Returns:
|
||||
Version string, or empty string if undetectable.
|
||||
"""
|
||||
# Try -version flag
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[tool_path, "-help"],
|
||||
[tool_path, "-version"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
)
|
||||
output = (result.stdout + result.stderr).strip()
|
||||
if output:
|
||||
return output.split("\n")[0][:120]
|
||||
# Scan all lines for a version pattern (skip batch echo noise)
|
||||
for raw_line in output.splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line:
|
||||
continue
|
||||
# Skip Windows batch echo noise
|
||||
if _is_batch_noise(line):
|
||||
continue
|
||||
m = re.search(r"(\d{4}\.\d+(?:\.\d+)?)", line)
|
||||
if m:
|
||||
return m.group(1)
|
||||
except (subprocess.TimeoutExpired, OSError):
|
||||
pass
|
||||
return ""
|
||||
|
||||
# Fallback: extract from path
|
||||
return _version_from_path(tool_path)
|
||||
|
||||
|
||||
def _is_batch_noise(line: str) -> bool:
|
||||
"""Check if a line is Windows batch script noise.
|
||||
|
||||
Args:
|
||||
line: A line of output text.
|
||||
|
||||
Returns:
|
||||
True if the line looks like batch echo/status noise.
|
||||
"""
|
||||
noise_patterns = [
|
||||
r"^ECHO\s+(is\s+(on|off)|处于)", # ECHO is on/off (en+zh)
|
||||
r"^[A-Z]:\\[^>]*>", # Prompt line like C:\path>
|
||||
]
|
||||
return any(re.search(p, line, re.IGNORECASE) for p in noise_patterns)
|
||||
|
||||
|
||||
# ── Main check function ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def check_vitis(config: Config) -> VitisCheckResult:
|
||||
"""Check availability of all required Xilinx tools.
|
||||
|
||||
Scans all versions under xilinx_root and picks the newest.
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
|
||||
Returns:
|
||||
VitisCheckResult with per-tool status.
|
||||
VitisCheckResult with per-tool status and version info.
|
||||
"""
|
||||
system = platform.system().lower()
|
||||
vitis_path = config.vitis_path if hasattr(config, 'vitis_path') else ""
|
||||
@@ -227,17 +457,25 @@ def _check_tool(
|
||||
system: str,
|
||||
xilinx_root: str = "",
|
||||
) -> ToolStatus:
|
||||
"""Check if a single tool is available."""
|
||||
"""Check if a single tool is available, picking the newest version."""
|
||||
best_path: str | None = None
|
||||
best_alias: str = ""
|
||||
|
||||
for alias in aliases:
|
||||
found = _find_executable(alias, vitis_path, system, xilinx_root)
|
||||
if found:
|
||||
version = _get_tool_version(found, tool_name)
|
||||
best_path = found
|
||||
best_alias = alias if alias != tool_name else ""
|
||||
break
|
||||
|
||||
if best_path:
|
||||
version = _get_tool_version(best_path)
|
||||
return ToolStatus(
|
||||
name=tool_name,
|
||||
found=True,
|
||||
path=found,
|
||||
path=best_path,
|
||||
version=version,
|
||||
alias=alias,
|
||||
alias=best_alias,
|
||||
)
|
||||
|
||||
return ToolStatus(
|
||||
@@ -251,6 +489,9 @@ def _check_tool(
|
||||
)
|
||||
|
||||
|
||||
# ── Status serialization ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def get_tool_status_dict(result: VitisCheckResult) -> dict[str, Any]:
|
||||
"""Convert a VitisCheckResult to a serializable dictionary.
|
||||
|
||||
|
||||
+217
-244
@@ -1,78 +1,84 @@
|
||||
"""Zynq JTAG presence checker for Zynq Flasher GUI.
|
||||
|
||||
Connects to hw_server via Vivado and scans the JTAG chain to detect
|
||||
Zynq devices, returning chip model and presence information.
|
||||
Checks for both PS (Processing System) and PL (Programmable Logic)
|
||||
recognition on the JTAG chain.
|
||||
Starts hw_server explicitly, then uses Vivado batch-mode TCL to scan
|
||||
the JTAG chain. On failure, Vivado's stdout/stderr is returned in the
|
||||
error message for debugging.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
from config_manager import Config
|
||||
from vitis_checker import get_vivado_path, get_hw_server_path, build_tool_command
|
||||
|
||||
|
||||
@dataclass
|
||||
class JtagDevice:
|
||||
"""Information about a device on the JTAG chain."""
|
||||
|
||||
name: str # Device name from JTAG chain (e.g. "xc7z100_1")
|
||||
is_zynq: bool # Whether this is a Zynq device
|
||||
is_arm_dap: bool # Whether this is an ARM DAP (PS side)
|
||||
is_pl_device: bool # Whether this is a PL device
|
||||
name: str
|
||||
idcode: str = ""
|
||||
is_zynq: bool = False
|
||||
is_arm_dap: bool = False
|
||||
is_pl_device: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class ZynqDevice:
|
||||
"""Information about a detected Zynq device on the JTAG chain."""
|
||||
|
||||
name: str # Device name from JTAG chain (e.g. "xc7z100_1")
|
||||
part_number: str # Extracted part number (e.g. "XC7Z100")
|
||||
family: str # Family (e.g. "zynq7")
|
||||
speed: str # Speed grade (e.g. "1", "2", "-1", "-2")
|
||||
idcode: str = "" # JTAG IDCODE if available
|
||||
is_programmable: bool = False # Whether the device is programmable
|
||||
name: str
|
||||
part_number: str
|
||||
family: str
|
||||
speed: str
|
||||
idcode: str = ""
|
||||
is_programmable: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class ZynqCheckResult:
|
||||
"""Result of Zynq device detection on the JTAG chain."""
|
||||
|
||||
devices: list[ZynqDevice] # All detected Zynq devices
|
||||
jtag_chain: list[JtagDevice] # All devices on JTAG chain
|
||||
is_present: bool # Whether any Zynq device was found
|
||||
ps_detected: bool # Whether PS (ARM DAP) is detected
|
||||
pl_detected: bool # Whether PL is detected
|
||||
hw_server_url: str # hw_server connection URL used
|
||||
error: str = "" # Error message if detection failed
|
||||
devices: list[ZynqDevice]
|
||||
jtag_chain: list[JtagDevice]
|
||||
is_present: bool
|
||||
ps_detected: bool
|
||||
pl_detected: bool
|
||||
hw_server_url: str
|
||||
error: str = ""
|
||||
|
||||
|
||||
# Zynq part number patterns from JTAG device names
|
||||
# Examples: xc7z010_1, xc7z020_1, xc7z100_1, xc7z030_1
|
||||
ZYNQ_PART_PATTERN = re.compile(
|
||||
r"^(xc\d+z\d+[a-z]?\d+)_\d+$", re.IGNORECASE
|
||||
)
|
||||
|
||||
# ARM DAP pattern (PS side)
|
||||
ARM_DAP_PATTERN = re.compile(r"^arm_dap_\d+$", re.IGNORECASE)
|
||||
|
||||
# hw_server default port
|
||||
# ── Patterns ──────────────────────────────────────────────────────────
|
||||
_DEVICE_LINE_RE = re.compile(r"^DEVICE:(\S+)\s+IDCODE:([01]+)")
|
||||
_ZYNQ_RE = re.compile(r"^xc\d+z\d+[a-z]?\d+", re.IGNORECASE)
|
||||
_ARM_DAP_RE = re.compile(r"^arm_dap", re.IGNORECASE)
|
||||
HW_SERVER_PORT = 3121
|
||||
|
||||
|
||||
def check_zynq_jtag(config: Config, callback: Callable | None = None) -> ZynqCheckResult:
|
||||
"""Check for Zynq devices on the JTAG chain via hw_server.
|
||||
def _idcode_to_hex(binary_str: str) -> str:
|
||||
try:
|
||||
return f"{int(binary_str, 2):08X}"
|
||||
except ValueError:
|
||||
return binary_str
|
||||
|
||||
Connects to hw_server using Vivado batch mode, scans the JTAG chain,
|
||||
and identifies Zynq devices by their JTAG device names. Also checks
|
||||
for PS (ARM DAP) and PL (Programmable Logic) recognition.
|
||||
|
||||
# ── Main entry ────────────────────────────────────────────────────────
|
||||
|
||||
def check_zynq_jtag(
|
||||
config: Config, callback: Callable | None = None
|
||||
) -> ZynqCheckResult:
|
||||
"""Check for Zynq devices on the JTAG chain.
|
||||
|
||||
1. Start hw_server (Vivado may not auto-start on Windows)
|
||||
2. Run Vivado batch TCL to enumerate JTAG devices
|
||||
3. On failure, include Vivado output in error for debugging
|
||||
|
||||
Args:
|
||||
config: Application configuration.
|
||||
@@ -82,178 +88,221 @@ def check_zynq_jtag(config: Config, callback: Callable | None = None) -> ZynqChe
|
||||
ZynqCheckResult with detected devices and status.
|
||||
"""
|
||||
hw_server_url = f"localhost:{HW_SERVER_PORT}"
|
||||
xilinx_root = getattr(config, "xilinx_path", "")
|
||||
|
||||
if callback:
|
||||
callback("start", "Connecting to hw_server...")
|
||||
|
||||
# Find vivado executable
|
||||
vivado_path = _get_vivado_path(
|
||||
config.vitis_path if hasattr(config, 'vitis_path') else ""
|
||||
)
|
||||
# ── 1. Find tools ──────────────────────────────────────────
|
||||
vivado_path = get_vivado_path(xilinx_root=xilinx_root)
|
||||
if not vivado_path:
|
||||
return ZynqCheckResult(
|
||||
devices=[],
|
||||
jtag_chain=[],
|
||||
is_present=False,
|
||||
ps_detected=False,
|
||||
pl_detected=False,
|
||||
hw_server_url=hw_server_url,
|
||||
error="Vivado not found",
|
||||
)
|
||||
return _fail("Vivado not found", hw_server_url)
|
||||
|
||||
# Create temporary TCL script
|
||||
tcl_content = _build_jtag_query_tcl()
|
||||
tcl_path = Path("/tmp/zynq_jtag_check_{}.tcl".format(os.getpid()))
|
||||
hw_server_path = get_hw_server_path(xilinx_root=xilinx_root)
|
||||
if not hw_server_path:
|
||||
return _fail("hw_server not found", hw_server_url)
|
||||
|
||||
# ── 2. Start hw_server ─────────────────────────────────────
|
||||
hw_proc: subprocess.Popen | None = None
|
||||
try:
|
||||
tcl_path.write_text(tcl_content)
|
||||
hw_proc = subprocess.Popen(
|
||||
build_tool_command(hw_server_path),
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
# Wait up to 10s for hw_server to bind
|
||||
started = False
|
||||
for _ in range(100):
|
||||
time.sleep(0.1)
|
||||
if _port_listening():
|
||||
started = True
|
||||
break
|
||||
if not started:
|
||||
return _fail("hw_server did not bind port 3121 within 10s", hw_server_url)
|
||||
except OSError as e:
|
||||
return _fail(f"Failed to start hw_server: {e}", hw_server_url)
|
||||
|
||||
if callback:
|
||||
callback("progress", "Scanning JTAG chain...")
|
||||
|
||||
# Run Vivado in batch mode
|
||||
cmd = [vivado_path, "-mode", "batch", "-source", str(tcl_path)]
|
||||
# ── 3. Run Vivado TCL ──────────────────────────────────────
|
||||
tcl = _build_jtag_query_tcl()
|
||||
python_timeout = config.step_timeouts.get("check_env", 120)
|
||||
|
||||
# Redirect Vivado journal/log to temp dir so we don't litter the cwd
|
||||
tmp = tempfile.gettempdir()
|
||||
vivado_args = [
|
||||
"-mode", "batch",
|
||||
"-tempDir", tmp,
|
||||
"-journal", str(Path(tmp) / "vivado_jtag.jou"),
|
||||
"-log", str(Path(tmp) / "vivado_jtag.log"),
|
||||
"-source", tcl,
|
||||
]
|
||||
|
||||
output_lines: list[str] = []
|
||||
try:
|
||||
cmd = build_tool_command(vivado_path, *vivado_args)
|
||||
# Use run() with communicate() under the hood — reliably
|
||||
# drains pipes on Windows even when .bat wrappers are involved
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
timeout=python_timeout,
|
||||
)
|
||||
|
||||
output = result.stdout + result.stderr
|
||||
for line in output.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped:
|
||||
output_lines.append(stripped)
|
||||
if callback:
|
||||
callback("progress", stripped)
|
||||
except subprocess.TimeoutExpired:
|
||||
return _fail(
|
||||
f"JTAG scan timed out after {python_timeout}s",
|
||||
hw_server_url,
|
||||
)
|
||||
except OSError as e:
|
||||
return _fail(f"Failed to run Vivado: {e}", hw_server_url)
|
||||
finally:
|
||||
try:
|
||||
Path(tcl).unlink(missing_ok=True)
|
||||
except Exception:
|
||||
pass
|
||||
for f in ("vivado_jtag.jou", "vivado_jtag.log"):
|
||||
try:
|
||||
(Path(tmp) / f).unlink(missing_ok=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Parse JTAG device information from output
|
||||
output = "\n".join(output_lines)
|
||||
|
||||
# Parse results
|
||||
jtag_chain = _parse_jtag_chain(output)
|
||||
devices = _filter_zynq_devices(jtag_chain)
|
||||
|
||||
if callback:
|
||||
if devices:
|
||||
if callback:
|
||||
ps_note = "PS" if any(d.is_arm_dap for d in jtag_chain) else ""
|
||||
pl_note = "PL" if any(d.is_pl_device for d in jtag_chain) else ""
|
||||
notes = ", ".join(filter(None, [ps_note, pl_note]))
|
||||
callback("complete", f"Found {len(devices)} Zynq device(s) [{notes}]")
|
||||
else:
|
||||
callback("error", "No Zynq device found on JTAG chain")
|
||||
# Include Vivado output so user can debug
|
||||
err_detail = _extract_relevant(output)
|
||||
msg = "No Zynq device found on JTAG chain"
|
||||
if err_detail:
|
||||
msg += f" | Vivado: {err_detail[:300]}"
|
||||
if callback:
|
||||
callback("error", msg)
|
||||
return ZynqCheckResult(
|
||||
devices=[], jtag_chain=jtag_chain,
|
||||
is_present=False,
|
||||
ps_detected=False, pl_detected=False,
|
||||
hw_server_url=hw_server_url,
|
||||
error=msg,
|
||||
)
|
||||
|
||||
# Check PS and PL detection
|
||||
ps_detected = any(d.is_arm_dap for d in jtag_chain)
|
||||
pl_detected = any(d.is_pl_device for d in jtag_chain)
|
||||
|
||||
return ZynqCheckResult(
|
||||
devices=devices,
|
||||
jtag_chain=jtag_chain,
|
||||
is_present=len(devices) > 0,
|
||||
ps_detected=ps_detected,
|
||||
pl_detected=pl_detected,
|
||||
devices=devices, jtag_chain=jtag_chain,
|
||||
is_present=True,
|
||||
ps_detected=ps_detected, pl_detected=pl_detected,
|
||||
hw_server_url=hw_server_url,
|
||||
error="" if devices else "No Zynq device found on JTAG chain",
|
||||
error="",
|
||||
)
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return ZynqCheckResult(
|
||||
devices=[],
|
||||
jtag_chain=[],
|
||||
is_present=False,
|
||||
ps_detected=False,
|
||||
pl_detected=False,
|
||||
hw_server_url=hw_server_url,
|
||||
error="JTAG scan timed out after 30s",
|
||||
)
|
||||
except OSError as e:
|
||||
return ZynqCheckResult(
|
||||
devices=[],
|
||||
jtag_chain=[],
|
||||
is_present=False,
|
||||
ps_detected=False,
|
||||
pl_detected=False,
|
||||
hw_server_url=hw_server_url,
|
||||
error=f"OS error: {e}",
|
||||
)
|
||||
finally:
|
||||
tcl_path.unlink(missing_ok=True)
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────
|
||||
|
||||
def _fail(error: str, url: str) -> ZynqCheckResult:
|
||||
return ZynqCheckResult(
|
||||
devices=[], jtag_chain=[], is_present=False,
|
||||
ps_detected=False, pl_detected=False,
|
||||
hw_server_url=url, error=error,
|
||||
)
|
||||
|
||||
|
||||
def _port_listening(port: int = HW_SERVER_PORT) -> bool:
|
||||
"""Check if hw_server is listening on its TCP port."""
|
||||
import socket
|
||||
try:
|
||||
with socket.create_connection(("localhost", port), timeout=1):
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _extract_relevant(output: str) -> str:
|
||||
"""Extract error/warning lines from Vivado output for user display."""
|
||||
lines = []
|
||||
for line in output.splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
if any(kw in line for kw in ("ERROR", "WARNING", "CRITICAL", "FAIL", "cannot", "No")):
|
||||
lines.append(line)
|
||||
return " | ".join(lines[-5:]) # Last 5 relevant lines
|
||||
|
||||
|
||||
# ── TCL builder ────────────────────────────────────────────────────────
|
||||
|
||||
def _build_jtag_query_tcl() -> str:
|
||||
"""Build TCL script for JTAG device query.
|
||||
|
||||
Returns:
|
||||
TCL script content as string.
|
||||
"""
|
||||
return """
|
||||
open_hw_manager
|
||||
"""Build Vivado TCL script file for JTAG device enumeration."""
|
||||
content = """\
|
||||
open_hw
|
||||
connect_hw_server
|
||||
open_hw_target
|
||||
puts "===JTAG_START==="
|
||||
foreach dev [get_hw_devices] {
|
||||
set name [get_property NAME $dev]
|
||||
puts "DEVICE:$name"
|
||||
set idcode [get_property IDCODE $dev]
|
||||
puts "DEVICE:$name IDCODE:$idcode"
|
||||
}
|
||||
puts "===JTAG_END==="
|
||||
close_hw
|
||||
disconnect_hw_server
|
||||
quit
|
||||
"""
|
||||
import os
|
||||
path = Path(tempfile.gettempdir()) / f"zynq_jtag_{os.getpid()}.tcl"
|
||||
path.write_text(content)
|
||||
return str(path)
|
||||
|
||||
|
||||
# ── Parsers ────────────────────────────────────────────────────────────
|
||||
|
||||
def _parse_jtag_chain(output: str) -> list[JtagDevice]:
|
||||
"""Parse JTAG chain information from Vivado output.
|
||||
|
||||
Args:
|
||||
output: Combined stdout/stderr from Vivado batch execution.
|
||||
|
||||
Returns:
|
||||
List of JtagDevice objects for all devices on the JTAG chain.
|
||||
"""
|
||||
"""Parse Vivado TCL output into JtagDevice list."""
|
||||
devices: list[JtagDevice] = []
|
||||
|
||||
# Extract device list from output
|
||||
devices_section = _extract_section(output, "===JTAG_START===", "===JTAG_END===")
|
||||
if not devices_section:
|
||||
section = _extract_section(output, "===JTAG_START===", "===JTAG_END===")
|
||||
if not section:
|
||||
return devices
|
||||
|
||||
for line in devices_section.splitlines():
|
||||
for line in section.splitlines():
|
||||
line = line.strip()
|
||||
if line.startswith("DEVICE:"):
|
||||
device_name = line[len("DEVICE:"):]
|
||||
device = _classify_jtag_device(device_name)
|
||||
devices.append(device)
|
||||
|
||||
if not line:
|
||||
continue
|
||||
match = _DEVICE_LINE_RE.match(line)
|
||||
if match:
|
||||
name = match.group(1)
|
||||
idcode_hex = _idcode_to_hex(match.group(2))
|
||||
devices.append(_classify_jtag_device(name, idcode_hex))
|
||||
return devices
|
||||
|
||||
|
||||
def _filter_zynq_devices(jtag_chain: list[JtagDevice]) -> list[ZynqDevice]:
|
||||
"""Filter Zynq devices from JTAG chain.
|
||||
|
||||
Args:
|
||||
jtag_chain: List of all JTAG devices.
|
||||
|
||||
Returns:
|
||||
List of ZynqDevice objects.
|
||||
"""
|
||||
zynq_devices: list[ZynqDevice] = []
|
||||
for jtag_dev in jtag_chain:
|
||||
if jtag_dev.is_zynq:
|
||||
zynq_dev = _parse_zynq_device(jtag_dev.name)
|
||||
zynq_dev = _parse_zynq_device(jtag_dev.name, jtag_dev.idcode)
|
||||
if zynq_dev:
|
||||
zynq_devices.append(zynq_dev)
|
||||
return zynq_devices
|
||||
|
||||
|
||||
def _extract_section(
|
||||
text: str, start_marker: str, end_marker: str
|
||||
) -> str | None:
|
||||
"""Extract text between two markers, handling duplicate markers.
|
||||
|
||||
Finds the last occurrence of each marker to avoid matching
|
||||
markers inside TCL script source code.
|
||||
|
||||
Args:
|
||||
text: Full output text.
|
||||
start_marker: Starting marker string.
|
||||
end_marker: Ending marker string.
|
||||
|
||||
Returns:
|
||||
Text between markers, or None if not found.
|
||||
"""
|
||||
# Find last occurrence of each marker to avoid matching
|
||||
# markers inside TCL script source code
|
||||
def _extract_section(text: str, start_marker: str, end_marker: str) -> str | None:
|
||||
start_idx = text.rfind(start_marker)
|
||||
end_idx = text.rfind(end_marker)
|
||||
if start_idx == -1 or end_idx == -1 or start_idx >= end_idx:
|
||||
@@ -261,105 +310,37 @@ def _extract_section(
|
||||
return text[start_idx + len(start_marker):end_idx].strip()
|
||||
|
||||
|
||||
def _classify_jtag_device(device_name: str) -> JtagDevice:
|
||||
"""Classify a JTAG device by its name.
|
||||
|
||||
Args:
|
||||
device_name: Device name from JTAG chain.
|
||||
|
||||
Returns:
|
||||
JtagDevice with classification information.
|
||||
"""
|
||||
is_zynq = bool(ZYNQ_PART_PATTERN.match(device_name))
|
||||
is_arm_dap = bool(ARM_DAP_PATTERN.match(device_name))
|
||||
is_pl_device = is_zynq and not is_arm_dap
|
||||
|
||||
def _classify_jtag_device(device_name: str, idcode: str = "") -> JtagDevice:
|
||||
is_zynq = bool(_ZYNQ_RE.match(device_name))
|
||||
is_arm_dap = bool(_ARM_DAP_RE.match(device_name))
|
||||
return JtagDevice(
|
||||
name=device_name,
|
||||
idcode=idcode,
|
||||
is_zynq=is_zynq,
|
||||
is_arm_dap=is_arm_dap,
|
||||
is_pl_device=is_pl_device,
|
||||
is_pl_device=is_zynq and not is_arm_dap,
|
||||
)
|
||||
|
||||
|
||||
def _parse_zynq_device(device_name: str) -> ZynqDevice | None:
|
||||
"""Parse a Zynq device from its JTAG chain name.
|
||||
|
||||
Args:
|
||||
device_name: Device name from JTAG chain (e.g. "xc7z100_1").
|
||||
|
||||
Returns:
|
||||
ZynqDevice if it's a Zynq chip, None otherwise.
|
||||
"""
|
||||
match = ZYNQ_PART_PATTERN.match(device_name)
|
||||
def _parse_zynq_device(device_name: str, idcode: str = "") -> ZynqDevice | None:
|
||||
match = _ZYNQ_RE.match(device_name)
|
||||
if not match:
|
||||
return None
|
||||
|
||||
part_base = match.group(1)
|
||||
# Convert to uppercase for part number
|
||||
part_number = part_base.upper()
|
||||
|
||||
# Extract family and speed from part number
|
||||
# Examples: XC7Z010 -> family=zynq7, speed=1; XC7Z100 -> family=zynq7, speed=1
|
||||
base = device_name.rsplit("_", 1)[0] if "_" in device_name else device_name
|
||||
part_number = base.upper()
|
||||
family = "zynq7"
|
||||
speed = "1" # Default speed grade
|
||||
|
||||
# Parse speed grade from the numeric suffix
|
||||
speed_match = re.search(r"(\d+)$", part_base)
|
||||
if speed_match:
|
||||
speed_num = int(speed_match.group(1))
|
||||
# Map numeric speed to speed grade string
|
||||
if speed_num >= 2:
|
||||
speed = "-2"
|
||||
elif speed_num >= 1:
|
||||
speed = "-1"
|
||||
|
||||
speed_match = re.search(r"(\d+)$", part_number)
|
||||
if speed_match:
|
||||
speed = "-2" if int(speed_match.group(1)) >= 2 else "-1"
|
||||
return ZynqDevice(
|
||||
name=device_name,
|
||||
part_number=part_number,
|
||||
family=family,
|
||||
speed=speed,
|
||||
name=device_name, part_number=part_number,
|
||||
family=family, speed=speed,
|
||||
idcode=idcode, is_programmable=True,
|
||||
)
|
||||
|
||||
|
||||
def _get_vivado_path(vitis_path: str) -> str | None:
|
||||
"""Find the Vivado executable path.
|
||||
|
||||
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 get_zynq_status_dict(result: ZynqCheckResult) -> dict:
|
||||
"""Convert a ZynqCheckResult to a serializable dictionary.
|
||||
|
||||
Args:
|
||||
result: ZynqCheckResult from check_zynq_jtag().
|
||||
|
||||
Returns:
|
||||
Dictionary suitable for JSON serialization or GUI display.
|
||||
"""
|
||||
return {
|
||||
"is_present": result.is_present,
|
||||
"ps_detected": result.ps_detected,
|
||||
@@ -367,23 +348,15 @@ def get_zynq_status_dict(result: ZynqCheckResult) -> dict:
|
||||
"hw_server_url": result.hw_server_url,
|
||||
"error": result.error,
|
||||
"jtag_chain": [
|
||||
{
|
||||
"name": dev.name,
|
||||
"is_zynq": dev.is_zynq,
|
||||
"is_arm_dap": dev.is_arm_dap,
|
||||
"is_pl_device": dev.is_pl_device,
|
||||
}
|
||||
for dev in result.jtag_chain
|
||||
{"name": d.name, "idcode": d.idcode,
|
||||
"is_zynq": d.is_zynq, "is_arm_dap": d.is_arm_dap,
|
||||
"is_pl_device": d.is_pl_device}
|
||||
for d in result.jtag_chain
|
||||
],
|
||||
"devices": [
|
||||
{
|
||||
"name": dev.name,
|
||||
"part_number": dev.part_number,
|
||||
"family": dev.family,
|
||||
"speed": dev.speed,
|
||||
"idcode": dev.idcode,
|
||||
"is_programmable": dev.is_programmable,
|
||||
}
|
||||
for dev in result.devices
|
||||
{"name": d.name, "part_number": d.part_number,
|
||||
"family": d.family, "speed": d.speed,
|
||||
"idcode": d.idcode, "is_programmable": d.is_programmable}
|
||||
for d in result.devices
|
||||
],
|
||||
}
|
||||
|
||||
@@ -1,14 +0,0 @@
|
||||
#-----------------------------------------------------------
|
||||
# Vivado v2023.2 (64-bit)
|
||||
# SW Build 4029153 on Fri Oct 13 20:13:54 MDT 2023
|
||||
# IP Build 4028589 on Sat Oct 14 00:45:43 MDT 2023
|
||||
# SharedData Build 4025554 on Tue Oct 10 17:18:54 MDT 2023
|
||||
# Start of session at: Wed Jun 10 15:01:29 2026
|
||||
# Process ID: 1790122
|
||||
# Current directory: /data/work/prj/Zynq_Flasher
|
||||
# Command line: vivado -mode batch -source /tmp/zynq_jtag_check_1789776.tcl
|
||||
# Log file: /data/work/prj/Zynq_Flasher/vivado.log
|
||||
# Journal file: /data/work/prj/Zynq_Flasher/vivado.jou
|
||||
# Running On: mkb, OS: Linux, CPU Frequency: 5069.987 MHz, CPU Physical cores: 32, Host memory: 134146 MB
|
||||
#-----------------------------------------------------------
|
||||
source /tmp/zynq_jtag_check_1789776.tcl
|
||||
@@ -1,14 +0,0 @@
|
||||
#-----------------------------------------------------------
|
||||
# Vivado v2023.2 (64-bit)
|
||||
# SW Build 4029153 on Fri Oct 13 20:13:54 MDT 2023
|
||||
# IP Build 4028589 on Sat Oct 14 00:45:43 MDT 2023
|
||||
# SharedData Build 4025554 on Tue Oct 10 17:18:54 MDT 2023
|
||||
# Start of session at: Wed Jun 10 15:02:25 2026
|
||||
# Process ID: 1790665
|
||||
# Current directory: /data/work/prj/Zynq_Flasher
|
||||
# Command line: vivado -mode batch -source /tmp/zynq_jtag_check_1789776.tcl
|
||||
# Log file: /data/work/prj/Zynq_Flasher/vivado.log
|
||||
# Journal file: /data/work/prj/Zynq_Flasher/vivado.jou
|
||||
# Running On: mkb, OS: Linux, CPU Frequency: 5073.251 MHz, CPU Physical cores: 32, Host memory: 134146 MB
|
||||
#-----------------------------------------------------------
|
||||
source /tmp/zynq_jtag_check_1789776.tcl
|
||||
Binary file not shown.
Reference in New Issue
Block a user