xiiregexbuilder

FPGA-Accelerated Regular Expression Matching Engine

tui/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()