|
| 1 | +#!/usr/bin/env python3 |
| 2 | +""" |
| 3 | +Run a verilator simulation, streaming verilator log and UART output. |
| 4 | +Optionally toggle tracing (SIGUSR1) when a trigger string appears on the UART. |
| 5 | +
|
| 6 | +The UART device path is extracted from the verilator log (a line in the |
| 7 | +format "$ screen /dev/pts/N"). |
| 8 | +
|
| 9 | +Usage: |
| 10 | + verilator_trace_trigger.py [--trace-trigger=STRING] [--verilator-bin=PATH] <elf> [<elf> ...] |
| 11 | +
|
| 12 | +Arguments: |
| 13 | + elfs One or more ELF files to pass to verilator with -E |
| 14 | +
|
| 15 | +Options: |
| 16 | + --verilator-bin PATH Path to the verilator simulation binary |
| 17 | + (default: build/lowrisc_mocha_top_chip_verilator_0/ |
| 18 | + sim-verilator/Vtop_chip_verilator) |
| 19 | + --trace-trigger STR Watch the UART for STR and toggle tracing (SIGUSR1) when found |
| 20 | +""" |
| 21 | + |
| 22 | +import argparse |
| 23 | +import asyncio |
| 24 | +import re |
| 25 | +import sys |
| 26 | +from pathlib import Path |
| 27 | +from typing import Optional |
| 28 | + |
| 29 | +import serial |
| 30 | + |
| 31 | +_CYAN = "\033[36m" |
| 32 | +_GREEN = "\033[32m" |
| 33 | +_DIM = "\033[2m" |
| 34 | +_RESET = "\033[0m" |
| 35 | + |
| 36 | +_DEFAULT_VERILATOR_BIN = str( |
| 37 | + Path(__file__).resolve().parent.parent |
| 38 | + / "build/lowrisc_mocha_top_chip_verilator_0/sim-verilator/Vtop_chip_verilator" |
| 39 | +) |
| 40 | + |
| 41 | + |
| 42 | +def parse_args() -> argparse.Namespace: |
| 43 | + parser = argparse.ArgumentParser( |
| 44 | + description="Run verilator, streaming its log and UART output. " |
| 45 | + "Optionally toggle tracing when a trigger string appears on the UART." |
| 46 | + ) |
| 47 | + parser.add_argument( |
| 48 | + "--trace-trigger", |
| 49 | + metavar="STRING", |
| 50 | + default=None, |
| 51 | + help="Watch the UART for STRING and send SIGUSR1 to toggle tracing when found", |
| 52 | + ) |
| 53 | + parser.add_argument( |
| 54 | + "--verilator-bin", |
| 55 | + default=_DEFAULT_VERILATOR_BIN, |
| 56 | + help=f"Path to the verilator simulation binary (default: {_DEFAULT_VERILATOR_BIN})", |
| 57 | + ) |
| 58 | + parser.add_argument("elfs", nargs="+", help="ELF files passed to verilator with -E") |
| 59 | + return parser.parse_args() |
| 60 | + |
| 61 | + |
| 62 | +async def drain_verilator( |
| 63 | + proc: asyncio.subprocess.Process, |
| 64 | + uart_path_holder: list[Optional[str]], |
| 65 | + uart_ready: asyncio.Event, |
| 66 | + kill_cmd_holder: list[Optional[str]], |
| 67 | + kill_ready: asyncio.Event, |
| 68 | +) -> None: |
| 69 | + """Read and print verilator output, capturing UART path and kill command.""" |
| 70 | + uart_pattern = re.compile(r"\$\s*screen\s+(\S+)") |
| 71 | + kill_pattern = re.compile(r"\$\s*(kill\s+-USR1\s+\d+)") |
| 72 | + |
| 73 | + async for line_bytes in proc.stdout: |
| 74 | + line = line_bytes.decode(errors="replace") |
| 75 | + sys.stderr.write(_DIM + line + _RESET) |
| 76 | + |
| 77 | + if not uart_ready.is_set(): |
| 78 | + m = uart_pattern.search(line) |
| 79 | + if m: |
| 80 | + uart_path_holder[0] = m.group(1) |
| 81 | + print( |
| 82 | + f"{_CYAN}[verilator] UART device: {uart_path_holder[0]}{_RESET}", |
| 83 | + file=sys.stderr, |
| 84 | + ) |
| 85 | + uart_ready.set() |
| 86 | + |
| 87 | + if not kill_ready.is_set(): |
| 88 | + m = kill_pattern.search(line) |
| 89 | + if m: |
| 90 | + kill_cmd_holder[0] = m.group(1) |
| 91 | + print( |
| 92 | + f"{_CYAN}[verilator] kill cmd: {kill_cmd_holder[0]}{_RESET}", |
| 93 | + file=sys.stderr, |
| 94 | + ) |
| 95 | + kill_ready.set() |
| 96 | + |
| 97 | + # Unblock any waiters if verilator exits before printing expected lines |
| 98 | + uart_ready.set() |
| 99 | + kill_ready.set() |
| 100 | + |
| 101 | + |
| 102 | +async def stream_uart( |
| 103 | + device_path: str, |
| 104 | + trigger_string: Optional[str] = None, |
| 105 | + trigger_event: Optional[asyncio.Event] = None, |
| 106 | +) -> None: |
| 107 | + """ |
| 108 | + Read from the UART device and print its output indefinitely until cancelled. |
| 109 | + If trigger_string is given, set trigger_event when it is found on the UART, |
| 110 | + then continue streaming. |
| 111 | + """ |
| 112 | + if trigger_string: |
| 113 | + print( |
| 114 | + f"{_CYAN}[uart] Watching '{device_path}' for '{trigger_string}' ...{_RESET}", |
| 115 | + file=sys.stderr, |
| 116 | + ) |
| 117 | + else: |
| 118 | + print(f"{_CYAN}[uart] Streaming '{device_path}' ...{_RESET}", file=sys.stderr) |
| 119 | + |
| 120 | + loop = asyncio.get_running_loop() |
| 121 | + with serial.Serial(device_path, timeout=1) as ser: |
| 122 | + while True: |
| 123 | + line_bytes = await loop.run_in_executor(None, ser.readline) |
| 124 | + line = line_bytes.decode(errors="replace") |
| 125 | + if line: |
| 126 | + sys.stdout.write(_GREEN + line + _RESET) |
| 127 | + sys.stdout.flush() |
| 128 | + if ( |
| 129 | + trigger_string |
| 130 | + and trigger_event |
| 131 | + and not trigger_event.is_set() |
| 132 | + and trigger_string in line |
| 133 | + ): |
| 134 | + print(f"{_CYAN}[uart] trigger found{_RESET}", file=sys.stderr) |
| 135 | + trigger_event.set() |
| 136 | + |
| 137 | + |
| 138 | +async def main() -> None: |
| 139 | + args = parse_args() |
| 140 | + |
| 141 | + verilator_cmd = [args.verilator_bin] + [item for elf in args.elfs for item in ["-E", elf]] |
| 142 | + print(f"{_CYAN}[verilator] Starting: {' '.join(verilator_cmd)}{_RESET}", file=sys.stderr) |
| 143 | + |
| 144 | + proc = await asyncio.create_subprocess_exec( |
| 145 | + *verilator_cmd, |
| 146 | + stdout=asyncio.subprocess.PIPE, |
| 147 | + stderr=asyncio.subprocess.STDOUT, |
| 148 | + ) |
| 149 | + |
| 150 | + uart_path_holder = [None] |
| 151 | + kill_cmd_holder = [None] |
| 152 | + uart_ready = asyncio.Event() |
| 153 | + kill_ready = asyncio.Event() |
| 154 | + |
| 155 | + drain_task = asyncio.create_task( |
| 156 | + drain_verilator(proc, uart_path_holder, uart_ready, kill_cmd_holder, kill_ready) |
| 157 | + ) |
| 158 | + |
| 159 | + await uart_ready.wait() |
| 160 | + |
| 161 | + if uart_path_holder[0] is None: |
| 162 | + print( |
| 163 | + f"{_CYAN}[verilator] warning: no UART device found in log, " |
| 164 | + f"skipping UART streaming{_RESET}", |
| 165 | + file=sys.stderr, |
| 166 | + ) |
| 167 | + await proc.wait() |
| 168 | + await drain_task |
| 169 | + return |
| 170 | + |
| 171 | + trigger_event = None |
| 172 | + if args.trace_trigger: |
| 173 | + await kill_ready.wait() |
| 174 | + if kill_cmd_holder[0] is None: |
| 175 | + print( |
| 176 | + f"{_CYAN}[verilator] warning: no kill command found in log, " |
| 177 | + f"cannot trigger tracing{_RESET}", |
| 178 | + file=sys.stderr, |
| 179 | + ) |
| 180 | + else: |
| 181 | + trigger_event = asyncio.Event() |
| 182 | + |
| 183 | + uart_task = asyncio.create_task( |
| 184 | + stream_uart( |
| 185 | + uart_path_holder[0], |
| 186 | + trigger_string=args.trace_trigger, |
| 187 | + trigger_event=trigger_event, |
| 188 | + ) |
| 189 | + ) |
| 190 | + |
| 191 | + if trigger_event is not None: |
| 192 | + await trigger_event.wait() |
| 193 | + kill_parts = kill_cmd_holder[0].split() |
| 194 | + print(f"{_CYAN}[trace_trigger] Executing: {' '.join(kill_parts)}{_RESET}", file=sys.stderr) |
| 195 | + kill_proc = await asyncio.create_subprocess_exec(*kill_parts) |
| 196 | + await kill_proc.wait() |
| 197 | + |
| 198 | + await proc.wait() |
| 199 | + uart_task.cancel() |
| 200 | + |
| 201 | + await drain_task |
| 202 | + |
| 203 | + |
| 204 | +if __name__ == "__main__": |
| 205 | + asyncio.run(main()) |
0 commit comments