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.
This commit is contained in:
+1
-2
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
+5
-26
@@ -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,
|
||||
)
|
||||
|
||||
+116
-146
@@ -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()
|
||||
|
||||
# ── 2. Start hw_server if needed ───────────────────────────
|
||||
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",
|
||||
)
|
||||
return _fail("hw_server not found", hw_server_url)
|
||||
|
||||
# ── 2. Start hw_server ─────────────────────────────────────
|
||||
hw_proc: subprocess.Popen | None = None
|
||||
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
|
||||
# Wait up to 10s for hw_server to bind
|
||||
started = False
|
||||
for _ in range(100):
|
||||
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",
|
||||
)
|
||||
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 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}",
|
||||
)
|
||||
return _fail(f"Failed to start hw_server: {e}", hw_server_url)
|
||||
|
||||
# ── 3. Run Vivado TCL ──────────────────────────────────────
|
||||
try:
|
||||
tcl = _build_jtag_query_tcl()
|
||||
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)
|
||||
result = subprocess.run(
|
||||
output_lines: list[str] = []
|
||||
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
build_tool_command(vivado_path, "-mode", "batch", "-source", tcl),
|
||||
capture_output=True, text=True, timeout=python_timeout,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
)
|
||||
# Clean up temp TCL file
|
||||
# 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
|
||||
output = "\n".join(output_lines)
|
||||
|
||||
# Parse results
|
||||
jtag_chain = _parse_jtag_chain(output)
|
||||
devices = _filter_zynq_devices(jtag_chain)
|
||||
|
||||
if callback:
|
||||
if devices:
|
||||
if callback:
|
||||
ps_note = "PS" if any(d.is_arm_dap for d in jtag_chain) else ""
|
||||
pl_note = "PL" if any(d.is_pl_device for d in jtag_chain) else ""
|
||||
notes = ", ".join(filter(None, [ps_note, pl_note]))
|
||||
callback("complete", f"Found {len(devices)} Zynq device(s) [{notes}]")
|
||||
else:
|
||||
callback("error", "No Zynq device found on JTAG chain")
|
||||
# Include Vivado output so user can debug
|
||||
err_detail = _extract_relevant(output)
|
||||
msg = "No Zynq device found on JTAG chain"
|
||||
if err_detail:
|
||||
msg += f" | Vivado: {err_detail[:300]}"
|
||||
if callback:
|
||||
callback("error", msg)
|
||||
return ZynqCheckResult(
|
||||
devices=[], jtag_chain=jtag_chain,
|
||||
is_present=False,
|
||||
ps_detected=False, pl_detected=False,
|
||||
hw_server_url=hw_server_url,
|
||||
error=msg,
|
||||
)
|
||||
|
||||
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,
|
||||
is_present=True,
|
||||
ps_detected=ps_detected, pl_detected=pl_detected,
|
||||
hw_server_url=hw_server_url,
|
||||
error="" if devices else "No Zynq device found on JTAG chain",
|
||||
error="",
|
||||
)
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
|
||||
# ── 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=hw_server_url,
|
||||
error=f"JTAG scan timed out after {config.step_timeouts.get('check_env', 120)}s",
|
||||
hw_server_url=url, error=error,
|
||||
)
|
||||
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:
|
||||
|
||||
|
||||
def _port_listening(port: int = HW_SERVER_PORT) -> bool:
|
||||
"""Check if hw_server is listening on its TCP port."""
|
||||
import socket
|
||||
try:
|
||||
hw_proc.terminate()
|
||||
hw_proc.wait(timeout=3)
|
||||
except subprocess.TimeoutExpired:
|
||||
hw_proc.kill()
|
||||
hw_proc.wait()
|
||||
with socket.create_connection(("localhost", port), timeout=1):
|
||||
return True
|
||||
except OSError:
|
||||
pass
|
||||
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
|
||||
],
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user