diff --git a/src/config_manager.py b/src/config_manager.py index 018e87a..2d35de8 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -35,7 +35,7 @@ DEFAULT_CONFIG: dict[str, Any] = { "inter_step_delay": 2, "boot_wait_delay": 10, "step_timeouts": { - "check_env": 30, + "check_env": 120, "flash_erase": 120, "flash_program": 600, "bitstream": 60, diff --git a/src/vitis_checker.py b/src/vitis_checker.py index 67f3d60..73718b0 100644 --- a/src/vitis_checker.py +++ b/src/vitis_checker.py @@ -56,6 +56,7 @@ TOOL_ALIASES: dict[str, list[str]] = { "xsct": ["xsct"], "program_flash": ["program_flash"], "bootgen": ["bootgen"], + "hw_server": ["hw_server"], } # Product subdirectories under xilinx_root to scan (in preference order) @@ -330,6 +331,24 @@ def get_vivado_path(vitis_path: str = "", xilinx_root: str = "") -> str | 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 ─────────────────────────────────────────── diff --git a/src/zynq_checker.py b/src/zynq_checker.py index 7f21130..d19173e 100644 --- a/src/zynq_checker.py +++ b/src/zynq_checker.py @@ -12,12 +12,13 @@ import os import re import subprocess import tempfile +import time from dataclasses import dataclass, field from pathlib import Path from typing import Callable from config_manager import Config -from vitis_checker import get_vivado_path, build_tool_command +from vitis_checker import get_vivado_path, get_hw_server_path, build_tool_command @dataclass @@ -87,42 +88,60 @@ def check_zynq_jtag(config: Config, callback: Callable | None = None) -> ZynqChe if callback: callback("start", "Connecting to hw_server...") - # Find vivado executable via Xilinx root scan - vivado_path = get_vivado_path( - xilinx_root=getattr(config, "xilinx_path", ""), - ) - if not vivado_path: + xilinx_root = getattr(config, "xilinx_path", "") + + # ── 1. Find and start hw_server ────────────────────────── + hw_server_path = get_hw_server_path(xilinx_root=xilinx_root) + if not hw_server_path: return ZynqCheckResult( - devices=[], - jtag_chain=[], - is_present=False, - ps_detected=False, - pl_detected=False, + devices=[], jtag_chain=[], is_present=False, + ps_detected=False, pl_detected=False, hw_server_url=hw_server_url, - error="Vivado not found", + error="hw_server not found — install Vivado/Vitis hardware server", ) - # Create temporary TCL script - tcl_content = _build_jtag_query_tcl() - tcl_path = Path(tempfile.gettempdir()) / f"zynq_jtag_check_{os.getpid()}.tcl" + hw_proc: subprocess.Popen | None = None try: + # Start hw_server in the background (it daemonises own I/O) + hw_proc = subprocess.Popen( + build_tool_command(hw_server_path), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + # Give hw_server a moment to bind its port + time.sleep(2) + + # ── 2. Find vivado ──────────────────────────────────── + vivado_path = get_vivado_path(xilinx_root=xilinx_root) + if not vivado_path: + return ZynqCheckResult( + devices=[], jtag_chain=[], is_present=False, + ps_detected=False, pl_detected=False, + hw_server_url=hw_server_url, + error="Vivado not found", + ) + + # ── 3. Create and run JTAG query TCL ────────────────── + tcl_content = _build_jtag_query_tcl() + tcl_path = Path(tempfile.gettempdir()) / f"zynq_jtag_check_{os.getpid()}.tcl" tcl_path.write_text(tcl_content) if callback: callback("progress", "Scanning JTAG chain...") - # Run Vivado in batch mode (with environment setup) - cmd = build_tool_command(vivado_path, "-mode", "batch", "-source", str(tcl_path)) + cmd = build_tool_command( + vivado_path, "-mode", "batch", "-source", str(tcl_path), + ) + python_timeout = config.step_timeouts.get("check_env", 120) result = subprocess.run( cmd, capture_output=True, text=True, - timeout=30, + timeout=python_timeout, ) + tcl_path.unlink(missing_ok=True) output = result.stdout + result.stderr - - # Parse JTAG device information from output jtag_chain = _parse_jtag_chain(output) devices = _filter_zynq_devices(jtag_chain) @@ -135,42 +154,42 @@ def check_zynq_jtag(config: Config, callback: Callable | None = None) -> ZynqChe 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, + devices=devices, jtag_chain=jtag_chain, is_present=len(devices) > 0, - ps_detected=ps_detected, - pl_detected=pl_detected, + 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: return ZynqCheckResult( - devices=[], - jtag_chain=[], - is_present=False, - ps_detected=False, - pl_detected=False, + devices=[], jtag_chain=[], is_present=False, + ps_detected=False, pl_detected=False, hw_server_url=hw_server_url, - error="JTAG scan timed out after 30s", + error=f"JTAG scan timed out after {config.step_timeouts.get('check_env', 120)}s", ) except OSError as e: return ZynqCheckResult( - devices=[], - jtag_chain=[], - is_present=False, - ps_detected=False, - pl_detected=False, + devices=[], jtag_chain=[], is_present=False, + ps_detected=False, pl_detected=False, hw_server_url=hw_server_url, error=f"OS error: {e}", ) finally: - tcl_path.unlink(missing_ok=True) + # ── 4. Kill hw_server ───────────────────────────────── + if hw_proc is not None and hw_proc.poll() is None: + try: + hw_proc.terminate() + hw_proc.wait(timeout=3) + except subprocess.TimeoutExpired: + hw_proc.kill() + hw_proc.wait() + except OSError: + pass # already dead def _build_jtag_query_tcl() -> str: