Compare commits

...
42 Commits
Author SHA1 Message Date
yuysh 97a116b36f fix: cancel SD timer on new log output; change popup prompt to QSPI 2026-06-12 18:29:36 +08:00
yuysh fce940926f feat: SD boot mode detection — kill hw_server + popup after 30s if Boot mode is SD 2026-06-12 18:16:35 +08:00
yuysh c0f7219d30 fix: replace Popen readline loop with subprocess.run for Windows pipe reliability; add close_hw before quit in TCL; revert console=True 2026-06-12 17:46:11 +08:00
yuysh 4da03e97d8 fix: open_hw_manager→open_hw for 2018.3 batch mode compat; revert console=True 2026-06-12 17:37:54 +08:00
yuysh c9eae220f0 Fix naming 2026-06-12 17:24:01 +08:00
yuysh e09f16a3aa fix: Xilinx 2018.3 tool discovery — add SDK dir, .bat PATH search, older versions 2026-06-12 17:20:49 +08:00
yuysh 47ea614ba2 fix: switch to console=False (GUI only, no terminal window) 2026-06-12 16:56:56 +08:00
yuysh 13ed0d89f8 fix: resolve config.yaml path to exe directory when frozen (PyInstaller) 2026-06-12 16:55:35 +08:00
yuysh 0d66f65ef2 feat: add PyInstaller spec + Windows build script (console=True) 2026-06-12 16:51:26 +08:00
yuysh 4c01acdf69 fix: print log messages to terminal in addition to GUI 2026-06-12 16:38:51 +08:00
yuysh c0cac5b2c5 fix: ping_ip bool unpack, skip sub-step visibility, IP 192.168.100.11 after Step 3
- ping_ip returns bool, not tuple — fixed 2 unpack errors
- TftpSubStepFrame: _skipped set prevents set_phase from overwriting Skip
- set_expanded(True) before skip marking so UI shows it immediately
- After Step 3 success, fix IP to 192.168.100.11
2026-06-12 16:35:32 +08:00
yuysh 5ed195e2d6 fix: skip marking, fixed panel ratio, dialog hints, 30s boot, ping+UDP fallback
- TftpSubStepFrame.set_step_skipped(): ◌ Skip in WARNING color
- _reload_config(): suppress callbacks to avoid duplicate log/status wipe
- _log_all_file_info(): removed clear_info(), updates in-place per-key
- Removed duplicate _log_file_info method
- main_frame grid: uniform col group for strict 7:3 ratio
- FileSelector browse title shows expected file type hint
- boot_wait_delay: default 10→30s
- Step 4 IP fallback: ping first, then UDP scan, extra boot_delay retry
2026-06-12 16:10:28 +08:00
yuysh 639f2e3ce6 feat: 5 improvements — file info, download verify, UART warning, status display, COM port reuse
- file_utils: compute_file_info() — CRC32, human-readable size, mtime
- StatusDisplay: simplified to info-only key-value panel; status messages
  removed (log-only); file info shown in status panel at startup/select
- config: added download_verify checkbox (default ON), skips Step 4.2
- UART warning: CTkTextbox with word-wrap and auto-height
- COM port: check_uart_available / test_serial_version accept optional
  UartMonitor to avoid redundant serial opens (Error 13 on Windows)
- layout: left:right 7:3 ratio, config panel stretches full width
2026-06-12 15:47:40 +08:00
yuysh 9502bba636 fix: config save broken (relative_to→relpath), startup validation, auto-create, untrack config.yaml
- _relative_path(): replace Path.relative_to() with os.path.relpath()
  (ValueError crossing filesystems silently returned absolute paths)
- _sync_config_from_ui(): always derive relative from get_path() absolute,
  never trust stale FileSelector._relative_path after Browse
- validate_paths(): check all file paths exist on startup, clear missing
- Clear UI selectors after validation to prevent re-save of stale paths
- _load_config(): auto-create blank config.yaml when missing on disk
- .gitignore: add config.yaml (user-specific, must not be tracked)
2026-06-12 15:02:09 +08:00
yuysh 9238cbc63e fix: correct indentation errors in widgets.py (TftpSubStepFrame, UartMonitorWindow) 2026-06-12 14:14:17 +08:00
yuysh f330f8c276 fix: remove 'uboot'/'u-boot' from fail keywords — too broad
Xilinx program_flash often outputs informational lines containing
'uboot' (mini u-boot image file, U-Boot version banner) which are
not errors. The 'failed' keyword already catches 'U-Boot failed'
and 'uboot failed' in actual failure scenarios.
2026-06-11 18:13:30 +08:00
yuysh 04ad61e694 fix: convert all TCL-embedded paths to forward slashes (cross-platform)
Windows backslashes in TCL strings are interpreted as escape sequences
(\f → form feed, \t → tab). This caused 'dow "Files\fsbl..."' to
fail with garbled paths — TCL couldn't find the file.

Added _tcl_path() helper: replaces \ with / in any path before
embedding in TCL. Forward slashes work on all platforms including
Windows. Applied to all 7 path sites: source, fpga -file, dow.
2026-06-11 18:11:44 +08:00
yuysh 4ca15c1190 fix: detect flash failure from output even when returncode is 0
program_flash can return exit code 0 while outputting errors like
'uboot failed' or 'cannot open'. Previously only returncode was
checked, so these silent failures caused the workflow to continue
past a failed erase/program step.

Now scans combined stdout+stderr for failure keywords
(uboot, failed, cannot, abort, invalid, unsupported, timeout)
and returns success=False. Also checks returncode != 0 as fallback.
2026-06-11 18:00:49 +08:00
yuysh 4453fc8e12 feat: validate JTAG Zynq part matches config.zynq_part (default XC7Z100)
Step 1 now compares detected part (e.g. XC7Z035) against config's
zynq_part. If mismatch: log error, return False to block subsequent
steps. Also returns False when no Zynq detected (was always True).
2026-06-11 17:38:27 +08:00
yuysh 799da5b2f3 fix: Windows common paths try .bat before .exe (Xilinx tools are .bat)
Vitis/Vivado on Windows uses .bat wrappers, not .exe. The hardcoded
fallback paths in _find_executable step 4 only looked for .exe files.
Also expanded to try both 2023.2 and 2022.2 + Vitis/Vivado combos.
2026-06-11 17:34:36 +08:00
yuysh 7e8427d522 fix: redirect Vivado .jou/.log to temp dir with -tempDir
Vivado batch mode creates vivado.jou, vivado.log, and backup files
in cwd by default. Use -tempDir to redirect all temp output to the
system temp directory, and clean up afterwards.
2026-06-11 17:26:41 +08:00
yuysh 66c04da5e0 fix: stream Vivado output live, remove broken normalize_port
JTAG: Start hw_server explicitly (Windows needs it), then run Vivado
TCL with Popen + line-by-line stdout streaming via callback so user
can see Vivado progress in real-time (not just on failure).

COM: Remove normalize_port — pyserial >= 3.0 handles COM>=10 prefix
internally. The manual \\.\ prefix was causing PermissionError(13)
on Windows where pyserial's own handling conflicted.
2026-06-11 17:24:31 +08:00
yuysh 954aed7c0d chore: expand .gitignore for all Vivado/Xilinx temp files
Added: vivado.jou, vivado.log, vivado_*.backup.log, vivado_*.str,
hs_err_pid*.log (Java crash), webtalk*, usage_statistics_*
2026-06-11 17:03:18 +08:00
yuysh 474e3f122b fix: use Vivado TCL for JTAG check — xsct 'targets' has broken TCL list parsing
xsct's [targets] returns unquoted TCL lists where multi-word names
(e.g. 'ARM Cortex-A9 MPCore #0') are split into individual words
by foreach, making it impossible to reconstruct device names.

Vivado TCL with open_hw_manager works reliably:
- Cold start (no hw_server): 7.1s on Linux
- hw_server already running: 6.8s on Linux
- Provides IDCODE (converted from Vivado binary to hex), device names

Smart hw_server lifecycle:
- Check port 3121 before starting — reuse if already running
- Start hw_server only when needed (poll port up to 3s)
- Don't kill hw_server that we didn't start

Output format parsed: DEVICE:arm_dap_0 IDCODE:010010111010...0111
→ arm_dap_0 idcode=4BA00477 PS detected
→ xc7z100_1 idcode=03736093 PL detected → XC7Z100

Tested on Linux 2023.2, cold + warm start, confirmed working.
2026-06-11 17:02:23 +08:00
yuysh dd35ab0846 fix: use xsct 'jtag targets' for raw JTAG chain with IDCODE
The previous TCL used 'targets' which returns XSCT target IDs (numbers
like 1, 2, APU, xc7z100) — each word was parsed as a separate device,
producing '1 (Other)', 'APU (Other)', etc.

Switch to 'jtag targets' which outputs raw JTAG chain devices:
  1  xc7z100 (idcode 03736093 irlen 4)
  2  arm_dap (idcode 4BA00477 irlen 4)

Changes:
- New _JTAG_LINE_RE pattern for parsing jtag targets output
- Zynq pattern relaxed for xsct names (xc7z100 not xc7z100_1)
- ARM DAP pattern: arm_dap (not arm_dap_N)
- JtagDevice now carries idcode field
- ZynqDevice propagates idcode + is_programmable=True
- Removed unused tempfile/Path/os imports
2026-06-11 16:45:37 +08:00
yuysh 86d65641cc chore: gitignore Vivado/JTAG temp files, delete existing ones 2026-06-11 16:37:21 +08:00
yuysh 7716171aeb fix: use xsct (not Vivado) for JTAG check + normalize COM>=10 ports on Windows
**JTAG check**: Switch from Vivado to xsct
- xsct is already proven to work (JTAG reboot succeeds)
- Much lighter than Vivado (no GUI infrastructure / open_hw_manager)
- Uses -eval inline TCL: connect; targets; disconnect; quit
- Removes dead tempfile/Path/os imports

**Windows COM port**: Add normalize_port() to serial_monitor.py
- COM ports >= 10 require NT namespace prefix (\.\COM11)
- Without it: PermissionError(13, '拒绝访问。')
- Applied at all 7 call sites: serial_monitor (5), reboot_manager, widgets
2026-06-11 16:29:32 +08:00
yuysh 91417f609e fix: start hw_server explicitly before Vivado JTAG scan
Previously the TCL script relied on Vivado's connect_hw_server to
auto-start hw_server. On Windows this either never worked (hw_server
never shows in Task Manager) or hung indefinitely during Vivado's
slow batch-mode initialisation.

Root cause: hw_server was never started.

Changes:
- vitis_checker.py: add 'hw_server' to TOOL_ALIASES + get_hw_server_path()
- zynq_checker.py: start hw_server via Popen before Vivado, kill in finally
  - 2s sleep for port binding after start
  - timeout from config.step_timeouts.check_env (default bumped to 120s)
  - clean terminate/kill sequence in finally
- config_manager.py: check_env default 30 → 120s
2026-06-11 15:57:53 +08:00
yuysh 72d0eceb8d fix: don't restart Boot Verify pulse after step 4 already completed it
Step 4's _run_step_4_load_main_program() already marks each sub-step
individually via set_step_complete/set_step_error. But the outer
_run_steps_worker() called set_phase('Boot Verify') on success, which
re-entered the 'running' phase and started a fresh pulse animation
(because _stop_pulse cleared _pulse_job).

Fix: on success, do nothing (sub-steps are already green). Only set
error on failure.
2026-06-11 15:49:10 +08:00
yuysh 1619dc194f fix: two-phase boot wait — don't stop on version when IP still pending
Previously the boot wait loop broke on the first detection of EITHER
IP or version. The Zynq firmware outputs its version string early in
the boot sequence (e.g. firmware header), while the board IP arrives
in a later burst via 'Boot=, IP=192.168.100.13, Ver=...'. The loop
would break on the version-only signal, leaving IP empty.

Changes:
1. Phase 1: log boot signal (version) but keep polling for IP
2. Phase 2: break only when IP address is actually found
3. After timeout: final check of latest_info for late-arriving IP
4. UART buffer fallback: parse all accumulated lines for IP
   (handles lines that arrived between the last poll and exit)
