xiiregexbuilder

 1import serial
 2import time
 3import sys
 4
 5# Set port depending on OS/Setup (Eg. /dev/ttyUSB1)
 6PORT = "COM4"
 7BAUD = 921600
 8
 9try:
10    ser = serial.Serial(PORT, 115200, timeout=1, rtscts=False)
11except Exception as e:
12    print(f"ERROR: Could not open {PORT}: {e}")
13    if "Permission denied" in str(e):
14        print(f"HINT: Try running with sudo: 'sudo ./venv/bin/python pii_demo.py'")
15    print("\n--- FALLING BACK TO MOCK MODE (Simulation) ---")
16    ser = None
17
18
19def mock_redact(text):
20    import re
21
22    # Simplified simulation of the FPGA patterns
23    patterns = [
24        r"[a-zA-Z0-9.]+@[a-zA-Z0-9.]+\.[a-zA-Z]+",  # Email
25        r"\d{10}",  # Phone
26        r"\d{3}-\d{2}-\d{4}",  # SSN
27        r"\d{16}",  # CC
28    ]
29    redacted = text
30    for p in patterns:
31        for match in re.finditer(p, redacted):
32            start, end = match.span()
33            redacted = redacted[:start] + "X" * (end - start) + redacted[end:]
34    return redacted
35
36
37def stream_text(text):
38    print(f"Original Stream: {text}\n")
39
40    if ser:
41        print("Redacted Stream: ", end="", flush=True)
42        ser.reset_input_buffer()
43
44        has_started = False
45        for char in text:
46            ser.write(char.encode())
47            scrubbed = ser.read(1).decode("utf-8", errors="replace")
48
49            # Skip leading spaces from the delay buffer initialization
50            if not has_started and scrubbed != " ":
51                has_started = True
52
53            if has_started:
54                print(scrubbed, end="", flush=True)
55
56            time.sleep(0.002)  # Faster streaming for large demo files
57    else:
58        print("(MOCK) Redacted Stream: ", end="", flush=True)
59        simulated = mock_redact(text)
60        for char in simulated:
61            print(char, end="", flush=True)
62            time.sleep(0.01)
63
64    print("\n\nDone.")
65
66
67if __name__ == "__main__":
68    import os
69
70    file_path = (
71        "demo_input.txt"
72        if os.path.exists("demo_input.txt")
73        else "../pii/demo_input.txt"
74    )
75    if not os.path.exists(file_path):
76        file_path = "pii/demo_input.txt"
77    with open(file_path, "r") as f:
78        demo_text = f.read()
79
80    # Add 128 spaces to flush the FPGA's 128-char delay buffer
81    stream_text(demo_text + " " * 128)