From 66c04da5e004aef9603ea4d4c1f7481b5ff70321 Mon Sep 17 00:00:00 2001 From: Jeremy Shen Date: Thu, 11 Jun 2026 17:24:31 +0800 Subject: [PATCH] fix: stream Vivado output live, remove broken normalize_port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/gui/widgets.py | 3 +- src/reboot_manager.py | 3 +- src/serial_monitor.py | 31 +--- src/zynq_checker.py | 324 +++++++++++++++++++----------------------- 4 files changed, 154 insertions(+), 207 deletions(-) diff --git a/src/gui/widgets.py b/src/gui/widgets.py index e624a3c..420807b 100644 --- a/src/gui/widgets.py +++ b/src/gui/widgets.py @@ -1075,8 +1075,7 @@ class UartMonitorWindow(ctk.CTkToplevel): def _bg_open() -> None: try: import serial - from serial_monitor import normalize_port - ser = serial.Serial(normalize_port(self._port), self._baudrate, timeout=1) + ser = serial.Serial(self._port, self._baudrate, timeout=1) ser.close() self.after(0, self._on_open_success) except Exception as e: diff --git a/src/reboot_manager.py b/src/reboot_manager.py index df90749..d30e3a2 100644 --- a/src/reboot_manager.py +++ b/src/reboot_manager.py @@ -125,10 +125,9 @@ def reboot_via_serial( try: import serial - from serial_monitor import normalize_port with serial.Serial( - normalize_port(config.serial_port), + config.serial_port, config.serial_baudrate, timeout=5, ) as ser: diff --git a/src/serial_monitor.py b/src/serial_monitor.py index 7982dbd..9e9acd5 100644 --- a/src/serial_monitor.py +++ b/src/serial_monitor.py @@ -7,7 +7,6 @@ Provides UartMonitor for continuous background monitoring. from __future__ import annotations -import platform import re import threading import time @@ -18,26 +17,6 @@ import serial import serial.tools.list_ports -def normalize_port(port: str) -> str: - """Normalize a serial port name for the current platform. - - On Windows, COM ports numbered >= 10 require the NT namespace - prefix (\\\\.\\COM11). Without it, CreateFile fails with - 'Access denied' or 'file not found'. - - Args: - port: Raw port name from config or detection. - - Returns: - Normalized port name suitable for serial.Serial(). - """ - if platform.system() == "Windows": - match = re.match(r"^COM(\d+)$", port, re.IGNORECASE) - if match and int(match.group(1)) >= 10: - return rf"\\.\{port}" - return port - - @dataclass class BootInfo: """Parsed information extracted from Zynq boot output.""" @@ -190,7 +169,7 @@ def check_uart_available( # 3. Try to open and read try: - with serial.Serial(normalize_port(port), baudrate, timeout=timeout) as ser: + with serial.Serial(port, baudrate, timeout=timeout) as ser: ser.reset_input_buffer() lines: list[str] = [] while ser.in_waiting: @@ -228,7 +207,7 @@ def read_serial_stream( serial.SerialException: If the port cannot be opened. """ lines: list[str] = [] - with serial.Serial(normalize_port(port), baudrate, timeout=timeout) as ser: + with serial.Serial(port, baudrate, timeout=timeout) as ser: # Clear any pending data ser.reset_input_buffer() while ser.in_waiting: @@ -264,7 +243,7 @@ def test_serial_version( import serial try: - with serial.Serial(normalize_port(port), baudrate, timeout=timeout) as ser: + with serial.Serial(port, baudrate, timeout=timeout) as ser: ser.reset_input_buffer() ser.reset_output_buffer() @@ -405,7 +384,7 @@ class UartMonitor: # Quick port check before starting thread try: - test_ser = serial.Serial(normalize_port(self._port), self._baudrate, timeout=1) + test_ser = serial.Serial(self._port, self._baudrate, timeout=1) test_ser.close() except serial.SerialException as e: if self.on_error: @@ -432,7 +411,7 @@ class UartMonitor: """Main read loop running in background thread.""" try: ser = serial.Serial( - normalize_port(self._port), + self._port, self._baudrate, timeout=self._timeout, ) diff --git a/src/zynq_checker.py b/src/zynq_checker.py index d258416..56767fd 100644 --- a/src/zynq_checker.py +++ b/src/zynq_checker.py @@ -1,13 +1,13 @@ """Zynq JTAG presence checker for Zynq Flasher GUI. -Uses Vivado batch-mode TCL to scan the JTAG chain via hw_server, -returning device names, IDCODEs, and PS/PL presence. +Starts hw_server explicitly, then uses Vivado batch-mode TCL to scan +the JTAG chain. On failure, Vivado's stdout/stderr is returned in the +error message for debugging. """ from __future__ import annotations import re -import socket import subprocess import tempfile import time @@ -24,7 +24,7 @@ class JtagDevice: """Information about a device on the JTAG chain.""" name: str - idcode: str = "" # IDCODE in hex (e.g. "03736093") + idcode: str = "" is_zynq: bool = False is_arm_dap: bool = False is_pl_device: bool = False @@ -35,9 +35,9 @@ class ZynqDevice: """Information about a detected Zynq device on the JTAG chain.""" name: str - part_number: str # e.g. "XC7Z100" - family: str # e.g. "zynq7" - speed: str # e.g. "-1" + part_number: str + family: str + speed: str idcode: str = "" is_programmable: bool = False @@ -56,31 +56,13 @@ class ZynqCheckResult: # ── Patterns ────────────────────────────────────────────────────────── -# Vivado TCL output: DEVICE:arm_dap_0 IDCODE:0100101110100...0111 _DEVICE_LINE_RE = re.compile(r"^DEVICE:(\S+)\s+IDCODE:([01]+)") - -# Zynq PL: xc7z100, xc7z100_0, xc7z100_1, etc. _ZYNQ_RE = re.compile(r"^xc\d+z\d+[a-z]?\d+", re.IGNORECASE) - -# ARM DAP (PS): arm_dap, arm_dap_0, arm_dap_1 _ARM_DAP_RE = re.compile(r"^arm_dap", re.IGNORECASE) - HW_SERVER_PORT = 3121 -# ── Helpers ─────────────────────────────────────────────────────────── - -def _port_open(host: str = "localhost", port: int = HW_SERVER_PORT) -> bool: - """Check if a TCP port is accepting connections.""" - try: - with socket.create_connection((host, port), timeout=1): - return True - except OSError: - return False - - def _idcode_to_hex(binary_str: str) -> str: - """Convert Vivado's binary IDCODE to uppercase hex.""" try: return f"{int(binary_str, 2):08X}" except ValueError: @@ -92,12 +74,11 @@ def _idcode_to_hex(binary_str: str) -> str: def check_zynq_jtag( config: Config, callback: Callable | None = None ) -> ZynqCheckResult: - """Check for Zynq devices on the JTAG chain via Vivado + hw_server. + """Check for Zynq devices on the JTAG chain. - Lifecycle: - 1. If hw_server is already running (port 3121 open) → reuse it - 2. If not running → start it, kill it when done - 3. Run Vivado batch TCL to enumerate JTAG devices + 1. Start hw_server (Vivado may not auto-start on Windows) + 2. Run Vivado batch TCL to enumerate JTAG devices + 3. On failure, include Vivado output in error for debugging Args: config: Application configuration. @@ -115,128 +96,150 @@ def check_zynq_jtag( # ── 1. Find tools ────────────────────────────────────────── vivado_path = get_vivado_path(xilinx_root=xilinx_root) if not vivado_path: - return ZynqCheckResult( - devices=[], jtag_chain=[], is_present=False, - ps_detected=False, pl_detected=False, - hw_server_url=hw_server_url, - error="Vivado not found", - ) + return _fail("Vivado not found", hw_server_url) hw_server_path = get_hw_server_path(xilinx_root=xilinx_root) - hw_server_was_running = _port_open() + if not hw_server_path: + return _fail("hw_server not found", hw_server_url) - # ── 2. Start hw_server if needed ─────────────────────────── + # ── 2. Start hw_server ───────────────────────────────────── hw_proc: subprocess.Popen | None = None - if not hw_server_was_running: - if not hw_server_path: - return ZynqCheckResult( - devices=[], jtag_chain=[], is_present=False, - ps_detected=False, pl_detected=False, - hw_server_url=hw_server_url, - error="hw_server not found — install Vivado/Vitis", - ) - try: - hw_proc = subprocess.Popen( - build_tool_command(hw_server_path), - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - # Wait for port to become available - for _ in range(30): # 3s max - if _port_open(): - break - time.sleep(0.1) - else: - return ZynqCheckResult( - devices=[], jtag_chain=[], is_present=False, - ps_detected=False, pl_detected=False, - hw_server_url=hw_server_url, - error="hw_server started but port 3121 not open", - ) - except OSError as e: - return ZynqCheckResult( - devices=[], jtag_chain=[], is_present=False, - ps_detected=False, pl_detected=False, - hw_server_url=hw_server_url, - error=f"Failed to start hw_server: {e}", - ) - - # ── 3. Run Vivado TCL ────────────────────────────────────── try: - tcl = _build_jtag_query_tcl() - if callback: - callback("progress", "Scanning JTAG chain...") - - python_timeout = config.step_timeouts.get("check_env", 120) - result = subprocess.run( - build_tool_command(vivado_path, "-mode", "batch", "-source", tcl), - capture_output=True, text=True, timeout=python_timeout, + hw_proc = subprocess.Popen( + build_tool_command(hw_server_path), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, ) - # Clean up temp TCL file + # Wait up to 10s for hw_server to bind + started = False + for _ in range(100): + time.sleep(0.1) + if _port_listening(): + started = True + break + if not started: + return _fail("hw_server did not bind port 3121 within 10s", hw_server_url) + except OSError as e: + return _fail(f"Failed to start hw_server: {e}", hw_server_url) + + if callback: + callback("progress", "Scanning JTAG chain...") + + # ── 3. Run Vivado TCL (stream output to callback) ────────── + tcl = _build_jtag_query_tcl() + python_timeout = config.step_timeouts.get("check_env", 120) + output_lines: list[str] = [] + + try: + proc = subprocess.Popen( + build_tool_command(vivado_path, "-mode", "batch", "-source", tcl), + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, + ) + # Read line by line so user sees real-time progress + deadline = time.monotonic() + python_timeout + while True: + line = proc.stdout.readline() if proc.stdout else "" + if not line: + if proc.poll() is not None: + break + if time.monotonic() > deadline: + proc.kill() + proc.wait() + return _fail( + f"JTAG scan timed out after {python_timeout}s", + hw_server_url, + ) + time.sleep(0.1) + continue + line = line.rstrip("\n\r") + output_lines.append(line) + if callback and line.strip(): + callback("progress", line) + except OSError as e: + return _fail(f"Failed to run Vivado: {e}", hw_server_url) + finally: try: Path(tcl).unlink(missing_ok=True) except Exception: pass - output = result.stdout + result.stderr - jtag_chain = _parse_jtag_chain(output) - devices = _filter_zynq_devices(jtag_chain) + output = "\n".join(output_lines) + # Parse results + jtag_chain = _parse_jtag_chain(output) + devices = _filter_zynq_devices(jtag_chain) + + if devices: if callback: - 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") - - ps_detected = any(d.is_arm_dap for d in jtag_chain) - pl_detected = any(d.is_pl_device for d in jtag_chain) - + 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=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: - return ZynqCheckResult( - devices=[], jtag_chain=[], is_present=False, + devices=[], jtag_chain=jtag_chain, + is_present=False, ps_detected=False, pl_detected=False, hw_server_url=hw_server_url, - error=f"JTAG scan timed out after {config.step_timeouts.get('check_env', 120)}s", + error=msg, ) - except OSError as e: - return ZynqCheckResult( - devices=[], jtag_chain=[], is_present=False, - ps_detected=False, pl_detected=False, - hw_server_url=hw_server_url, - error=f"OS error: {e}", - ) - finally: - # ── 4. Kill hw_server only if WE started it ──────────── - 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 + + 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="", + ) + + +# ── Helpers ─────────────────────────────────────────────────────────── + +def _fail(error: str, url: str) -> ZynqCheckResult: + return ZynqCheckResult( + devices=[], jtag_chain=[], is_present=False, + ps_detected=False, pl_detected=False, + hw_server_url=url, error=error, + ) + + +def _port_listening(port: int = HW_SERVER_PORT) -> bool: + """Check if hw_server is listening on its TCP port.""" + import socket + try: + with socket.create_connection(("localhost", port), timeout=1): + return True + except OSError: + return False + + +def _extract_relevant(output: str) -> str: + """Extract error/warning lines from Vivado output for user display.""" + lines = [] + for line in output.splitlines(): + line = line.strip() + if not line: + continue + if any(kw in line for kw in ("ERROR", "WARNING", "CRITICAL", "FAIL", "cannot", "No")): + lines.append(line) + return " | ".join(lines[-5:]) # Last 5 relevant lines # ── TCL builder ──────────────────────────────────────────────────────── def _build_jtag_query_tcl() -> str: - """Build Vivado TCL script file for JTAG device enumeration. - - Returns: - Path to the temporary TCL script file. - """ + """Build Vivado TCL script file for JTAG device enumeration.""" content = """\ open_hw_manager connect_hw_server @@ -259,17 +262,11 @@ quit # ── Parsers ──────────────────────────────────────────────────────────── def _parse_jtag_chain(output: str) -> list[JtagDevice]: - """Parse Vivado TCL output into JtagDevice list. - - Expected format: - DEVICE:arm_dap_0 IDCODE:01001011101000000000010001110111 - DEVICE:xc7z100_1 IDCODE:00000011011100110110000010010011 - """ + """Parse Vivado TCL output into JtagDevice list.""" devices: list[JtagDevice] = [] section = _extract_section(output, "===JTAG_START===", "===JTAG_END===") if not section: return devices - for line in section.splitlines(): line = line.strip() if not line: @@ -279,12 +276,10 @@ def _parse_jtag_chain(output: str) -> list[JtagDevice]: name = match.group(1) idcode_hex = _idcode_to_hex(match.group(2)) devices.append(_classify_jtag_device(name, idcode_hex)) - return devices def _filter_zynq_devices(jtag_chain: list[JtagDevice]) -> list[ZynqDevice]: - """Filter Zynq devices from JTAG chain, propagating IDCODE.""" zynq_devices: list[ZynqDevice] = [] for jtag_dev in jtag_chain: if jtag_dev.is_zynq: @@ -295,7 +290,6 @@ def _filter_zynq_devices(jtag_chain: list[JtagDevice]) -> list[ZynqDevice]: def _extract_section(text: str, start_marker: str, end_marker: str) -> str | None: - """Extract text between last occurrence of two markers.""" start_idx = text.rfind(start_marker) end_idx = text.rfind(end_marker) if start_idx == -1 or end_idx == -1 or start_idx >= end_idx: @@ -304,51 +298,36 @@ def _extract_section(text: str, start_marker: str, end_marker: str) -> str | Non def _classify_jtag_device(device_name: str, idcode: str = "") -> JtagDevice: - """Classify a JTAG device by name (Vivado format: xc7z100_1, arm_dap_0).""" is_zynq = bool(_ZYNQ_RE.match(device_name)) is_arm_dap = bool(_ARM_DAP_RE.match(device_name)) - is_pl_device = is_zynq and not is_arm_dap - return JtagDevice( name=device_name, idcode=idcode, is_zynq=is_zynq, is_arm_dap=is_arm_dap, - is_pl_device=is_pl_device, + is_pl_device=is_zynq and not is_arm_dap, ) def _parse_zynq_device(device_name: str, idcode: str = "") -> ZynqDevice | None: - """Parse Zynq device name (Vivado format: xc7z100_1).""" match = _ZYNQ_RE.match(device_name) if not match: return None - - # Strip index suffix: xc7z100_1 → XC7Z100 base = device_name.rsplit("_", 1)[0] if "_" in device_name else device_name part_number = base.upper() - family = "zynq7" speed = "-1" speed_match = re.search(r"(\d+)$", part_number) if speed_match: - speed_num = int(speed_match.group(1)) - speed = "-2" if speed_num >= 2 else "-1" - + speed = "-2" if int(speed_match.group(1)) >= 2 else "-1" return ZynqDevice( - name=device_name, - part_number=part_number, - family=family, - speed=speed, - idcode=idcode, - is_programmable=True, + name=device_name, part_number=part_number, + family=family, speed=speed, + idcode=idcode, is_programmable=True, ) -# ── Status dict ──────────────────────────────────────────────────────── - def get_zynq_status_dict(result: ZynqCheckResult) -> dict: - """Convert ZynqCheckResult to a serializable dict.""" return { "is_present": result.is_present, "ps_detected": result.ps_detected, @@ -356,24 +335,15 @@ def get_zynq_status_dict(result: ZynqCheckResult) -> dict: "hw_server_url": result.hw_server_url, "error": result.error, "jtag_chain": [ - { - "name": dev.name, - "idcode": dev.idcode, - "is_zynq": dev.is_zynq, - "is_arm_dap": dev.is_arm_dap, - "is_pl_device": dev.is_pl_device, - } - for dev in result.jtag_chain + {"name": d.name, "idcode": d.idcode, + "is_zynq": d.is_zynq, "is_arm_dap": d.is_arm_dap, + "is_pl_device": d.is_pl_device} + for d in result.jtag_chain ], "devices": [ - { - "name": dev.name, - "part_number": dev.part_number, - "family": dev.family, - "speed": dev.speed, - "idcode": dev.idcode, - "is_programmable": dev.is_programmable, - } - for dev in result.devices + {"name": d.name, "part_number": d.part_number, + "family": d.family, "speed": d.speed, + "idcode": d.idcode, "is_programmable": d.is_programmable} + for d in result.devices ], }