xiiregexbuilder
FPGA-Accelerated Regular Expression Matching Enginetui/processor.py (5914B)
1import os
2import sys
3import time
4import serial
5import serial.tools.list_ports
6from rich.console import Console
7from rich.table import Table
8from rich.panel import Panel
9from rich.layout import Layout
10from rich.syntax import Syntax
11from rich.markup import escape
12
13console = Console()
14
15# Global State
16last_match_binary = "0" * 16
17raw_hex = "0000"
18ser = None
19status_msg = "Ready"
20
21
22def get_regex_list():
23 input_file = sys.argv[1] if len(sys.argv) > 1 else "processor/regex.txt"
24 if os.path.exists(input_file):
25 with open(input_file, "r") as f:
26 return [
27 line.strip() for line in f if line.strip() and not line.startswith("#")
28 ]
29 return [
30 "ab*c",
31 "ab*",
32 "a(b|c)",
33 "b*a",
34 "c+",
35 "(a|b|c)a",
36 "ab?c",
37 "(abc)*",
38 "a.c",
39 "b(ab)c",
40 "(a|c)*",
41 "cc*b",
42 ]
43
44
45def find_fpga_port():
46 ports = serial.tools.list_ports.comports()
47 for port in ports:
48 if "COM14" in port.device:
49 return port.device
50 ports.sort(
51 key=lambda x: int(x.device.replace("COM", "") if "COM" in x.device else 0),
52 reverse=True,
53 )
54 for port in ports:
55 desc = port.description.lower()
56 if any(
57 x in desc
58 for x in ["usb", "uart", "nexys", "digilent", "cp210x", "ch340", "serial"]
59 ):
60 return port.device
61 return None
62
63
64def get_rasm():
65 rasm_file = sys.argv[2] if len(sys.argv) > 2 else "processor/build/regexes.rasm"
66 if os.path.exists(rasm_file):
67 with open(rasm_file, "r") as f:
68 return f.read()
69 return ""
70
71
72def get_regex_text():
73 input_file = sys.argv[1] if len(sys.argv) > 1 else "processor/regex.txt"
74 if os.path.exists(input_file):
75 with open(input_file, "r") as f:
76 return f.read()
77 return ""
78
79
80def update_display(layout):
81 regex_list = get_regex_list()
82 port_info = (
83 f"[bold green]CONNECTED: {ser.port}[/bold green]"
84 if ser
85 else "[bold red]DISCONNECTED[/bold red]"
86 )
87 layout["header"].update(
88 Panel(
89 f"[bold cyan]XIIRegex Dashboard[/bold cyan] | {port_info} | {status_msg}",
90 border_style="cyan",
91 )
92 )
93
94 match_view = Table(
95 expand=True, title=f"Hardware Match Results (Bits: {last_match_binary})"
96 )
97 match_view.add_column("ID", width=4, style="dim")
98 match_view.add_column("Pattern", style="cyan")
99 match_view.add_column("Status", justify="right")
100
101 bits = last_match_binary[::-1]
102 for i, name in enumerate(regex_list):
103 is_match = bits[i] == "1" if i < len(bits) else False
104 status = "[bold green]MATCH[/bold green]" if is_match else "[dim]no match[/dim]"
105 match_view.add_row(str(i), escape(name), status)
106
107 layout["matches"].update(Panel(match_view, border_style="blue"))
108 layout["code"].update(
109 Panel(
110 Syntax(get_regex_text(), "python", theme="monokai", line_numbers=True),
111 title="regex.txt (Source Patterns)",
112 border_style="green",
113 )
114 )
115 layout["footer"].update(
116 Panel(
117 "Type a string and press Enter. 'exit' to quit. Use 'make update_regex' to flash changes.",
118 border_style="dim",
119 )
120 )
121
122
123def main():
124 global ser, status_msg, last_match_binary, raw_hex
125 port = find_fpga_port()
126 if port:
127 try:
128 # Added dsrdtr and rtscts settings
129 ser = serial.Serial(port, 115200, timeout=0.1, dsrdtr=False, rtscts=False)
130 ser.dtr = True # Force DTR High
131 ser.rts = True # Force RTS High
132 status_msg = f"Connected to {port}"
133 except Exception as e:
134 status_msg = f"Connection Failed: {e}"
135 ser = None
136 else:
137 status_msg = "FPGA Not Found. Please check COM14."
138
139 layout = Layout()
140 layout.split_column(
141 Layout(name="header", size=3),
142 Layout(name="main", ratio=1),
143 Layout(name="footer", size=3),
144 )
145 layout["main"].split_row(
146 Layout(name="matches", ratio=1), Layout(name="code", ratio=1)
147 )
148
149 while True:
150 os.system("cls" if os.name == "nt" else "clear")
151 update_display(layout)
152 console.print(layout)
153
154 try:
155 test_str = console.input("\n[bold yellow]Enter test string: [/bold yellow]")
156 if test_str.lower() == "exit":
157 break
158
159 if ser and ser.is_open:
160 ser.reset_input_buffer()
161
162 # Protocol: START, CHARS, END
163 ser.write(b"\x01")
164 time.sleep(0.01)
165
166 for char in test_str:
167 ser.write(char.encode("ascii"))
168 time.sleep(0.005)
169
170 ser.write(b"\r")
171 ser.flush() # Ensure everything is sent
172
173 # Wait for response
174 start_time = time.time()
175 response = b""
176 while len(response) < 2 and (time.time() - start_time) < 1.0:
177 if ser.in_waiting > 0:
178 response += ser.read(ser.in_waiting)
179 time.sleep(0.01)
180
181 if len(response) >= 2:
182 res_bytes = response[-2:]
183 val = (res_bytes[0] << 8) | res_bytes[1]
184 raw_hex = f"{res_bytes[0]:02X}{res_bytes[1]:02X}"
185 last_match_binary = bin(val)[2:].zfill(16)
186 status_msg = (
187 f"[bold green]Match Received! (Raw: {raw_hex})[/bold green]"
188 )
189 else:
190 status_msg = "[bold red]Error: No response (Check if LED13 flickered)[/bold red]"
191
192 else:
193 status_msg = "Error: Serial port not available"
194 except KeyboardInterrupt:
195 break
196
197
198if __name__ == "__main__":
199 main()