xiiregexbuilder
FPGA-Accelerated Regular Expression Matching Enginetui/engine.py (16342B)
1#!/usr/bin/env python3
2"""
3Interactive TUI for XIIRegexBuilder
4=====================================================================
5
6Connects to the FPGA over a serial port (115200-8N1), lets the user type
7input strings, and displays a live colour-coded table showing which of
8the N regex patterns were matched by the most recent string.
9
10Usage
11─────
12 python engine.py [--port PORT] [--baud BAUD] [--regexes FILE]
13
14 --port Serial port (default: auto-detect first USB-Serial)
15 --baud Baud rate (default: 115200)
16 --regexes Path to inputs/regexes.txt for label display (optional but
17 strongly recommended - falls back to "Regex 0", "Regex 1", …)
18
19Protocol
20────────
21 Send : "<string>\n"
22 Receive: "MATCH=<bits> BYTES=<8hexdigits> HITS=<4hex,…>\r\n"
23 Send : "?\n" → receive current counter snapshot (no NFA change)
24
25Controls
26────────
27 Enter - send current input to FPGA
28 Ctrl-C / q - quit
29 Ctrl-L - clear input line
30 ? - query current counters without sending a string
31"""
32
33import argparse
34import sys
35import threading
36import time
37import re as _re
38import msvcrt
39
40try:
41 import serial
42 import serial.tools.list_ports
43except ImportError:
44 print("ERROR: pyserial not found. Run 'pip install pyserial'")
45 sys.exit(1)
46
47try:
48 from rich import box
49 from rich.console import Console
50 from rich.console import Group
51 from rich.live import Live
52 from rich.panel import Panel
53 from rich.table import Table
54 from rich.text import Text
55except ImportError:
56 print("ERROR: rich not found. Run 'pip install rich'")
57 sys.exit(1)
58
59# ─────────────────────────────────────────────────────────────────────────────
60BANNER = r"""
61██╗ ██╗██╗██╗██████╗ ███████╗ ██████╗ ███████╗██╗ ██╗██████╗ ██╗ ██╗██╗██╗ ██████╗ ███████╗██████╗
62╚██╗██╔╝██║██║██╔══██╗██╔════╝██╔════╝ ██╔════╝╚██╗██╔╝██╔══██╗██║ ██║██║██║ ██╔══██╗██╔════╝██╔══██╗
63 ╚███╔╝ ██║██║██████╔╝█████╗ ██║ ███╗█████╗ ╚███╔╝ ██████╔╝██║ ██║██║██║ ██║ ██║█████╗ ██████╔╝
64 ██╔██╗ ██║██║██╔══██╗██╔══╝ ██║ ██║██╔══╝ ██╔██╗ ██╔══██╗██║ ██║██║██║ ██║ ██║██╔══╝ ██╔══██╗
65██╔╝ ██╗██║██║██║ ██║███████╗╚██████╔╝███████╗██╔╝ ██╗██████╔╝╚██████╔╝██║███████╗██████╔╝███████╗██║ ██║
66╚═╝ ╚═╝╚═╝╚═╝╚═╝ ╚═╝╚══════╝ ╚═════╝ ╚══════╝╚═╝ ╚═╝╚═════╝ ╚═════╝ ╚═╝╚══════╝╚═════╝ ╚══════╝╚═╝ ╚═╝
67"""
68
69RESPONSE_RE = _re.compile(
70 r"MATCH=([01]+)\s+BYTES=([0-9A-Fa-f]+)\s+HITS=([0-9A-Fa-f,]+)"
71)
72
73
74# ─────────────────────────────────────────────────────────────────────────────
75def load_regexes(path: str) -> list[str]:
76 """Parse regexes.txt - skip blank lines and comments."""
77 patterns = []
78 try:
79 with open(path) as f:
80 for line in f:
81 stripped = line.strip()
82 if stripped and not stripped.startswith("#"):
83 patterns.append(stripped)
84 except FileNotFoundError:
85 pass
86 return patterns
87
88
89def auto_detect_port() -> str | None:
90 """Return the first USB-Serial port found, or None."""
91 for p in serial.tools.list_ports.comports():
92 desc = p.description.lower()
93 if "usb" in desc or "uart" in desc or "serial" in desc or "ftdi" in desc:
94 return p.device
95 # Fallback: first available port
96 ports = list(serial.tools.list_ports.comports())
97 return ports[0].device if ports else None
98
99
100# ─────────────────────────────────────────────────────────────────────────────
101class MatchState:
102 """Shared state between the UART reader thread and the TUI renderer."""
103
104 def __init__(self, num_regex: int):
105 self.lock = threading.Lock()
106 self.num_regex = num_regex
107 self.match_bits: list[int] = [0] * num_regex
108 self.byte_count: int = 0
109 self.hit_counts: list[int] = [0] * num_regex
110 self.last_string: str = ""
111 self.last_response_raw: str = ""
112 self.status_msg: str = "Waiting for first result…"
113 self.connected: bool = False
114
115 def update_from_response(self, raw: str, sent_string: str) -> bool:
116 m = RESPONSE_RE.search(raw)
117 if not m:
118 return False
119 bits_str, bytes_hex, hits_str = m.group(1), m.group(2), m.group(3)
120
121 bits_str = bits_str.zfill(self.num_regex)
122 match_bits = [int(b) for b in bits_str] # index 0 = MSB
123 # Reverse so index 0 = regex 0 (LSB of the hardware bitmask)
124 match_bits_ordered = list(reversed(match_bits))
125
126 hit_list = [int(h, 16) for h in hits_str.split(",")]
127 while len(hit_list) < self.num_regex:
128 hit_list.append(0)
129
130 with self.lock:
131 self.match_bits = match_bits_ordered[: self.num_regex]
132 self.byte_count = int(bytes_hex, 16)
133 self.hit_counts = hit_list[: self.num_regex]
134 self.last_string = sent_string
135 self.last_response_raw = raw.strip()
136 self.status_msg = "OK"
137 return True
138
139
140# ─────────────────────────────────────────────────────────────────────────────
141def reader_thread(
142 ser: serial.Serial, state: MatchState, pending: list, pending_lock: threading.Lock
143) -> None:
144 """Background thread: read lines from FPGA and update state."""
145 buf = ""
146 while True:
147 try:
148 chunk = ser.read(256).decode("ascii", errors="replace")
149 buf += chunk
150 while "\n" in buf:
151 line, buf = buf.split("\n", 1)
152 line = line.strip()
153 if not line:
154 continue
155 # Retrieve the string we sent (if any)
156 with pending_lock:
157 sent = pending.pop(0) if pending else ""
158 state.update_from_response(line, sent)
159 except serial.SerialException:
160 with state.lock:
161 state.status_msg = "Serial error - disconnected"
162 state.connected = False
163 break
164 except Exception:
165 pass
166
167
168# ─────────────────────────────────────────────────────────────────────────────
169def build_match_table(state: MatchState, patterns: list[str]) -> Table:
170 """Build the live Rich table showing match results."""
171 table = Table(
172 title="[bold cyan]Regex Match Results[/bold cyan]",
173 box=box.ROUNDED,
174 expand=True,
175 show_lines=True,
176 )
177
178 with state.lock:
179 bits = state.match_bits[:]
180 hits = state.hit_counts[:]
181 n = state.num_regex
182
183 num_cols = 2 if n > 8 else 1
184 rows_per_col = (n + num_cols - 1) // num_cols
185
186 for _ in range(num_cols):
187 table.add_column("#", style="dim", width=4, justify="right")
188 table.add_column("Pattern", style="white", min_width=20)
189 table.add_column("MATCH", justify="center", width=8)
190 table.add_column("Hits", justify="right", width=8)
191
192 for row_idx in range(rows_per_col):
193 row_data = []
194 for col_idx in range(num_cols):
195 i = col_idx * rows_per_col + row_idx
196 if i < n:
197 label = patterns[i] if i < len(patterns) else f"Regex {i}"
198 if len(label) > 30:
199 label = label[:27] + "..."
200 matched = bool(bits[i]) if i < len(bits) else False
201 hit_cnt = hits[i] if i < len(hits) else 0
202
203 if matched:
204 match_cell = Text("● MATCH", style="bold green")
205 else:
206 match_cell = Text("○ - ", style="dim red")
207
208 row_data.extend(
209 [
210 str(i),
211 f"[italic]{label}[/italic]",
212 match_cell,
213 f"[yellow]{hit_cnt}[/yellow]",
214 ]
215 )
216 else:
217 row_data.extend(["", "", "", ""])
218
219 table.add_row(*row_data)
220
221 return table
222
223
224def build_stats_panel(state: MatchState) -> Panel:
225 with state.lock:
226 bc = state.byte_count
227 last_s = state.last_string or "(none)"
228 raw = state.last_response_raw or "- "
229 status = state.status_msg
230 conn = state.connected
231
232 conn_label = (
233 "[bold green]● CONNECTED[/bold green]"
234 if conn
235 else "[bold red]● DISCONNECTED[/bold red]"
236 )
237
238 text = (
239 f"{conn_label}\n"
240 f"[bold]Last string :[/bold] [cyan]{last_s}[/cyan]\n"
241 f"[bold]Total bytes :[/bold] {bc:,}\n"
242 f"[bold]Status :[/bold] {status}\n"
243 f"[bold]Raw response:[/bold] [dim]{raw}[/dim]"
244 )
245 return Panel(text, title="[bold]Statistics[/bold]", border_style="blue")
246
247
248# ─────────────────────────────────────────────────────────────────────────────
249def run_tui(port: str, baud: int, patterns: list[str]) -> None:
250 console = Console()
251 num_regex = max(len(patterns), 1)
252 state = MatchState(num_regex)
253
254 # Open serial port
255 try:
256 ser = serial.Serial(port, baud, timeout=0.05)
257 state.connected = True
258 except serial.SerialException as e:
259 console.print(f"[red]Cannot open {port}: {e}[/red]")
260 sys.exit(1)
261
262 pending: list[str] = []
263 pending_lock = threading.Lock()
264
265 t = threading.Thread(
266 target=reader_thread, args=(ser, state, pending, pending_lock), daemon=True
267 )
268 t.start()
269
270 input_buffer = ""
271
272 def render_tui():
273 # Build layout elements
274 table = build_match_table(state, patterns)
275 stats = build_stats_panel(state)
276
277 input_panel = Panel(
278 f"[bold yellow]Enter string:[/bold yellow] [cyan]{input_buffer}[/cyan][blink]█[/blink]\n"
279 f"[dim]Enter:send, Esc/F10:quit, F1/Ctrl-R:query, Ctrl-U:clear[/dim]",
280 title="[bold]Interaction[/bold]",
281 border_style="yellow",
282 )
283
284 return Group(table, stats, input_panel)
285
286 console.print(BANNER, style="bold cyan")
287 console.print(f"[green]Connected to [bold]{port}[/bold] @ {baud} baud[/green]\n")
288
289 # MAIN INTERACTIVE LOOP
290 with Live(render_tui(), console=console, refresh_per_second=10) as live:
291 while True:
292 live.update(render_tui())
293
294 if msvcrt.kbhit():
295 char = msvcrt.getch()
296
297 # Handle extended keys (F-keys, arrows, etc)
298 if char in (b"\x00", b"\xe0"):
299 char = char + msvcrt.getch()
300
301 # 1. Quit Commands
302 if char in (
303 b"\x1b",
304 b"\x11",
305 b"\x18",
306 b"\x00D",
307 ): # Esc, Ctrl-Q, Ctrl-X, F10
308 break
309 elif char == b"\x03": # Ctrl-C
310 raise KeyboardInterrupt
311
312 # 2. Query Commands
313 elif char in (b"\x13", b"\x12", b"\x00;"): # Ctrl-S, Ctrl-R, F1
314 # Hotkey for counter query
315 try:
316 ser.write(b"?")
317 ser.flush()
318 with state.lock:
319 state.status_msg = "Sent Query (?)"
320 except serial.SerialException as exc:
321 with state.lock:
322 state.status_msg = f"Write error: {exc}"
323 continue
324
325 # 3. Input Commands
326 elif char == b"\r": # Enter
327 if not input_buffer:
328 continue
329
330 # Handle explicit "?" in the input buffer
331 if input_buffer == "?":
332 try:
333 ser.write(b"?")
334 ser.flush()
335 with state.lock:
336 state.status_msg = "Sent Query (?)"
337 except serial.SerialException as exc:
338 with state.lock:
339 state.status_msg = f"Write error: {exc}"
340 input_buffer = ""
341 continue
342
343 # Send normally to FPGA with newline
344 payload = input_buffer + "\n"
345 try:
346 with pending_lock:
347 pending.append(input_buffer)
348 ser.write(payload.encode("ascii"))
349 ser.flush()
350 except serial.SerialException as exc:
351 with state.lock:
352 state.status_msg = f"Write error: {exc}"
353
354 input_buffer = ""
355
356 elif char == b"\x08": # Backspace
357 input_buffer = input_buffer[:-1]
358 elif char in (b"\x0c", b"\x15"): # Ctrl-L, Ctrl-U (Clear Buffer)
359 input_buffer = ""
360
361 # 4. Printable Input
362 elif len(char) == 1:
363 try:
364 decoded = char.decode("ascii")
365 if decoded.isprintable():
366 input_buffer += decoded
367 except UnicodeDecodeError:
368 pass
369
370 time.sleep(0.01)
371
372 ser.close()
373 console.print("\n[bold cyan]Bye![/bold cyan]")
374
375
376# ─────────────────────────────────────────────────────────────────────────────
377def main() -> None:
378 parser = argparse.ArgumentParser(description="Interactive TUI for XIIRegexBuilder")
379 parser.add_argument(
380 "--port", default=None, help="Serial port (default: auto-detect)"
381 )
382 parser.add_argument(
383 "--baud", type=int, default=115200, help="Baud rate (default: 115200)"
384 )
385 parser.add_argument(
386 "--regexes",
387 default="inputs/regexes.txt",
388 help="Path to regexes.txt for pattern labels",
389 )
390 args = parser.parse_args()
391
392 patterns = load_regexes(args.regexes)
393
394 port = args.port
395 if port is None:
396 port = auto_detect_port()
397 if port is None:
398 print("ERROR: No serial port found. Pass --port explicitly.")
399 sys.exit(1)
400 print(f"Auto-detected port: {port}")
401
402 run_tui(port, args.baud, patterns)
403
404
405if __name__ == "__main__":
406 main()