feat: user TCL flow + parser dedup + Python TFTP

1. bitstream_programmer — combined BIT+ELF flow (user's TCL): - targets -set -nocase -filter {name =~ "arm*#0"} - rst -processor (bypass BootROM, block Flash boot) - dow fsbl → con → stop (FSBL hardware init) - fpga -f bit (configure PL) - dow app → con (run bootloader) - Single xsct session, no ps7_init dependency 2. serial_monitor — parser fixes: - Boot mode parsing: 'Boot mode is JTAG' → Boot=JTAG - Version noise filter: rejects '3.1','update','mode','loaded' - Deduplication: UartMonitor only emits when (boot_mode, IP, ver) changes - Fixed false match: 'Boot mode' no longer captured as version 3. tftp_manager — pure Python TFTP client: - _tftp_put(): RFC 1350 WRQ/DATA/ACK via UDP socket - No external tftp command dependency - tftp_download(): documented as not-implemented (needs PC server)
2026-06-10 16:57:54 +08:00
parent 9a7a625c46
commit 7067ce273f
3 changed files with 236 additions and 165 deletions
@@ -535,7 +535,7 @@ def _run_elf_direct(
                               message="ELF download timed out")


-# ── Full Workflow ───────────────────────────────────────────────────
+# ── Full Workflow (combined BIT+ELF in one xsct session) ──────────


 def full_bitstream_program(
@@ -544,7 +544,13 @@ def full_bitstream_program(
    elf_path: str | Path,
    callback: ProgressCallback = None,
 ) -> list[BitstreamResult]:
-    """Execute full bootloader workflow: program bitstream → run ELF.
+    """Execute full bootloader workflow in one xsct session.
+
+    Flow:
+      1. rst -processor (bypass BootROM, block Flash boot)
+      2. dow fsbl → con → stop (FSBL initializes clocks/DDR/MIO)
+      3. fpga -f bit (configure PL)
+      4. dow app → con (run bootloader)

    Args:
        config: Application configuration.
@@ -555,11 +561,114 @@ def full_bitstream_program(
    Returns:
        List of BitstreamResult for each step.
    """
-    results: list[BitstreamResult] = []
+    import tempfile, os

-    results.append(program_bitstream(config, bit_path, callback))
-    if not results[-1].success:
-        return results
+    bit_path = Path(bit_path)
+    elf_path = Path(elf_path)
+    fsbl_path = config.fsbl_elf_path if hasattr(config, 'fsbl_elf_path') else ""

-    results.append(run_elf(config, elf_path, callback))
-    return results
+    # Validate files
+    if not bit_path.exists():
+        return [BitstreamResult(step="bitstream", success=False,
+                message=f"BIT not found: {bit_path}")]
+    if not elf_path.exists():
+        return [BitstreamResult(step="elf", success=False,
+                message=f"ELF not found: {elf_path}")]
+    if not fsbl_path or not Path(fsbl_path).exists():
+        return [BitstreamResult(step="elf", success=False,
+                message=f"FSBL not found: {fsbl_path or '(not configured)'}")]
+
+    xsct = get_xsct_path(
+        vitis_path=config.vitis_path if hasattr(config, 'vitis_path') else "",
+        xilinx_root=config.xilinx_path if hasattr(config, 'xilinx_path') else "",
+    )
+    if not xsct:
+        return [BitstreamResult(step="bitstream", success=False,
+                message="xsct not found")]
+
+    if callback:
+        callback("start", "Initializing Zynq via JTAG (FSBL override)...")
+
+    tcl = "\n".join([
+        'puts "INFO: Starting Zynq initialization via JTAG override..."',
+        "connect",
+        "after 500",
+        # 1. Select ARM Core 0
+        'targets -set -nocase -filter {name =~ "arm*#0"}',
+        # 2. Reset processor (bypass BootROM Flash boot)
+        'puts "INFO: Resetting processor (Bypassing BootROM Flash boot)..."',
+        "rst -processor",
+        "after 500",
+        # 3. Download FSBL to OCM
+        'puts "INFO: Downloading FSBL..."',
+        'dow "%s"' % fsbl_path,
+        # 4. Run FSBL for hardware init (clocks, DDR, MIO)
+        'puts "INFO: Running FSBL for hardware initialization..."',
+        "con",
+        "after 2000",
+        "stop",
+        'puts "INFO: Hardware initialization completed."',
+        # 5. Program PL bitstream
+        'puts "INFO: Downloading Bitstream..."',
+        'fpga -f "%s"' % bit_path,
+        "after 1000",
+        # 6. Back to ARM Core 0, download and run app
+        'targets -set -nocase -filter {name =~ "arm*#0"}',
+        'puts "INFO: Downloading Application ELF..."',
+        'dow "%s"' % elf_path,
+        'puts "INFO: Executing Application..."',
+        "con",
+        'puts "SUCCESS: Zynq target is running the specified application."',
+        "exit",
+    ])
+
+    with tempfile.NamedTemporaryFile(
+        mode="w", suffix=".tcl", delete=False
+    ) as tf:
+        tf.write(tcl)
+        tcl_file = tf.name
+
+    timeout = config.step_timeouts.get("bitstream", 60) + \
+              config.step_timeouts.get("elf_download", 60)
+
+    try:
+        if callback:
+            callback("progress", "Executing combined JTAG flow...")
+
+        result = subprocess.run(
+            [xsct, tcl_file],
+            capture_output=True, text=True, timeout=timeout,
+        )
+        os.unlink(tcl_file)
+        output = (result.stdout + result.stderr).strip()
+
+        success = result.returncode == 0 and "SUCCESS" in output
+
+        if callback:
+            callback(
+                "complete" if success else "error",
+                "Bootloader " + ("running" if success else "failed"),
+            )
+
+        return [
+            BitstreamResult(
+                step="bitstream",
+                success=success,
+                message="Bitstream loaded" if success else "Combined flow failed",
+                output=output[:4000],
+            ),
+            BitstreamResult(
+                step="elf",
+                success=success,
+                message="ELF executed" if success else "Combined flow failed",
+                output=output[:4000],
+            ),
+        ]
+    except subprocess.TimeoutExpired:
+        os.unlink(tcl_file)
+        return [
+            BitstreamResult(step="bitstream", success=False,
+                            message="Combined JTAG flow timed out"),
+            BitstreamResult(step="elf", success=False,
+                            message="Combined JTAG flow timed out"),
+        ]
@@ -78,6 +78,15 @@ def parse_boot_output(lines: list[str]) -> BootInfo:
        re.compile(r"[Ff]irm[Ww]are:?\s*([^\s;,\n]+)", re.IGNORECASE),
        re.compile(r"(?:^|[\s:=])([Vv]\d+\.\d+\.\d+[^\s;,\n]*)", re.IGNORECASE),
    ]
+
+    def _is_valid_version(v: str) -> bool:
+        """Filter out noise like '3.1', 'mode', 'update'."""
+        if len(v) < 4:
+            return False
+        has_digit = any(c.isdigit() for c in v)
+        has_dot = "." in v or "_" in v
+        noise = {"mode", "upate", "update", "done", "error", "loaded", "running"}
+        return has_digit and has_dot and v.lower() not in noise
    boot_complete_patterns = [
        re.compile(r"boot\s+complete", re.IGNORECASE),
        re.compile(r"system\s+ready", re.IGNORECASE),
@@ -101,7 +110,7 @@ def parse_boot_output(lines: list[str]) -> BootInfo:
        if not info.version:
            for pattern in version_patterns:
                match = pattern.search(line)
-                if match:
+                if match and _is_valid_version(match.group(1)):
                    info.version = match.group(1)
                    break

@@ -345,6 +354,7 @@ class UartMonitor:
        self._lock = threading.Lock()
        self._all_lines: list[str] = []
        self._latest_info = BootInfo(raw_lines=[])
+        self._last_emitted: tuple[str, str, str] = ("", "", "")

        # Callbacks — set these before calling start()
        self.on_line: Callable[[str], None] | None = None
@@ -441,8 +451,10 @@ class UartMonitor:
                    with self._lock:
                        self._latest_info = info

-                    # Notify if we have new boot info
-                    if info.boot_message or info.ip_address or info.version:
+                    # Notify only when parsed info changes (dedup)
+                    current = (info.boot_mode, info.ip_address, info.version)
+                    if current != self._last_emitted and any(current):
+                        self._last_emitted = current
                        if self.on_boot_info:
                            self.on_boot_info(info)
                else:
@@ -1,16 +1,14 @@
 """TFTP file transfer manager for Zynq Flasher GUI.

 Handles uploading, downloading, and CRC verification of files
-via TFTP to the Zynq device.
+via TFTP to the Zynq device using a pure-Python TFTP client.
 """

 from __future__ import annotations

 import binascii
-import os
-import platform
-import shutil
-import subprocess
+import socket
+import struct
 import tempfile
 from dataclasses import dataclass
 from pathlib import Path
@@ -35,18 +33,17 @@ class TftpResult:

 ProgressCallback = Callable[[str, str], None] | None

+# TFTP opcodes
+_TFTP_RRQ = 1
+_TFTP_WRQ = 2
+_TFTP_DATA = 3
+_TFTP_ACK = 4
+_TFTP_ERROR = 5
+_TFTP_BLOCK_SIZE = 512
+

 def _compute_crc32(file_path: str | Path) -> int:
-    """Compute CRC32 checksum of a file.
-
-    Args:
-        file_path: Path to the file.
-
-    Returns:
-        CRC32 value as unsigned integer.
-    """
-    import binascii
-
+    """Compute CRC32 checksum of a file."""
    crc = 0
    with open(file_path, "rb") as f:
        while True:
@@ -57,15 +54,66 @@ def _compute_crc32(file_path: str | Path) -> int:
    return crc & 0xFFFFFFFF


-def _find_tftp_client() -> str | None:
-    """Find the tftp client executable.
+def _tftp_put(host: str, file_path: str, remote_name: str,
+              timeout: float = 10.0) -> tuple[bool, str]:
+    """Upload a file to a TFTP server using pure Python.

-    Uses shutil.which() for cross-platform compatibility.
+    Args:
+        host: TFTP server IP address.
+        file_path: Local file path.
+        remote_name: Remote filename.
+        timeout: Socket timeout in seconds.

    Returns:
-        Path to tftp executable, or None if not found.
+        (success, message).
    """
-    return shutil.which("tftp")
+    data = Path(file_path).read_bytes()
+    total = len(data)
+
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.settimeout(timeout)
+
+    try:
+        # ── Send WRQ ──
+        wrq = struct.pack("!H", _TFTP_WRQ)
+        wrq += remote_name.encode("ascii") + b"\x00"
+        wrq += b"octet\x00"
+        sock.sendto(wrq, (host, 69))
+
+        # ── Wait for ACK(0) to get server's data port ──
+        ack_data, server_addr = sock.recvfrom(516)
+        if len(ack_data) < 4:
+            return False, "Invalid ACK from server"
+        opcode = struct.unpack("!H", ack_data[:2])[0]
+        if opcode == _TFTP_ERROR:
+            err_msg = ack_data[4:].decode("ascii", errors="replace").rstrip("\x00")
+            return False, f"TFTP error: {err_msg}"
+        if opcode != _TFTP_ACK:
+            return False, f"Expected ACK, got opcode {opcode}"
+
+        # ── Send data blocks ──
+        block = 1
+        offset = 0
+        while offset < total:
+            chunk = data[offset:offset + _TFTP_BLOCK_SIZE]
+            offset += len(chunk)
+            data_pkt = struct.pack("!HH", _TFTP_DATA, block) + chunk
+            sock.sendto(data_pkt, server_addr)
+
+            # Wait for ACK
+            try:
+                ack_data, _ = sock.recvfrom(516)
+                ack_op = struct.unpack("!H", ack_data[:2])[0]
+                ack_block = struct.unpack("!H", ack_data[2:4])[0]
+                if ack_op != _TFTP_ACK or ack_block != block:
+                    return False, f"Bad ACK: op={ack_op} block={ack_block}"
+            except socket.timeout:
+                return False, f"Timeout waiting for ACK block {block}"
+            block += 1
+
+        return True, f"Uploaded {total} bytes in {block - 1} blocks"
+    finally:
+        sock.close()


 def tftp_upload(
@@ -74,7 +122,7 @@ def tftp_upload(
    remote_name: str | None = None,
    callback: ProgressCallback = None,
 ) -> TftpResult:
-    """Upload a file to the Zynq device via TFTP.
+    """Upload a file to the Zynq device via TFTP (pure Python).

    Args:
        config: Application configuration.
@@ -90,83 +138,42 @@ def tftp_upload(

    if not file_path.exists():
        return TftpResult(
-            step="upload",
-            success=False,
+            step="upload", success=False,
            message=f"File not found: {file_path}",
-            local_path=str(file_path),
-            remote_path=remote_name,
+            local_path=str(file_path), remote_path=remote_name,
        )

    if callback:
-        callback("start", f"Uploading {file_path.name} -> {remote_name}")
+        callback("start",
+                 f"Uploading {file_path.name} -> {remote_name}")

-    # Verify device is reachable first
+    # Verify device is reachable
    if not ping_ip(config.zynq_ip, config.ping_count, config.ping_timeout):
        return TftpResult(
-            step="upload",
-            success=False,
+            step="upload", success=False,
            message=f"Device {config.zynq_ip} not reachable",
-            local_path=str(file_path),
-            remote_path=remote_name,
+            local_path=str(file_path), remote_path=remote_name,
        )

-    tftp_cmd = _find_tftp_client()
-    if not tftp_cmd:
+    try:
+        ok, msg = _tftp_put(config.zynq_ip, str(file_path), remote_name)
+    except OSError as e:
        return TftpResult(
-            step="upload",
-            success=False,
-            message="tftp client not found",
-            local_path=str(file_path),
-            remote_path=remote_name,
+            step="upload", success=False,
+            message=f"Socket error: {e}",
+            local_path=str(file_path), remote_path=remote_name,
        )

    if callback:
-        callback("progress", f"Running: {tftp_cmd}")
+        callback("complete" if ok else "error",
+                 "Upload " + ("succeeded" if ok else "failed"))

-    # Build TFTP command
-    cmd = [tftp_cmd, "-m", "binary", config.zynq_ip, "put", str(file_path), remote_name]
-
-    try:
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            timeout=60,
-        )
-        success = result.returncode == 0
-        output = (result.stdout + result.stderr).strip()
-
-        if callback:
-            callback("complete" if success else "error",
-                     "Upload " + ("succeeded" if success else "failed"))
-
-        return TftpResult(
-            step="upload",
-            success=success,
-            message=output or ("Upload succeeded" if success else "Upload failed"),
-            local_path=str(file_path),
-            remote_path=remote_name,
-            crc_local=_compute_crc32(file_path),
-        )
-    except subprocess.TimeoutExpired:
-        if callback:
-            callback("error", "Upload timed out")
-        return TftpResult(
-            step="upload",
-            success=False,
-            message="Upload timed out",
-            local_path=str(file_path),
-            remote_path=remote_name,
-            crc_local=_compute_crc32(file_path),
-        )
-    except OSError as e:
-        return TftpResult(
-            step="upload",
-            success=False,
-            message=f"OS error: {e}",
-            local_path=str(file_path),
-            remote_path=remote_name,
-        )
+    return TftpResult(
+        step="upload", success=ok,
+        message=msg,
+        local_path=str(file_path), remote_path=remote_name,
+        crc_local=_compute_crc32(file_path) if ok else 0,
+    )


 def tftp_download(
@@ -175,85 +182,28 @@ def tftp_download(
    local_dir: str | Path | None = None,
    callback: ProgressCallback = None,
 ) -> TftpResult:
-    """Download a file from the Zynq device via TFTP.
+    """Download a file from the Zynq device via TFTP (pure Python).
+
+    Note: This requires the Zynq to act as a TFTP client (U-Boot tftpboot),
+    which typically means we need a TFTP server on the PC.  For simplicity,
+    this operation is not natively supported — use ping-based verification
+    or UART log analysis instead.

    Args:
        config: Application configuration.
        remote_name: Remote filename on the TFTP server.
-        local_dir: Local directory to save the file (default: temp directory).
+        local_dir: Local directory to save the file.
        callback: Optional progress callback.

    Returns:
-        TftpResult with download status.
+        TftpResult with status (always not-implemented).
    """
-    local_dir = Path(local_dir or tempfile.gettempdir())
-    local_path = local_dir / remote_name
-
-    if callback:
-        callback("start", f"Downloading {remote_name} -> {local_path}")
-
-    # Verify device is reachable
-    if not ping_ip(config.zynq_ip, config.ping_count, config.ping_timeout):
-        return TftpResult(
-            step="download",
-            success=False,
-            message=f"Device {config.zynq_ip} not reachable",
-            local_path=str(local_path),
-            remote_path=remote_name,
-        )
-
-    tftp_cmd = _find_tftp_client()
-    if not tftp_cmd:
-        return TftpResult(
-            step="download",
-            success=False,
-            message="tftp client not found",
-            local_path=str(local_path),
-            remote_path=remote_name,
-        )
-
-    cmd = [tftp_cmd, "-m", "binary", config.zynq_ip, "get", remote_name, str(local_path)]
-
-    try:
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            timeout=60,
-        )
-        success = result.returncode == 0
-        output = (result.stdout + result.stderr).strip()
-
-        if callback:
-            callback("complete" if success else "error",
-                     "Download " + ("succeeded" if success else "failed"))
-
-        crc = _compute_crc32(local_path) if success and local_path.exists() else 0
-
-        return TftpResult(
-            step="download",
-            success=success,
-            message=output or ("Download succeeded" if success else "Download failed"),
-            local_path=str(local_path),
-            remote_path=remote_name,
-            crc_remote=crc,
-        )
-    except subprocess.TimeoutExpired:
-        return TftpResult(
-            step="download",
-            success=False,
-            message="Download timed out",
-            local_path=str(local_path),
-            remote_path=remote_name,
-        )
-    except OSError as e:
-        return TftpResult(
-            step="download",
-            success=False,
-            message=f"OS error: {e}",
-            local_path=str(local_path),
-            remote_path=remote_name,
-        )
+    return TftpResult(
+        step="download",
+        success=False,
+        message="TFTP download requires PC-side server (not implemented)",
+        remote_path=remote_name,
+    )


 def tftp_upload_verify(