5. 3s pre-scan delay before UDP discover to let network stack settle
   (board's UDP server starts at the very end of boot sequence)
2026-06-11 15:41:21 +08:00
yuysh 267cae93b6 fix: use tempfile.gettempdir() instead of hardcoded /tmp/ for JTAG TCL file
/tmp/ doesn't exist on Windows — Path('/tmp/...') resolves to C:\tmp\ which
fails with 'No such file or directory'. Use tempfile.gettempdir() for
cross-platform temp directory resolution.

Also cleaned up the get_vivado_path() call (removed obsolete vitis_path kwarg).
2026-06-11 15:36:04 +08:00
yuysh 17c8f2d68b refactor: replace all config.vitis_path with config.xilinx_path across codebase
Affected files:
- flash_programmer.py (2 call sites) — get_program_flash_path
- bitstream_programmer.py (3 get_xsct_path + 1 get_vivado_path call sites)
  Also: removed duplicate _get_vivado_path() (now uses shared one from vitis_checker)
  Also: removed unused shutil import
- reboot_manager.py (1 call site) — get_xsct_path

All now use getattr(config, 'xilinx_path', '') consistently.
The only remaining 'config.vitis_path' reference is in vitis_checker.py's
check_vitis() itself (safe legacy fallback for old configs).
2026-06-11 15:33:54 +08:00
yuysh 61f0e6cd2b 🐛 fix: Step 1 JTAG check uses shared vivado path finder + build_tool_command
Root cause (2 bugs):
1. check_zynq_jtag() used config.vitis_path which no longer exists
   (config now uses xilinx_path) → always got empty path → 'Vivado not found'
2. _get_vivado_path() only checked hardcoded Linux paths, no version
   scanning, no .bat support on Windows

Fix:
- Added get_vivado_path() in vitis_checker.py (consistent with
  get_xsct_path, get_program_flash_path, etc.)
- Uses _find_executable() shared logic: scans Vivado/*/bin +
  Vitis/*/bin, picks newest version, supports .bat on Windows
- zynq_checker.py now uses get_vivado_path() + build_tool_command()
  for proper environment setup (settings64 sourcing on Linux)
- Removed obsolete _get_vivado_path() + unused shutil import
2026-06-11 15:30:43 +08:00
yuysh 6d686964e2 🐛 fix: Step 4 boot wait uses UART window monitor data, not just main monitor
Root cause: When UART window is open, self._uart_monitor is None
(the window creates its own UartMonitor). Step 4 boot wait loop
only checked self._uart_monitor.latest_info → always saw empty data
→ fell through to IP scanning even though UART had detected the IP.

Fix:
- UartMonitorWindow.get_latest_info() exposes parsed BootInfo
- Step 4 checks both: self._uart_monitor first, then UART window
- Null guard on info before accessing .ip_address
2026-06-11 15:21:36 +08:00
yuysh da94cc6cdc 🐛 fix: lambda closure crash + .bat version noise on Windows
Bug 1 — UartMonitorWindow crash on serial open failure:
- lambda: self._show_unavailable(str(e))  →  lambda e=e: ...
- Python 3 deletes 'e' after except block; default arg captures it

Bug 2 — xsct version shows 'ECHO 处于关闭状态。' on Windows:
- .bat wrappers output 'ECHO is off' / 'ECHO 处于关闭状态。' noise
- _is_batch_noise() filters known patterns (EN+ZH)
- Version regex now scans all output lines, skipping noise lines
- Path-based fallback (2023.2) catches any remaining undetectable cases
2026-06-11 15:18:08 +08:00
yuysh 70c88815cc feat: auto-discover Zynq IP after reboot (UART or UDP scan)
After Step 4 reboot, the Zynq may obtain a different IP via DHCP.

IP discovery strategy:
1. UART available → parse IP from boot output (fast, zero network traffic)
2. UART unavailable → scan 192.168.100.11–30 via UDP ver() probe
   (port = last_octet - 3, 0.5s timeout each, ~10s max for 20 IPs)

If IP changed → auto-update config.zynq_ip + UI field
2026-06-11 15:11:59 +08:00
yuysh 32b201cf8e feat: Windows .bat support + settings64 env setup for Xilinx tools
Tool discovery:
- On Windows, scan .bat files in addition to .exe
  (Xilinx Windows tools use .bat wrappers, not .exe)
- Prefer .exe over .bat when both exist (same version dir)

Environment setup:
- _find_settings_script(): walk up from bin/ to find settings64.sh/.bat
- build_tool_command(): wrap Linux tool invocation with 'source settings64.sh'
  Windows .bat wrappers handle env internally, no extra wrapping needed
- reboot_manager.py now uses build_tool_command() for proper env

Example output (Linux):
  build_tool_command('/data/xilinx/Vitis/2023.2/bin/xsct', 'script.tcl')
  → ['bash', '-c', 'source .../settings64.sh && .../xsct script.tcl']
2026-06-11 15:08:25 +08:00
yuysh 21acc16c8c feat: smart Xilinx version scanning — pick newest, log all versions
_version discovery:
- _scan_xilinx_root() collects ALL versions under Vitis/*/bin/ and Vivado/*/bin/
- _parse_version_tuple() semantic version comparison (2023.2 > 2018.3)
- Defaults to newest version, Vitis preferred over Vivado for same version
- _find_executable() search order: PATH → legacy vitis_path → scan xilinx_root → common paths

_version reporting:
- _get_tool_version(): tries -version flag first, falls back to path parsing
- _version_from_path(): extracts version from install path (e.g., .../Vitis/2023.2/bin/xsct → 2023.2)
- Step 1 logs each tool with version: ✓ xsct: /path [v2023.2]
- If multiple versions found, lists all: xsct versions found: 2023.2 (Vitis), 2018.3 (Vivado)

Cross-platform:
- Handles Windows path separators via Path()
- Expanded common Windows install paths
2026-06-11 14:56:40 +08:00
yuysh 475042c85c 🐛 fix: JTAG reboot 'Already running' error after rst -system
rst -system inherently resumes ARM execution on Zynq.
The subsequent 'con' command fails with 'Already running'
because the core is already executing after system reset.

Fix: wrap 'con' in 'catch {}' — if core is already running
the error is silently caught; if stopped, con resumes it.
xsct exits 0 (success) in both cases.
2026-06-11 14:48:45 +08:00
yuysh f6cf5de96f feat: UART boot wait, serial validation, xsct logging, layout fix
Config:
- Add boot_wait_delay (default 10s) for post-reboot wait time

Layout:
- Restore row weights for proportional scaling when maximized
- CTkScrollableFrame handles overflow when window is small

Step 3:
- Log xsct/vivado tool output (last 20 lines per result)
  for detailed diagnostics

Step 4:
- UART boot wait: monitor for IP/version via UART after reboot
- If UART unavailable, fall back to config.boot_wait_delay
- Reduces unnecessary wait time when boot is detected early

Serial:
- Real-time port validation on selection change (CTkComboBox command)
- Persistent UART unavailable warning until valid port selected
- Auto-restart UART monitor on port switch
- Graceful cross-platform handling
2026-06-11 14:44:58 +08:00
yuysh 9b07a4d5ca feat: add scrollbar via CTkScrollableFrame for window resize
- Replace main_frame CTkFrame → CTkScrollableFrame for vertical scrollbar
- Remove main_frame.grid_rowconfigure weight — content natural height
- Remove right_frame.grid_rowconfigure weight — prevents squashing
- Left column (workflow + log) retains weight=1 for dominant width
- Right column (config/serial/utility/status) keeps natural narrow width
2026-06-11 14:31:17 +08:00
yuysh 84142bb229 🎨 style: compact 2-col Utility panel, both reboot buttons red
- Reboot (UDP) now uses DANGER_COLOR like Reboot (JTAG)
- 2×2 grid layout: reboot row + version row
- Buttons stretch with sticky=ew, equal column weights
- Title spans both columns
2026-06-11 14:27:48 +08:00
18 changed files with 2335 additions and 724 deletions
+16 -1
View File
@@ -6,6 +6,7 @@ dist/
build/ build/
*.egg-info/ *.egg-info/
*.egg *.egg
*.exe
# ── Python ─────────────────────────────────────────────────── # ── Python ───────────────────────────────────────────────────
__pycache__/ __pycache__/
@@ -33,8 +34,22 @@ htmlcov/
# ── Opencode (internal tooling) ────────────────────────────── # ── Opencode (internal tooling) ──────────────────────────────
.opencode/ .opencode/
# ── Config overrides (user-specific) ───────────────────────── # ── Config (user-specific, auto-generated if missing) ─────────
config.yaml
config/zynq_flasher.yaml config/zynq_flasher.yaml
# ── Logs ───────────────────────────────────────────────────── # ── Logs ─────────────────────────────────────────────────────
*.log *.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
+63
View File
@@ -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,
)
+47
View File
@@ -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
View File
@@ -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
View File
@@ -11,14 +11,13 @@ Compatible with Xilinx 2018.3 — 2023.2.
from __future__ import annotations from __future__ import annotations
import os import os
import shutil
import subprocess import subprocess
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Callable from typing import Callable
from config_manager import Config from config_manager import Config
from vitis_checker import get_xsct_path from vitis_checker import get_xsct_path, get_vivado_path
# ── Types ─────────────────────────────────────────────────────────── # ── Types ───────────────────────────────────────────────────────────
@@ -38,33 +37,13 @@ ProgressCallback = Callable[[str, str], None] | None
# ── Helpers ───────────────────────────────────────────────────────── # ── Helpers ─────────────────────────────────────────────────────────
def _get_vivado_path(vitis_path: str) -> str | None: def _tcl_path(path: str | Path) -> str:
"""Find the Vivado executable. """Convert a filesystem path to TCL-safe format.
Uses shutil.which() for PATH resolution, then common install paths. TCL interprets backslashes as escape sequences (\\f, \\t, \\n).
Forward slashes work on all platforms including Windows.
Args:
vitis_path: Optional Vitis installation directory.
Returns:
Path to vivado executable, or None.
""" """
found = shutil.which("vivado") return str(path).replace("\\", "/")
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
def _find_ps7_init_tcl(config: Config) -> str: def _find_ps7_init_tcl(config: Config) -> str:
@@ -167,16 +146,15 @@ def program_bitstream(
# Try xsct first (preferred) # Try xsct first (preferred)
xsct = get_xsct_path( xsct = get_xsct_path(
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "", xilinx_root=getattr(config, "xilinx_path", ""),
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
) )
if xsct: if xsct:
timeout = config.step_timeouts.get("bitstream", 60) timeout = config.step_timeouts.get("bitstream", 60)
return _program_with_xsct(xsct, bit_path, callback, ps7_tcl, timeout) return _program_with_xsct(xsct, bit_path, callback, ps7_tcl, timeout)
# Fallback to Vivado # Fallback to Vivado
vivado = _get_vivado_path( vivado = get_vivado_path(
config.vitis_path if hasattr(config, 'vitis_path') else "" xilinx_root=getattr(config, "xilinx_path", ""),
) )
if vivado: if vivado:
return _program_with_vivado(vivado, bit_path, callback) return _program_with_vivado(vivado, bit_path, callback)
@@ -218,11 +196,11 @@ def _program_with_xsct(
lines = ["connect"] lines = ["connect"]
if ps7_init_tcl and Path(ps7_init_tcl).exists(): if ps7_init_tcl and Path(ps7_init_tcl).exists():
lines.append("targets 1") lines.append("targets 1")
lines.append(f'source "{ps7_init_tcl}"') lines.append(f'source "{_tcl_path(ps7_init_tcl)}"')
lines.append("ps7_init") lines.append("ps7_init")
if callback: if callback:
callback("progress", "PS initialized (clocks, DDR, MIO)") 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(): if ps7_init_tcl and Path(ps7_init_tcl).exists():
lines.append("ps7_post_config") lines.append("ps7_post_config")
if callback: if callback:
@@ -293,12 +271,14 @@ def _program_with_vivado(
callback("progress", "Using Vivado Hardware Manager...") callback("progress", "Using Vivado Hardware Manager...")
tcl_content = f""" tcl_content = f"""
open_hw_manager open_hw
connect_hw_server connect_hw_server
open_hw_target open_hw_target
current_hw_device [lindex [get_hw_devices] 0] current_hw_device [lindex [get_hw_devices] 0]
program_hw_device -file "{bit_path}" program_hw_device -file "{bit_path}"
refresh_hw_device refresh_hw_device
close_hw
disconnect_hw_server
quit quit
""" """
tcl_path = bit_path.parent / "temp_program.tcl" tcl_path = bit_path.parent / "temp_program.tcl"
@@ -364,8 +344,7 @@ def run_elf(
callback("start", f"Running ELF: {elf_path.name}...") callback("start", f"Running ELF: {elf_path.name}...")
xsct = get_xsct_path( xsct = get_xsct_path(
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "", xilinx_root=getattr(config, "xilinx_path", ""),
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
) )
if not xsct: if not xsct:
return BitstreamResult( return BitstreamResult(
@@ -413,13 +392,13 @@ def _run_elf_via_fsbl(
tcl = "\n".join([ tcl = "\n".join([
"connect", "connect",
"targets 2", "targets 2",
"dow \"%s\"" % fsbl_path, "dow {%s}" % _tcl_path(fsbl_path),
"puts \"FSBL downloaded\"", "puts \"FSBL downloaded\"",
"con", "con",
"after 3000", "after 3000",
"stop", "stop",
"puts \"FSBL done, CPU halted\"", "puts \"FSBL done, CPU halted\"",
"dow \"%s\"" % elf_path, "dow {%s}" % _tcl_path(elf_path),
"mwr 0xF800025C 0x00000001", "mwr 0xF800025C 0x00000001",
"con", "con",
"puts \"App running\"", "puts \"App running\"",
@@ -493,7 +472,7 @@ def _run_elf_direct(
tcl = "\n".join([ tcl = "\n".join([
"connect", "connect",
"targets 2", "targets 2",
"dow \"%s\"" % elf_path, "dow {%s}" % _tcl_path(elf_path),
"mwr 0xF800025C 0x00000001", "mwr 0xF800025C 0x00000001",
"con", "con",
"exit", "exit",
@@ -579,8 +558,7 @@ def full_bitstream_program(
message=f"FSBL not found: {fsbl_path or '(not configured)'}")] message=f"FSBL not found: {fsbl_path or '(not configured)'}")]
xsct = get_xsct_path( xsct = get_xsct_path(
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "", xilinx_root=getattr(config, "xilinx_path", ""),
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
) )
if not xsct: if not xsct:
return [BitstreamResult(step="bitstream", success=False, return [BitstreamResult(step="bitstream", success=False,
@@ -653,18 +631,18 @@ def full_bitstream_program(
"rst -processor", "rst -processor",
"after 500", "after 500",
'puts "INFO: Downloading FSBL..."', 'puts "INFO: Downloading FSBL..."',
'dow "%s"' % fsbl_path, 'dow {%s}' % _tcl_path(fsbl_path),
'puts "INFO: Running FSBL for hardware initialization..."', 'puts "INFO: Running FSBL for hardware initialization..."',
"con", "con",
"after 2000", "after 2000",
"stop", "stop",
'puts "INFO: Hardware initialization completed."', 'puts "INFO: Hardware initialization completed."',
'puts "INFO: Downloading Bitstream..."', 'puts "INFO: Downloading Bitstream..."',
'fpga -f "%s"' % bit_path, 'fpga -f {%s}' % _tcl_path(bit_path),
"after 1000", "after 1000",
'targets -set -nocase -filter {name =~ "arm*#0"}', 'targets -set -nocase -filter {name =~ "arm*#0"}',
'puts "INFO: Downloading Application ELF..."', 'puts "INFO: Downloading Application ELF..."',
'dow "%s"' % elf_path, 'dow {%s}' % _tcl_path(elf_path),
'puts "INFO: Executing Application..."', 'puts "INFO: Executing Application..."',
"con", "con",
'puts "SUCCESS: Zynq target is running the specified application."', 'puts "SUCCESS: Zynq target is running the specified application."',
+43 -3
View File
@@ -17,6 +17,7 @@ import yaml
DEFAULT_CONFIG: dict[str, Any] = { DEFAULT_CONFIG: dict[str, Any] = {
"xilinx_path": "", "xilinx_path": "",
"zynq_ip": "192.168.100.11", "zynq_ip": "192.168.100.11",
"zynq_part": "XC7Z100",
"tftp_upload_name": "z7bin", "tftp_upload_name": "z7bin",
"serial_port": "", "serial_port": "",
"serial_baudrate": 115200, "serial_baudrate": 115200,
@@ -27,14 +28,16 @@ DEFAULT_CONFIG: dict[str, Any] = {
"flash_type": "qspi-x4-single", "flash_type": "qspi-x4-single",
"flash_model": "s25fl256s1", "flash_model": "s25fl256s1",
"erase_all": False, "erase_all": False,
"download_verify": True,
"firmware_bin_path": "", "firmware_bin_path": "",
"reboot_timeout": 30, "reboot_timeout": 30,
"ping_timeout": 5, "ping_timeout": 5,
"ping_count": 3, "ping_count": 3,
"uart_delay": 3, "uart_delay": 3,
"inter_step_delay": 2, "inter_step_delay": 2,
"boot_wait_delay": 30,
"step_timeouts": { "step_timeouts": {
"check_env": 30, "check_env": 120,
"flash_erase": 120, "flash_erase": 120,
"flash_program": 600, "flash_program": 600,
"bitstream": 60, "bitstream": 60,
@@ -54,6 +57,7 @@ class Config:
xilinx_path: str = "" xilinx_path: str = ""
zynq_ip: str = "192.168.100.11" zynq_ip: str = "192.168.100.11"
zynq_part: str = "XC7Z100"
tftp_upload_name: str = "z7bin" tftp_upload_name: str = "z7bin"
serial_port: str = "" serial_port: str = ""
serial_baudrate: int = 115200 serial_baudrate: int = 115200
@@ -64,12 +68,14 @@ class Config:
flash_type: str = "qspi-x4-single" flash_type: str = "qspi-x4-single"
flash_model: str = "s25fl256s1" flash_model: str = "s25fl256s1"
erase_all: bool = False erase_all: bool = False
download_verify: bool = True
firmware_bin_path: str = "" firmware_bin_path: str = ""
reboot_timeout: int = 30 reboot_timeout: int = 30
ping_timeout: int = 5 ping_timeout: int = 5
ping_count: int = 3 ping_count: int = 3
uart_delay: int = 3 uart_delay: int = 3
inter_step_delay: int = 2 inter_step_delay: int = 2
boot_wait_delay: int = 30
step_timeouts: dict[str, int] = field(default_factory=lambda: {}) step_timeouts: dict[str, int] = field(default_factory=lambda: {})
# Internal: path to the config file on disk # Internal: path to the config file on disk
@@ -172,12 +178,14 @@ class Config:
"flash_type": self.flash_type, "flash_type": self.flash_type,
"flash_model": self.flash_model, "flash_model": self.flash_model,
"erase_all": self.erase_all, "erase_all": self.erase_all,
"download_verify": self.download_verify,
"firmware_bin_path": self._relative_path(self.firmware_bin_path), "firmware_bin_path": self._relative_path(self.firmware_bin_path),
"reboot_timeout": self.reboot_timeout, "reboot_timeout": self.reboot_timeout,
"ping_timeout": self.ping_timeout, "ping_timeout": self.ping_timeout,
"ping_count": self.ping_count, "ping_count": self.ping_count,
"uart_delay": self.uart_delay, "uart_delay": self.uart_delay,
"inter_step_delay": self.inter_step_delay, "inter_step_delay": self.inter_step_delay,
"boot_wait_delay": self.boot_wait_delay,
"step_timeouts": self.step_timeouts, "step_timeouts": self.step_timeouts,
} }
return data return data
@@ -185,8 +193,11 @@ class Config:
def _relative_path(self, path: str) -> str: def _relative_path(self, path: str) -> str:
"""Convert an absolute path to a relative path from the config file. """Convert an absolute path to a relative path from the config file.
Uses :func:`os.path.relpath` which works across filesystem
boundaries (unlike :meth:`Path.relative_to`).
Args: Args:
path: Absolute path string. path: Absolute or relative path string.
Returns: Returns:
Relative path string, or the original if conversion fails. Relative path string, or the original if conversion fails.
@@ -196,7 +207,8 @@ class Config:
try: try:
target = Path(path) target = Path(path)
if target.is_absolute(): if target.is_absolute():
return str(target.relative_to(self._config_path.parent)) base = str(self._config_path.parent)
return os.path.relpath(str(target), base)
except (ValueError, OSError): except (ValueError, OSError):
pass pass
return path return path
@@ -214,6 +226,34 @@ class Config:
return Path() return Path()
return (self._config_path.parent / relative_path).resolve() return (self._config_path.parent / relative_path).resolve()
def validate_paths(self) -> list[str]:
"""Validate file paths in config and clear those that don't exist.
Checks each file path field, resolves it against the config
directory, and clears it if the target file does not exist
on the current filesystem.
Returns:
List of field names that were cleared.
"""
cleared: list[str] = []
path_fields = [
"bootloader_bit_path",
"bootloader_elf_path",
"bootloader_bin_path",
"fsbl_elf_path",
"firmware_bin_path",
]
for attr in path_fields:
val = getattr(self, attr, "")
if not val:
continue
resolved = self.resolve_path(val)
if not resolved.is_file():
setattr(self, attr, "")
cleared.append(attr)
return cleared
@property @property
def config_path(self) -> Path | None: def config_path(self) -> Path | None:
"""Return the path to the config file on disk.""" """Return the path to the config file on disk."""
+82
View File
@@ -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
View File
@@ -10,6 +10,7 @@ Compatible with Xilinx 2018.3 — 2023.2.
from __future__ import annotations from __future__ import annotations
import binascii import binascii
import re
import subprocess import subprocess
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -35,6 +36,27 @@ ProgressCallback = Callable[[str, str], None] | None
# ── Helpers ───────────────────────────────────────────────────────── # ── 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( def _run_command(
cmd: list[str], cmd: list[str],
@@ -113,8 +135,7 @@ def wipe_flash(
FlashResult with status. FlashResult with status.
""" """
flash_tool = get_program_flash_path( flash_tool = get_program_flash_path(
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "", xilinx_root=getattr(config, "xilinx_path", ""),
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
) )
if not flash_tool: if not flash_tool:
return FlashResult( return FlashResult(
@@ -168,12 +189,27 @@ def wipe_flash(
callback=callback, 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=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( return FlashResult(
step="wipe", step="wipe",
success=success, success=True,
message="Flash erased successfully" if success else "Flash erase failed", message="Flash erased successfully",
output=(result.stdout + result.stderr)[:4000], output=combined[:4000],
) )
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
return FlashResult(step="wipe", success=False, message="Flash erase timed out") 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( flash_tool = get_program_flash_path(
vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "", xilinx_root=getattr(config, "xilinx_path", ""),
xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
) )
if not flash_tool: if not flash_tool:
return FlashResult( return FlashResult(
@@ -254,12 +289,27 @@ def program_flash_bin(
try: try:
result = _run_command(cmd, timeout=config.step_timeouts.get("flash_program", 600), callback=callback) 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=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( return FlashResult(
step="program", step="program",
success=success, success=True,
message="Flash programmed successfully" if success else "Flash programming failed", message="Flash programmed successfully",
output=(result.stdout + result.stderr)[:4000], output=combined[:4000],
) )
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
return FlashResult(step="program", success=False, message="Flash programming timed out (600s)") return FlashResult(step="program", success=False, message="Flash programming timed out (600s)")
+942 -117
View File
File diff suppressed because it is too large Load Diff
+170 -48
View File
@@ -6,6 +6,7 @@ consistent styling from gui/styles.py.
from __future__ import annotations from __future__ import annotations
import os
import threading import threading
import customtkinter as ctk import customtkinter as ctk
@@ -436,54 +437,91 @@ class ProgressIndicator(ctk.CTkFrame):
class StatusDisplay(ctk.CTkFrame): class StatusDisplay(ctk.CTkFrame):
"""Status display widget for operation results. """Info display widget — persistent key-value file/device info.
Shows 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): def __init__(self, master, **kwargs):
"""Initialize the status display.
Args:
master: Parent widget.
"""
super().__init__(master, **kwargs) super().__init__(master, **kwargs)
self._info_items: list[tuple[str, str]] = []
self._create_widgets() self._create_widgets()
def _create_widgets(self) -> None: def _create_widgets(self) -> None:
"""Create the status display UI.""" self.grid_rowconfigure(0, weight=1)
self._status_label = ctk.CTkLabel( self.grid_columnconfigure(0, weight=1)
self._info_text = ctk.CTkTextbox(
self, self,
text="Ready",
font=FONT_BODY, font=FONT_BODY,
anchor="w", corner_radius=CORNER_RADIUS,
state="disabled",
wrap="word",
) )
self._status_label.pack( self._info_text.grid(
fill="x", padx=PADDING, pady=PADDING_SMALL row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL
) )
# Pre-configure tags
for name, color in [
("info_label", PRIMARY_COLOR),
("info_value", DESC_TEXT_COLOR if _is_dark() else DESC_TEXT_COLOR_LIGHT),
]:
self._info_text.tag_config(name, foreground=color)
def set_status(self, message: str, status: str = "info") -> None: def set_status(self, message: str, status: str = "info") -> None:
"""Update the status message and color. """Compatibility stub — status messages go to log now."""
def set_info(self, key: str, value: str) -> None:
"""Set or update a persistent info line.
Displays as ``key : value`` in the info block.
Args: Args:
message: Status message text. key: Label for the info line.
status: Status type for coloring ('info', 'success', 'error', 'warning'). value: Value string.
""" """
self._status_label.configure(text=message) for i, (k, _) in enumerate(self._info_items):
colors = { if k == key:
"info": INFO_COLOR, self._info_items[i] = (key, value)
"success": SUCCESS_COLOR, break
"error": DANGER_COLOR, else:
"warning": WARNING_COLOR, self._info_items.append((key, value))
} self._refresh_info()
self._status_label.configure(text_color=colors.get(status, INFO_COLOR))
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): class FileSelector(ctk.CTkFrame):
"""File selector widget with browse button and path display. """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__( def __init__(
@@ -506,7 +544,11 @@ class FileSelector(ctk.CTkFrame):
self._callback = callback self._callback = callback
self._file_types = file_types or [("All files", "*")] self._file_types = file_types or [("All files", "*")]
self._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) self._create_widgets(label_text)
@@ -524,10 +566,10 @@ class FileSelector(ctk.CTkFrame):
) )
label.grid(row=0, column=0, sticky="w", padx=(PADDING, PADDING_SMALL)) 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._entry = ctk.CTkEntry(
self, self,
textvariable=self._selected_path, textvariable=self._display_text,
font=FONT_MONO, font=FONT_MONO,
state="readonly", state="readonly",
) )
@@ -549,33 +591,60 @@ class FileSelector(ctk.CTkFrame):
def _browse(self) -> None: def _browse(self) -> None:
"""Open file dialog and set selected path.""" """Open file dialog and set selected path."""
import customtkinter
from tkinter import filedialog from tkinter import filedialog
file_path = filedialog.askopenfilename( file_path = filedialog.askopenfilename(
title=f"Select {self._selected_path.get() or 'file'}", title=f"Select {self._hint_text}",
filetypes=self._file_types, filetypes=self._file_types,
) )
if file_path: if file_path:
self.set_path(file_path) self.set_path(file_path)
def set_path(self, path: str) -> None: 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: Args:
path: Absolute path to the selected file. path: Absolute or relative path to the selected file.
""" """
self._selected_path.set(path) self._full_path = path
if self._callback: 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) self._callback(path)
def get_path(self) -> str: def get_path(self) -> str:
"""Get the currently selected file path. """Get the currently selected file path (absolute).
Returns: Returns:
Selected file path string. 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): class LogDisplay(ctk.CTkFrame):
@@ -619,6 +688,7 @@ class LogDisplay(ctk.CTkFrame):
""" """
self._text_widget.configure(state="normal") self._text_widget.configure(state="normal")
self._text_widget.insert("end", message + "\n") self._text_widget.insert("end", message + "\n")
self._text_widget.update_idletasks()
self._text_widget.see("end") self._text_widget.see("end")
self._text_widget.configure(state="disabled") self._text_widget.configure(state="disabled")
@@ -649,15 +719,15 @@ class SubStepFrame(ctk.CTkFrame):
"""Collapsible frame showing flash operation sub-steps with status. """Collapsible frame showing flash operation sub-steps with status.
Colors match StepHeader: gray pending, blue running (+pulse), 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: def __init__(self, parent: ctk.CTkFrame) -> None:
super().__init__(parent, fg_color="transparent") super().__init__(parent, fg_color="transparent")
self._collapsed = False self._collapsed = True
self._toggle_btn = ctk.CTkButton( 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, 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._toggle_btn.grid(row=0, column=0, padx=(4, 4), pady=(2, 2), sticky="w")
@@ -682,6 +752,16 @@ class SubStepFrame(ctk.CTkFrame):
self.grid_columnconfigure(1, weight=1) self.grid_columnconfigure(1, weight=1)
self._pulse_job: str | None = None 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: def _toggle(self) -> None:
self._collapsed = not self._collapsed self._collapsed = not self._collapsed
self._toggle_btn.configure(text="" if self._collapsed else "") self._toggle_btn.configure(text="" if self._collapsed else "")
@@ -696,7 +776,7 @@ class SubStepFrame(ctk.CTkFrame):
item["pct"].grid() item["pct"].grid()
def set_expanded(self, expanded: bool) -> None: def set_expanded(self, expanded: bool) -> None:
if expanded != self._collapsed: if expanded == self._collapsed:
self._toggle() self._toggle()
def set_phase(self, phase: str, pct: int = -1) -> None: def set_phase(self, phase: str, pct: int = -1) -> None:
@@ -783,19 +863,20 @@ class TftpSubStepFrame(ctk.CTkFrame):
Sub-steps: Upload, Download Verify, CRC Check, Reboot, Boot Verify. Sub-steps: Upload, Download Verify, CRC Check, Reboot, Boot Verify.
Colors match StepHeader: gray pending, blue running (+pulse), 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, substep_names: list[str] | None = None) -> None: def __init__(self, parent: ctk.CTkFrame, substep_names: list[str] | None = None) -> None:
super().__init__(parent, fg_color="transparent") super().__init__(parent, fg_color="transparent")
self._collapsed = False self._collapsed = True
self._pulse_job: str | None = None self._pulse_job: str | None = None
self._skipped: set[int] = set()
self._substep_names = substep_names or ["Upload", "Download Verify", "CRC Check", "Reboot", "Boot Verify"] self._substep_names = substep_names or ["Upload", "Download Verify", "CRC Check", "Reboot", "Boot Verify"]
self._num_steps = len(self._substep_names) self._num_steps = len(self._substep_names)
self._toggle_btn = ctk.CTkButton( 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, 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._toggle_btn.grid(row=0, column=0, padx=(4, 4), pady=(2, 2), sticky="w")
@@ -810,12 +891,24 @@ class TftpSubStepFrame(ctk.CTkFrame):
for i, name in enumerate(self._substep_names): for i, name in enumerate(self._substep_names):
dot = ctk.CTkLabel(self, text="", font=FONT_BODY, text_color=ACCENT_PENDING, anchor="w") 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") 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") 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") label.grid(row=i + 1, column=1, padx=(2, 4), pady=(2, 2), sticky="w")
self._sub_items.append({"dot": dot, "label": label}) 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) 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: def _toggle(self) -> None:
self._collapsed = not self._collapsed self._collapsed = not self._collapsed
self._toggle_btn.configure(text="" if self._collapsed else "") self._toggle_btn.configure(text="" if self._collapsed else "")
@@ -823,12 +916,14 @@ class TftpSubStepFrame(ctk.CTkFrame):
if self._collapsed: if self._collapsed:
item["dot"].grid_remove() item["dot"].grid_remove()
item["label"].grid_remove() item["label"].grid_remove()
item["pct"].grid_remove()
else: else:
item["dot"].grid() item["dot"].grid()
item["label"].grid() item["label"].grid()
item["pct"].grid()
def set_expanded(self, expanded: bool) -> None: def set_expanded(self, expanded: bool) -> None:
if expanded != self._collapsed: if expanded == self._collapsed:
self._toggle() self._toggle()
def set_phase(self, phase: str, pct: int = -1) -> None: def set_phase(self, phase: str, pct: int = -1) -> None:
@@ -840,16 +935,22 @@ class TftpSubStepFrame(ctk.CTkFrame):
if idx < 0: if idx < 0:
return return
pct_text = f"{pct}%" if pct >= 0 else ""
for i, item in enumerate(self._sub_items): for i, item in enumerate(self._sub_items):
if i in self._skipped:
continue # leave skipped items untouched
if i < idx: if i < idx:
item["dot"].configure(text="", text_color=_C_DONE) item["dot"].configure(text="", text_color=_C_DONE)
item["label"].configure(text_color=_C_DONE) item["label"].configure(text_color=_C_DONE)
item["pct"].configure(text="OK", text_color=_C_DONE)
elif i == idx: elif i == idx:
item["dot"].configure(text="", text_color=_C_RUNNING) item["dot"].configure(text="", text_color=_C_RUNNING)
item["label"].configure(text_color=_C_RUNNING) item["label"].configure(text_color=_C_RUNNING)
item["pct"].configure(text=pct_text, text_color=_C_RUNNING)
else: else:
item["dot"].configure(text="", text_color=_C_PENDING) item["dot"].configure(text="", text_color=_C_PENDING)
item["label"].configure(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: if phase in self._substep_names and not self._pulse_job:
self._start_pulse(idx) self._start_pulse(idx)
@@ -863,6 +964,7 @@ class TftpSubStepFrame(ctk.CTkFrame):
item = self._sub_items[idx] item = self._sub_items[idx]
item["dot"].configure(text="", text_color=_C_DONE) item["dot"].configure(text="", text_color=_C_DONE)
item["label"].configure(text_color=_C_DONE) item["label"].configure(text_color=_C_DONE)
item["pct"].configure(text="OK", text_color=_C_DONE)
self._stop_pulse() self._stop_pulse()
def set_step_error(self, phase: str) -> None: def set_step_error(self, phase: str) -> None:
@@ -876,6 +978,18 @@ class TftpSubStepFrame(ctk.CTkFrame):
item["label"].configure(text_color=_C_ERROR) item["label"].configure(text_color=_C_ERROR)
self._stop_pulse() self._stop_pulse()
def set_step_skipped(self, phase: str) -> None:
"""Mark a specific step as skipped (warning color, Skip text)."""
idx = self._substep_names.index(phase) if phase in self._substep_names else -1
if idx < 0:
return
self._skipped.add(idx)
item = self._sub_items[idx]
item["dot"].configure(text="", text_color=WARNING_COLOR)
item["label"].configure(text_color=WARNING_COLOR)
item["pct"].configure(text="Skip", text_color=WARNING_COLOR)
self._stop_pulse()
def _start_pulse(self, idx: int) -> None: def _start_pulse(self, idx: int) -> None:
"""Start pulsing the running sub-step.""" """Start pulsing the running sub-step."""
_LIGHT = RUNNING_FADE_LIGHT _LIGHT = RUNNING_FADE_LIGHT
@@ -900,10 +1014,12 @@ class TftpSubStepFrame(ctk.CTkFrame):
def reset(self) -> None: def reset(self) -> None:
self._stop_pulse() self._stop_pulse()
self._skipped.clear()
_C = ACCENT_PENDING _C = ACCENT_PENDING
for i, item in enumerate(self._sub_items): for i, item in enumerate(self._sub_items):
item["dot"].configure(text="", text_color=_C) item["dot"].configure(text="", text_color=_C)
item["label"].configure(text_color=_C) item["label"].configure(text_color=_C)
item["pct"].configure(text="", text_color=_C)
if not self._collapsed: if not self._collapsed:
self._toggle() self._toggle()
self._toggle() # collapse then expand to reset self._toggle() # collapse then expand to reset
@@ -956,7 +1072,7 @@ class UartMonitorWindow(ctk.CTkToplevel):
self._build_log_area() self._build_log_area()
self._build_status_bar() self._build_status_bar()
# Try to open serial port after window is drawn # Try to open serial port after window is drawn
self.after(100, self._try_open) self.after(100, self._try_open)
# ── Title Bar ────────────────────────────────────────────── # ── Title Bar ──────────────────────────────────────────────
@@ -1068,7 +1184,7 @@ class UartMonitorWindow(ctk.CTkToplevel):
ser.close() ser.close()
self.after(0, self._on_open_success) self.after(0, self._on_open_success)
except Exception as e: 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() threading.Thread(target=_bg_open, daemon=True).start()
@@ -1179,6 +1295,12 @@ class UartMonitorWindow(ctk.CTkToplevel):
"""True if the UART monitor thread is currently running.""" """True if the UART monitor thread is currently running."""
return self._monitor is not None and self._monitor.is_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: def _on_close(self) -> None:
"""Cleanup UartMonitor and destroy window.""" """Cleanup UartMonitor and destroy window."""
if self._monitor: if self._monitor:
+270 -83
View File
@@ -1,14 +1,15 @@
"""Reboot manager for Zynq Flasher GUI. """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 from __future__ import annotations
import importlib import os
import platform import socket
import shutil
import subprocess import subprocess
import tempfile
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from typing import Callable from typing import Callable
@@ -26,85 +27,75 @@ class RebootResult:
output: str = "" output: str = ""
@dataclass
class UdpVersionResult:
"""Result of a UDP version query."""
success: bool
message: str
version: str = ""
ProgressCallback = Callable[[str, str], None] | None 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: Args:
config: Application configuration. ip: Zynq IP address string.
timeout: Seconds to wait for reboot acknowledgment.
callback: Optional progress callback.
Returns: 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: try:
result = subprocess.run( octets = ip.strip().split(".")
cmd, last = int(octets[3])
capture_output=True, return last - 3
text=True, except (ValueError, IndexError):
timeout=10, return -1
)
if callback:
if result.returncode == 0:
callback("progress", f"Reboot command sent to {ip}")
else:
callback("error", f"SSH reboot failed: {result.stderr.strip()}")
# Wait for device to come back
if result.returncode == 0 and callback:
callback("progress", f"Waiting {timeout}s for device to reboot...")
return RebootResult( def _send_udp_command(
success=result.returncode == 0, ip: str,
message="Reboot command sent" if result.returncode == 0 else f"SSH reboot failed: {result.stderr.strip()}", command: bytes,
method="ssh", timeout: float = 5.0,
output=result.stdout, buffer_size: int = 4096,
) ) -> tuple[bool, str]:
except subprocess.TimeoutExpired: """Send a UDP command to the Zynq and receive the response.
return RebootResult(
success=False, Args:
message="SSH reboot timed out", ip: Zynq IP address.
method="ssh", command: Command bytes to send (e.g., b'reboot()').
) timeout: Receive timeout in seconds.
except FileNotFoundError: buffer_size: Max receive buffer size.
return RebootResult(
success=False, Returns:
message="ssh command not found", (success, response_text).
method="ssh", """
) 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: except OSError as e:
return RebootResult( return False, f"UDP error: {e}"
success=False, finally:
message=f"OS error: {e}", sock.close()
method="ssh",
)
# ── Reboot methods ───────────────────────────────────────────────────
def reboot_via_serial( def reboot_via_serial(
@@ -113,7 +104,7 @@ def reboot_via_serial(
) -> RebootResult: ) -> RebootResult:
"""Reboot the Zynq device via serial port. """Reboot the Zynq device via serial port.
Sends a reboot command through the serial console. Sends reboot() through the serial console.
Args: Args:
config: Application configuration. config: Application configuration.
@@ -144,16 +135,16 @@ def reboot_via_serial(
# Send Ctrl+C to break out of any running process # Send Ctrl+C to break out of any running process
ser.write(b"\x03") ser.write(b"\x03")
time.sleep(0.5) time.sleep(0.5)
# Send reboot command # Send reboot command as function call
ser.write(b"reboot\r\n") ser.write(b"reboot()\r\n")
ser.reset_input_buffer() ser.reset_input_buffer()
if callback: if callback:
callback("progress", "Reboot command sent via serial") callback("progress", "reboot() sent via serial")
return RebootResult( return RebootResult(
success=True, success=True,
message="Reboot command sent via serial", message="reboot() sent via serial",
method="serial", method="serial",
) )
except ImportError: 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( def reboot_zynq(
config: Config, config: Config,
method: str = "auto", method: str = "auto",
@@ -184,12 +363,12 @@ def reboot_zynq(
) -> RebootResult: ) -> RebootResult:
"""Reboot the Zynq device using the best available method. """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: Args:
config: Application configuration. config: Application configuration.
method: Reboot method: 'auto', 'ssh', or 'serial'. method: Reboot method: 'auto', 'jtag', 'udp', or 'serial'.
timeout: Seconds to wait for reboot. timeout: Seconds to wait for reboot (ssh-only, kept for compat).
callback: Optional progress callback. callback: Optional progress callback.
Returns: Returns:
@@ -197,17 +376,25 @@ def reboot_zynq(
""" """
timeout = timeout or config.reboot_timeout timeout = timeout or config.reboot_timeout
if method == "ssh": if method == "jtag":
return reboot_via_ssh(config, timeout, callback) return reboot_via_jtag(config, callback)
elif method == "udp":
return reboot_via_udp(config, callback)
elif method == "serial": elif method == "serial":
return reboot_via_serial(config, callback) return reboot_via_serial(config, callback)
# Auto: try SSH first, then serial # Auto: try JTAG first, then UDP, then serial
result = reboot_via_ssh(config, timeout, callback) result = reboot_via_jtag(config, callback)
if result.success: if result.success:
return result return result
if callback: 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) return reboot_via_serial(config, callback)
+23 -11
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import re import re
import threading import threading
import time
from dataclasses import dataclass from dataclasses import dataclass
from typing import Callable from typing import Callable
@@ -136,6 +137,7 @@ def check_uart_available(
port: str, port: str,
baudrate: int = 115200, baudrate: int = 115200,
timeout: float = 3.0, timeout: float = 3.0,
monitor: UartMonitor | None = None,
) -> tuple[bool, str]: ) -> tuple[bool, str]:
"""Check whether the UART serial port is available and readable. """Check whether the UART serial port is available and readable.
@@ -144,19 +146,24 @@ def check_uart_available(
considered available (the device may simply not be outputting yet). considered available (the device may simply not be outputting yet).
If the port cannot be opened at all, it is unavailable. If the port cannot be opened at all, it is unavailable.
If *monitor* is provided and currently running, the function
skips the port-open test (which would fail on Windows with
Error 13 due to exclusive access) and returns True immediately.
Args: Args:
port: Serial port device path (e.g., '/dev/ttyUSB0'). port: Serial port device path (e.g., '/dev/ttyUSB0').
baudrate: Baud rate for serial communication. baudrate: Baud rate for serial communication.
timeout: Seconds to wait for data after opening. timeout: Seconds to wait for data after opening.
monitor: Optional running UartMonitor — if active, reuse it.
Returns: Returns:
Tuple of (is_available, reason_string). Tuple of (is_available, reason_string).
- (True, "Port open") — port is usable
- (True, "Port open, no data yet") — port open, no data currently
- (False, "No serial ports detected") — system has no serial ports
- (False, "Port not in detected ports") — configured port missing
- (False, "Failed to open: <error>") — port open error
""" """
# Reuse existing monitor when available (avoids Error 13 on Windows)
if monitor is not None and monitor.is_running():
return True, "Already monitoring"
# 1. Quick port list check
# 1. Quick port list check # 1. Quick port list check
ports = detect_serial_ports() ports = detect_serial_ports()
if not ports: if not ports:
@@ -222,28 +229,33 @@ def test_serial_version(
port: str, port: str,
baudrate: int = 115200, baudrate: int = 115200,
timeout: float = 5.0, timeout: float = 5.0,
monitor: UartMonitor | None = None,
) -> tuple[bool, str, str]: ) -> tuple[bool, str, str]:
"""Test serial port by sending 'ver()' command and parsing response. """Test serial port by sending 'ver()' command and parsing response.
Sends 'ver\\r\\n' to the serial port and reads the response. If *monitor* is running, skips opening a new connection and
Attempts to parse version string from the response. returns the latest parsed version from the monitor instead.
Args: Args:
port: Serial port device path (e.g., '/dev/ttyUSB0'). port: Serial port device path (e.g., '/dev/ttyUSB0').
baudrate: Baud rate for serial communication. baudrate: Baud rate for serial communication.
timeout: Read timeout in seconds. timeout: Read timeout in seconds.
monitor: Optional running UartMonitor — if active, reuse it.
Returns: Returns:
Tuple of (success, message, version_string). Tuple of (success, message, version_string).
- (True, "Version: x.y.z", "x.y.z") — version detected
- (True, "Port responded", "") — port responded but no version
- (False, "Error message", "") — error occurred
""" """
import serial import serial
# Reuse existing monitor when available (avoids Error 13 on Windows)
if monitor is not None and monitor.is_running():
info = monitor.latest_info
if info and info.version:
return True, f"Version: {info.version}", info.version
return True, "Port monitored, no version parsed yet", ""
try: try:
with serial.Serial(port, baudrate, timeout=timeout) as ser: with serial.Serial(port, baudrate, timeout=timeout) as ser:
# Clear buffers
ser.reset_input_buffer() ser.reset_input_buffer()
ser.reset_output_buffer() ser.reset_output_buffer()
+44 -22
View File
@@ -9,7 +9,6 @@ from __future__ import annotations
import binascii import binascii
import socket import socket
import struct import struct
import tempfile
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Callable 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, 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. """Upload a file to a TFTP server using pure Python.
Args: Args:
@@ -63,6 +63,7 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
file_path: Local file path. file_path: Local file path.
remote_name: Remote filename. remote_name: Remote filename.
timeout: Socket timeout in seconds. timeout: Socket timeout in seconds.
callback: Optional progress callback (status, message).
Returns: Returns:
(success, message). (success, message).
@@ -112,6 +113,8 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
pct = offset * 100 // total pct = offset * 100 // total
print(f"[TFTP] Block {block}/{-(-total // block_size)} ({pct}%) sent to {srv_ip}:{srv_port}", print(f"[TFTP] Block {block}/{-(-total // block_size)} ({pct}%) sent to {srv_ip}:{srv_port}",
file=sys.stderr) file=sys.stderr)
if callback:
callback("progress", f"{pct}%")
# Wait for ACK # Wait for ACK
try: try:
@@ -130,7 +133,9 @@ def _tftp_put(host: str, file_path: str, remote_name: str,
def _tftp_get(host: str, remote_name: str, local_dir: str | Path, def _tftp_get(host: str, remote_name: str, local_dir: str | Path,
timeout: float = 30.0) -> tuple[bool, 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. """Download a file from a TFTP server using pure Python.
Args: Args:
@@ -138,6 +143,8 @@ def _tftp_get(host: str, remote_name: str, local_dir: str | Path,
remote_name: Remote filename. remote_name: Remote filename.
local_dir: Local directory to save the downloaded file. local_dir: Local directory to save the downloaded file.
timeout: Socket timeout in seconds. timeout: Socket timeout in seconds.
callback: Optional progress callback (status, message).
expected_size: Expected file size in bytes (for progress percentage).
Returns: Returns:
(success, message, local_path). (success, message, local_path).
@@ -170,46 +177,56 @@ def _tftp_get(host: str, remote_name: str, local_dir: str | Path,
srv_ip, srv_port = server_addr srv_ip, srv_port = server_addr
block = struct.unpack("!H", data_pkt[2:4])[0] block = struct.unpack("!H", data_pkt[2:4])[0]
file_data = data_pkt[4:] chunk = data_pkt[4:]
import sys import sys
print(f"[TFTP] DATA block 1 received from {srv_ip}:{srv_port} " print(f"[TFTP] DATA block {block} received from {srv_ip}:{srv_port} "
f"({len(file_data)} bytes)", file=sys.stderr) f"({len(chunk)} bytes)", file=sys.stderr)
# ── ACK block 0 and receive subsequent blocks ── # ── ACK this block with the actual block number ──
ack = struct.pack("!HH", _TFTP_ACK, 0) file_data = bytearray(chunk)
ack = struct.pack("!HH", _TFTP_ACK, block)
sock.sendto(ack, server_addr) sock.sendto(ack, server_addr)
total = len(file_data) if callback:
while len(file_data) == _TFTP_BLOCK_SIZE: pct = len(file_data) * 100 // expected_size if expected_size > 0 else -1
# Wait for next DATA block callback("progress", f"{pct}%")
# ── Receive subsequent blocks while last chunk was full-size ──
while len(chunk) == _TFTP_BLOCK_SIZE:
try: try:
data_pkt, _ = sock.recvfrom(516) data_pkt, _ = sock.recvfrom(516)
if len(data_pkt) < 4: if len(data_pkt) < 4:
return False, "Incomplete DATA packet", local_path return False, "Incomplete DATA packet", local_path
opcode = struct.unpack("!H", data_pkt[:2])[0] 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: if opcode != _TFTP_DATA:
return False, f"Expected DATA, got opcode {opcode}", local_path return False, f"Expected DATA, got opcode {opcode}", local_path
new_block = struct.unpack("!H", data_pkt[2:4])[0] new_block = struct.unpack("!H", data_pkt[2:4])[0]
chunk = data_pkt[4:] chunk = data_pkt[4:]
file_data += chunk file_data += chunk
total = len(file_data)
block = new_block block = new_block
if block % 100 == 0: if block % 100 == 0:
pct = total * 100 // max(total, 1) print(f"[TFTP] Block {block} ({len(file_data)} bytes) "
print(f"[TFTP] Block {block}/{block} ({pct}%) received from {srv_ip}:{srv_port}", f"received from {srv_ip}:{srv_port}", file=sys.stderr)
file=sys.stderr)
# ACK this block # ACK this block
ack = struct.pack("!HH", _TFTP_ACK, block) ack = struct.pack("!HH", _TFTP_ACK, block)
sock.sendto(ack, server_addr) 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: except socket.timeout:
return False, f"Timeout waiting for DATA block {block}", local_path return False, f"Timeout waiting for DATA block {block + 1}", local_path
# ── Write file ── # ── Write file ──
local_path.write_bytes(file_data) total = len(file_data)
return True, f"Downloaded {total} bytes to {local_path}", local_path local_path.write_bytes(bytes(file_data))
return True, f"Downloaded {total} bytes in {block} blocks to {local_path}", local_path
finally: finally:
sock.close() sock.close()
@@ -254,7 +271,8 @@ def tftp_upload(
) )
try: 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: except OSError as e:
return TftpResult( return TftpResult(
step="upload", success=False, step="upload", success=False,
@@ -279,20 +297,23 @@ def tftp_download(
remote_name: str, remote_name: str,
local_dir: str | Path | None = None, local_dir: str | Path | None = None,
callback: ProgressCallback = None, callback: ProgressCallback = None,
expected_size: int = 0,
) -> TftpResult: ) -> TftpResult:
"""Download a file from the Zynq device via TFTP (pure Python). """Download a file from the Zynq device via TFTP (pure Python).
Args: Args:
config: Application configuration. config: Application configuration.
remote_name: Remote filename on the TFTP server. remote_name: Remote filename on the TFTP server.
local_dir: Local directory to save the file (default: temp directory). local_dir: Local directory to save the file (default: project .tmp/tftp/).
callback: Optional progress callback. callback: Optional progress callback.
expected_size: Expected file size in bytes (for progress percentage).
Returns: Returns:
TftpResult with download status and CRC. TftpResult with download status and CRC.
""" """
if local_dir is None: if local_dir is None:
local_dir = tempfile.gettempdir() # Use project-local temp dir for cross-platform compatibility
local_dir = Path(__file__).resolve().parent.parent / ".tmp" / "tftp"
local_dir = Path(local_dir) local_dir = Path(local_dir)
if callback: if callback:
@@ -307,7 +328,8 @@ def tftp_download(
) )
try: try:
ok, msg, local_path = _tftp_get(config.zynq_ip, remote_name, local_dir) ok, msg, local_path = _tftp_get(config.zynq_ip, remote_name, local_dir,
callback=callback, expected_size=expected_size)
except OSError as e: except OSError as e:
return TftpResult( return TftpResult(
step="download", success=False, step="download", success=False,
+311 -70
View File
@@ -5,6 +5,12 @@ Checks for required Xilinx tools:
- program_flash — QSPI flash programming - program_flash — QSPI flash programming
- bootgen — BOOT.BIN generation (image creation only, not hardware) - 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. Provides shared helpers used by flash_programmer and bitstream_programmer.
""" """
@@ -12,6 +18,7 @@ from __future__ import annotations
import os import os
import platform import platform
import re
import shutil import shutil
import subprocess import subprocess
from dataclasses import dataclass from dataclasses import dataclass
@@ -49,8 +56,104 @@ TOOL_ALIASES: dict[str, list[str]] = {
"xsct": ["xsct"], "xsct": ["xsct"],
"program_flash": ["program_flash"], "program_flash": ["program_flash"],
"bootgen": ["bootgen"], "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( def _find_executable(
exec_name: str, exec_name: str,
@@ -61,56 +164,55 @@ def _find_executable(
"""Find an executable by name using PATH, Vitis, and Xilinx root. """Find an executable by name using PATH, Vitis, and Xilinx root.
Search order: Search order:
1. System PATH (shutil.which) 1. System PATH (shutil.which) — tries bare name, .exe, .bat on Windows
2. vitis_path/bin (legacy config) 2. vitis_path/bin (legacy config)
3. xilinx_root/Vitis/*/bin and Vivado/*/bin (multi-version scan) 3. xilinx_root/Vitis/*/bin, Vivado/*/bin, SDK/*/bin — pick NEWEST version
4. Common install paths 4. Common install paths (cross-platform)
Args: Args:
exec_name: Executable name (without extension). exec_name: Executable name (without extension).
vitis_path: Legacy Vitis installation directory. vitis_path: Legacy Vitis installation directory.
system: Platform identifier (linux, windows, darwin). 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: Returns:
Absolute path to the executable, or None. Absolute path to the best executable, or None.
""" """
exec_suffix = ".exe" if system == "windows" else "" exec_suffix = ".exe" if system == "windows" else ""
# 1. System PATH # 1. System PATH — search without explicit extension first
found = shutil.which(f"{exec_name}{exec_suffix}") # 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: if found:
return found return found
# 2. Legacy vitis_path/bin # 2. Legacy vitis_path/bin
if vitis_path: if vitis_path:
candidate = Path(vitis_path) / "bin" / f"{exec_name}{exec_suffix}" for sfx in (".bat", ".exe") if system == "windows" else (exec_suffix,):
if candidate.is_file(): candidate = Path(vitis_path) / "bin" / f"{exec_name}{sfx}"
return str(candidate) 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: if xilinx_root:
xr = Path(xilinx_root) candidates = _scan_xilinx_root(xilinx_root, exec_name, exec_suffix)
for product in ("Vitis", "Vivado"): if candidates:
pdir = xr / product # Preference: Vitis over Vivado if same version
if not pdir.is_dir(): # candidates are already sorted by version desc;
continue # within same version, Vitis appears first (scanned first)
# Scan version directories, newest first return str(candidates[0][1])
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)
# 4. Common install paths (cross-platform) # 4. Common install paths (cross-platform fallback)
if system == "windows": if system == "windows":
common = [ common: list[str] = []
f"C:/Xilinx/Vitis/2023.2/bin/{exec_name}.exe", for sfx in (".bat", ".exe"):
f"C:/Xilinx/Vivado/2023.2/bin/{exec_name}.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: else:
common = [ common = [
f"/data/xilinx/Vitis/2023.2/bin/{exec_name}", f"/data/xilinx/Vitis/2023.2/bin/{exec_name}",
@@ -126,78 +228,206 @@ def _find_executable(
return None return None
def get_xsct_path(vitis_path: str = "", xilinx_root: str = "") -> str | None: def get_all_found_versions(
"""Find the xsct (Xilinx Software Command Line Tool) executable. 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: Args:
vitis_path: Legacy Vitis install path.
xilinx_root: Xilinx root directory. xilinx_root: Xilinx root directory.
exec_name: Executable name without extension.
Returns: 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) return _find_executable("xsct", vitis_path, platform.system().lower(), xilinx_root)
def get_program_flash_path(vitis_path: str = "", xilinx_root: str = "") -> str | None: def get_program_flash_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
"""Find the program_flash executable. """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.
"""
return _find_executable("program_flash", vitis_path, platform.system().lower(), xilinx_root) return _find_executable("program_flash", vitis_path, platform.system().lower(), xilinx_root)
def get_bootgen_path(vitis_path: str = "", xilinx_root: str = "") -> str | None: def get_bootgen_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
"""Find the bootgen executable. """Find the bootgen executable."""
Args:
vitis_path: Legacy Vitis install path.
xilinx_root: Xilinx root directory.
Returns:
Absolute path to bootgen, or None.
"""
return _find_executable("bootgen", vitis_path, platform.system().lower(), xilinx_root) return _find_executable("bootgen", vitis_path, platform.system().lower(), xilinx_root)
def _get_tool_version(tool_path: str, tool_name: str) -> str: def get_vivado_path(vitis_path: str = "", xilinx_root: str = "") -> str | None:
"""Get the version/h banner from a tool executable. """Find the Vivado executable (JTAG operations via hw_server).
Args: Args:
tool_path: Absolute path to the tool. vitis_path: Legacy Vitis installation directory.
tool_name: Canonical tool name. xilinx_root: Xilinx root directory.
Returns: 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: try:
result = subprocess.run( result = subprocess.run(
[tool_path, "-help"], [tool_path, "-version"],
capture_output=True, capture_output=True,
text=True, text=True,
timeout=10, timeout=10,
) )
output = (result.stdout + result.stderr).strip() output = (result.stdout + result.stderr).strip()
if output: 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): except (subprocess.TimeoutExpired, OSError):
pass 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: def check_vitis(config: Config) -> VitisCheckResult:
"""Check availability of all required Xilinx tools. """Check availability of all required Xilinx tools.
Scans all versions under xilinx_root and picks the newest.
Args: Args:
config: Application configuration. config: Application configuration.
Returns: Returns:
VitisCheckResult with per-tool status. VitisCheckResult with per-tool status and version info.
""" """
system = platform.system().lower() system = platform.system().lower()
vitis_path = config.vitis_path if hasattr(config, 'vitis_path') else "" vitis_path = config.vitis_path if hasattr(config, 'vitis_path') else ""
@@ -227,18 +457,26 @@ def _check_tool(
system: str, system: str,
xilinx_root: str = "", xilinx_root: str = "",
) -> ToolStatus: ) -> 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: for alias in aliases:
found = _find_executable(alias, vitis_path, system, xilinx_root) found = _find_executable(alias, vitis_path, system, xilinx_root)
if found: if found:
version = _get_tool_version(found, tool_name) best_path = found
return ToolStatus( best_alias = alias if alias != tool_name else ""
name=tool_name, break
found=True,
path=found, if best_path:
version=version, version = _get_tool_version(best_path)
alias=alias, return ToolStatus(
) name=tool_name,
found=True,
path=best_path,
version=version,
alias=best_alias,
)
return ToolStatus( return ToolStatus(
name=tool_name, name=tool_name,
@@ -251,6 +489,9 @@ def _check_tool(
) )
# ── Status serialization ─────────────────────────────────────────────
def get_tool_status_dict(result: VitisCheckResult) -> dict[str, Any]: def get_tool_status_dict(result: VitisCheckResult) -> dict[str, Any]:
"""Convert a VitisCheckResult to a serializable dictionary. """Convert a VitisCheckResult to a serializable dictionary.
+240 -260
View File
@@ -1,78 +1,85 @@
"""Zynq JTAG presence checker for Zynq Flasher GUI. """Zynq JTAG presence checker for Zynq Flasher GUI.
Connects to hw_server via Vivado and scans the JTAG chain to detect Starts hw_server explicitly, then uses Vivado batch-mode TCL to scan
Zynq devices, returning chip model and presence information. the JTAG chain. On failure, Vivado's stdout/stderr is returned in the
Checks for both PS (Processing System) and PL (Programmable Logic) error message for debugging.
recognition on the JTAG chain.
""" """
from __future__ import annotations from __future__ import annotations
import os
import re import re
import shutil
import subprocess import subprocess
import tempfile
import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from typing import Callable from typing import Callable
from config_manager import Config from config_manager import Config
from vitis_checker import get_vivado_path, get_hw_server_path, build_tool_command
@dataclass @dataclass
class JtagDevice: class JtagDevice:
"""Information about a device on the JTAG chain.""" """Information about a device on the JTAG chain."""
name: str # Device name from JTAG chain (e.g. "xc7z100_1") name: str
is_zynq: bool # Whether this is a Zynq device idcode: str = ""
is_arm_dap: bool # Whether this is an ARM DAP (PS side) is_zynq: bool = False
is_pl_device: bool # Whether this is a PL device is_arm_dap: bool = False
is_pl_device: bool = False
@dataclass @dataclass
class ZynqDevice: class ZynqDevice:
"""Information about a detected Zynq device on the JTAG chain.""" """Information about a detected Zynq device on the JTAG chain."""
name: str # Device name from JTAG chain (e.g. "xc7z100_1") name: str
part_number: str # Extracted part number (e.g. "XC7Z100") part_number: str
family: str # Family (e.g. "zynq7") family: str
speed: str # Speed grade (e.g. "1", "2", "-1", "-2") speed: str
idcode: str = "" # JTAG IDCODE if available idcode: str = ""
is_programmable: bool = False # Whether the device is programmable is_programmable: bool = False
@dataclass @dataclass
class ZynqCheckResult: class ZynqCheckResult:
"""Result of Zynq device detection on the JTAG chain.""" """Result of Zynq device detection on the JTAG chain."""
devices: list[ZynqDevice] # All detected Zynq devices devices: list[ZynqDevice]
jtag_chain: list[JtagDevice] # All devices on JTAG chain jtag_chain: list[JtagDevice]
is_present: bool # Whether any Zynq device was found is_present: bool
ps_detected: bool # Whether PS (ARM DAP) is detected ps_detected: bool
pl_detected: bool # Whether PL is detected pl_detected: bool
hw_server_url: str # hw_server connection URL used hw_server_url: str
error: str = "" # Error message if detection failed error: str = ""
hw_server_pid: int | None = None
# Zynq part number patterns from JTAG device names # ── Patterns ──────────────────────────────────────────────────────────
# Examples: xc7z010_1, xc7z020_1, xc7z100_1, xc7z030_1 _DEVICE_LINE_RE = re.compile(r"^DEVICE:(\S+)\s+IDCODE:([01]+)")
ZYNQ_PART_PATTERN = re.compile( _ZYNQ_RE = re.compile(r"^xc\d+z\d+[a-z]?\d+", re.IGNORECASE)
r"^(xc\d+z\d+[a-z]?\d+)_\d+$", re.IGNORECASE _ARM_DAP_RE = re.compile(r"^arm_dap", re.IGNORECASE)
)
# ARM DAP pattern (PS side)
ARM_DAP_PATTERN = re.compile(r"^arm_dap_\d+$", re.IGNORECASE)
# hw_server default port
HW_SERVER_PORT = 3121 HW_SERVER_PORT = 3121
def check_zynq_jtag(config: Config, callback: Callable | None = None) -> ZynqCheckResult: def _idcode_to_hex(binary_str: str) -> str:
"""Check for Zynq devices on the JTAG chain via hw_server. 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 # ── Main entry ────────────────────────────────────────────────────────
for PS (ARM DAP) and PL (Programmable Logic) recognition.
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: Args:
config: Application configuration. config: Application configuration.
@@ -82,178 +89,227 @@ def check_zynq_jtag(config: Config, callback: Callable | None = None) -> ZynqChe
ZynqCheckResult with detected devices and status. ZynqCheckResult with detected devices and status.
""" """
hw_server_url = f"localhost:{HW_SERVER_PORT}" hw_server_url = f"localhost:{HW_SERVER_PORT}"
xilinx_root = getattr(config, "xilinx_path", "")
if callback: if callback:
callback("start", "Connecting to hw_server...") callback("start", "Connecting to hw_server...")
# Find vivado executable # ── 1. Find tools ──────────────────────────────────────────
vivado_path = _get_vivado_path( vivado_path = get_vivado_path(xilinx_root=xilinx_root)
config.vitis_path if hasattr(config, 'vitis_path') else ""
)
if not vivado_path: if not vivado_path:
return ZynqCheckResult( return _fail("Vivado not found", hw_server_url)
devices=[],
jtag_chain=[],
is_present=False,
ps_detected=False,
pl_detected=False,
hw_server_url=hw_server_url,
error="Vivado not found",
)
# Create temporary TCL script hw_server_path = get_hw_server_path(xilinx_root=xilinx_root)
tcl_content = _build_jtag_query_tcl() if not hw_server_path:
tcl_path = Path("/tmp/zynq_jtag_check_{}.tcl".format(os.getpid())) return _fail("hw_server not found", hw_server_url)
# ── 2. Start hw_server ─────────────────────────────────────
hw_proc: subprocess.Popen | None = None
hw_pid: int | None = None
try: try:
tcl_path.write_text(tcl_content) hw_proc = subprocess.Popen(
build_tool_command(hw_server_path),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
hw_pid = hw_proc.pid
# 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, pid=hw_pid)
except OSError as e:
return _fail(f"Failed to start hw_server: {e}", hw_server_url)
if callback: if callback:
callback("progress", "Scanning JTAG chain...") callback("progress", "Scanning JTAG chain...")
# Run Vivado in batch mode # ── 3. Run Vivado TCL ──────────────────────────────────────
cmd = [vivado_path, "-mode", "batch", "-source", str(tcl_path)] 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( result = subprocess.run(
cmd, cmd,
capture_output=True, capture_output=True,
text=True, text=True,
timeout=30, timeout=python_timeout,
) )
output = result.stdout + result.stderr output = result.stdout + result.stderr
for line in output.splitlines():
# Parse JTAG device information from output stripped = line.strip()
jtag_chain = _parse_jtag_chain(output) if stripped:
devices = _filter_zynq_devices(jtag_chain) output_lines.append(stripped)
if callback:
if callback: callback("progress", stripped)
if devices:
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")
# 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,
hw_server_url=hw_server_url,
error="" if devices else "No Zynq device found on JTAG chain",
)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
return ZynqCheckResult( return _fail(
devices=[], f"JTAG scan timed out after {python_timeout}s",
jtag_chain=[], hw_server_url,
is_present=False, pid=hw_pid,
ps_detected=False,
pl_detected=False,
hw_server_url=hw_server_url,
error="JTAG scan timed out after 30s",
) )
except OSError as e: except OSError as e:
return ZynqCheckResult( return _fail(f"Failed to run Vivado: {e}", hw_server_url, pid=hw_pid)
devices=[],
jtag_chain=[],
is_present=False,
ps_detected=False,
pl_detected=False,
hw_server_url=hw_server_url,
error=f"OS error: {e}",
)
finally: finally:
tcl_path.unlink(missing_ok=True) 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
output = "\n".join(output_lines)
# Parse results
jtag_chain = _parse_jtag_chain(output)
devices = _filter_zynq_devices(jtag_chain)
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:
# 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,
hw_server_pid=hw_pid,
)
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=True,
ps_detected=ps_detected, pl_detected=pl_detected,
hw_server_url=hw_server_url,
error="",
hw_server_pid=hw_pid,
)
# ── Helpers ───────────────────────────────────────────────────────────
def _fail(error: str, url: str, pid: int | None = None) -> ZynqCheckResult:
return ZynqCheckResult(
devices=[], jtag_chain=[], is_present=False,
ps_detected=False, pl_detected=False,
hw_server_url=url, error=error,
hw_server_pid=pid,
)
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: def _build_jtag_query_tcl() -> str:
"""Build TCL script for JTAG device query. """Build Vivado TCL script file for JTAG device enumeration."""
content = """\
Returns: open_hw
TCL script content as string.
"""
return """
open_hw_manager
connect_hw_server connect_hw_server
open_hw_target open_hw_target
puts "===JTAG_START===" puts "===JTAG_START==="
foreach dev [get_hw_devices] { foreach dev [get_hw_devices] {
set name [get_property NAME $dev] set name [get_property NAME $dev]
puts "DEVICE:$name" set idcode [get_property IDCODE $dev]
puts "DEVICE:$name IDCODE:$idcode"
} }
puts "===JTAG_END===" puts "===JTAG_END==="
close_hw
disconnect_hw_server
quit 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]: def _parse_jtag_chain(output: str) -> list[JtagDevice]:
"""Parse JTAG chain information from Vivado output. """Parse Vivado TCL output into JtagDevice list."""
Args:
output: Combined stdout/stderr from Vivado batch execution.
Returns:
List of JtagDevice objects for all devices on the JTAG chain.
"""
devices: list[JtagDevice] = [] devices: list[JtagDevice] = []
section = _extract_section(output, "===JTAG_START===", "===JTAG_END===")
# Extract device list from output if not section:
devices_section = _extract_section(output, "===JTAG_START===", "===JTAG_END===")
if not devices_section:
return devices return devices
for line in section.splitlines():
for line in devices_section.splitlines():
line = line.strip() line = line.strip()
if line.startswith("DEVICE:"): if not line:
device_name = line[len("DEVICE:"):] continue
device = _classify_jtag_device(device_name) match = _DEVICE_LINE_RE.match(line)
devices.append(device) if match:
name = match.group(1)
idcode_hex = _idcode_to_hex(match.group(2))
devices.append(_classify_jtag_device(name, idcode_hex))
return devices return devices
def _filter_zynq_devices(jtag_chain: list[JtagDevice]) -> list[ZynqDevice]: 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] = [] zynq_devices: list[ZynqDevice] = []
for jtag_dev in jtag_chain: for jtag_dev in jtag_chain:
if jtag_dev.is_zynq: 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: if zynq_dev:
zynq_devices.append(zynq_dev) zynq_devices.append(zynq_dev)
return zynq_devices return zynq_devices
def _extract_section( def _extract_section(text: str, start_marker: str, end_marker: str) -> str | None:
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
start_idx = text.rfind(start_marker) start_idx = text.rfind(start_marker)
end_idx = text.rfind(end_marker) end_idx = text.rfind(end_marker)
if start_idx == -1 or end_idx == -1 or start_idx >= end_idx: if start_idx == -1 or end_idx == -1 or start_idx >= end_idx:
@@ -261,105 +317,37 @@ def _extract_section(
return text[start_idx + len(start_marker):end_idx].strip() return text[start_idx + len(start_marker):end_idx].strip()
def _classify_jtag_device(device_name: str) -> JtagDevice: def _classify_jtag_device(device_name: str, idcode: str = "") -> JtagDevice:
"""Classify a JTAG device by its name. is_zynq = bool(_ZYNQ_RE.match(device_name))
is_arm_dap = bool(_ARM_DAP_RE.match(device_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
return JtagDevice( return JtagDevice(
name=device_name, name=device_name,
idcode=idcode,
is_zynq=is_zynq, is_zynq=is_zynq,
is_arm_dap=is_arm_dap, 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: def _parse_zynq_device(device_name: str, idcode: str = "") -> ZynqDevice | None:
"""Parse a Zynq device from its JTAG chain name. match = _ZYNQ_RE.match(device_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)
if not match: if not match:
return None return None
base = device_name.rsplit("_", 1)[0] if "_" in device_name else device_name
part_base = match.group(1) part_number = base.upper()
# 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
family = "zynq7" family = "zynq7"
speed = "1" # Default speed grade speed = "-1"
speed_match = re.search(r"(\d+)$", part_number)
# Parse speed grade from the numeric suffix
speed_match = re.search(r"(\d+)$", part_base)
if speed_match: if speed_match:
speed_num = int(speed_match.group(1)) speed = "-2" if int(speed_match.group(1)) >= 2 else "-1"
# Map numeric speed to speed grade string
if speed_num >= 2:
speed = "-2"
elif speed_num >= 1:
speed = "-1"
return ZynqDevice( return ZynqDevice(
name=device_name, name=device_name, part_number=part_number,
part_number=part_number, family=family, speed=speed,
family=family, idcode=idcode, is_programmable=True,
speed=speed,
) )
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: 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 { return {
"is_present": result.is_present, "is_present": result.is_present,
"ps_detected": result.ps_detected, "ps_detected": result.ps_detected,
@@ -367,23 +355,15 @@ def get_zynq_status_dict(result: ZynqCheckResult) -> dict:
"hw_server_url": result.hw_server_url, "hw_server_url": result.hw_server_url,
"error": result.error, "error": result.error,
"jtag_chain": [ "jtag_chain": [
{ {"name": d.name, "idcode": d.idcode,
"name": dev.name, "is_zynq": d.is_zynq, "is_arm_dap": d.is_arm_dap,
"is_zynq": dev.is_zynq, "is_pl_device": d.is_pl_device}
"is_arm_dap": dev.is_arm_dap, for d in result.jtag_chain
"is_pl_device": dev.is_pl_device,
}
for dev in result.jtag_chain
], ],
"devices": [ "devices": [
{ {"name": d.name, "part_number": d.part_number,
"name": dev.name, "family": d.family, "speed": d.speed,
"part_number": dev.part_number, "idcode": d.idcode, "is_programmable": d.is_programmable}
"family": dev.family, for d in result.devices
"speed": dev.speed,
"idcode": dev.idcode,
"is_programmable": dev.is_programmable,
}
for dev in result.devices
], ],
} }
-14
View File
@@ -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
-14
View File
@@ -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.