added hax/ascii/dec readout and note appending

main
Jake 2026-02-16 15:12:35 +08:00
parent c5113f2b6c
commit 3b2d3c9502
3 changed files with 4016 additions and 57 deletions

View File

@ -1,52 +1,10 @@
# Serial Proxy (MITM) # RealRobots Serial Proxy (MITM)
A small Python app that sits between the Arduino IDE and a real COM device. It creates (or uses) a **virtual COM port** that you open in the Arduino IDE, while it talks to the **real device** on another port. All traffic is **recorded** both ways and **passed through unchanged**. A small Python app that sits between the Arduino IDE and a real COM device. It connects to a serial devices and passes all messages to and from a chosen virtual com device
## What it does
1. **CLI** — Lists connected COM ports; you choose the real device (e.g. your Arduino).
2. **Virtual port** — Tries to create a virtual COM pair with [com0com](https://sourceforge.net/projects/com0com/); if that fails, you pick one end of an existing pair.
3. **Relay** — Data from Arduino IDE → virtual port → this app → real device, and the reverse. No modification, only copying.
4. **Logging** — Every byte in both directions is logged to `serial_log.txt` (and optionally to the console) with timestamps and hex.
## Requirements ## Requirements
- Python 3.10+ - Python 3.10+
- [pyserial](https://pypi.org/project/pyserial/): `pip install -r requirements.txt` - [pyserial](https://pypi.org/project/pyserial/): `pip install -r requirements.txt`
- **Virtual COM pair** (for “create virtual port” to work): install [com0com](https://sourceforge.net/projects/com0com/) and, if you want the app to create pairs automatically, run the script (or `setupc.exe`) with **Administrator** rights so it can create port pairs. - **Virtual COM pair** https://eterlogic.com/Products.VSPE_Download.html
## Usage
1. Install dependencies:
```bash
pip install -r requirements.txt
```
2. (Optional) Install com0com and, if needed, run the app as Administrator so it can create a virtual pair.
3. Run the proxy:
```bash
python serial_proxy.py
```
4. Select the **real device** COM port (e.g. the actual Arduino).
5. If the app created a virtual pair, it will tell you which port to select in the Arduino IDE (e.g. COM252). Otherwise, select the end of the pair that **this app** will use; in the Arduino IDE you must select the **other** end of that pair.
6. Enter baud rate (default 115200) or press Enter.
7. In the Arduino IDE, choose the indicated virtual port and use Serial Monitor / upload as usual. All traffic is logged to `serial_log.txt`.
Stop with **Ctrl+C**.
## Log file
- Path: `serial_log.txt` in the same folder as `serial_proxy.py`.
- Each line has timestamp, direction (`IDE->device` or `device->IDE`), byte count, hex dump, and a raw repr of the bytes.
## Without com0com
If you dont install com0com, the app cannot create a virtual pair. Youll need another way to get a **pair** of linked virtual ports (e.g. another null-modem/virtual serial driver). Then run the app, select the **real** device port, and when asked, select the port that **this app** should open (one end of the pair). Use the **other** end of that pair in the Arduino IDE.
## Summary
| You select in the app | You select in Arduino IDE |
|------------------------|----------------------------|
| Real device (e.g. COM3) | — |
| Virtual port “for this app” (e.g. COM251) | The **other** end of the pair (e.g. COM252) |
Traffic flows: **Arduino IDE ↔ virtual pair ↔ this app ↔ real device**, with full logging both ways.

File diff suppressed because it is too large Load Diff

View File

@ -71,6 +71,37 @@ def _is_elevation_error(exc: BaseException) -> bool:
return "740" in str(exc) or "elevation" in str(exc).lower() return "740" in str(exc) or "elevation" in str(exc).lower()
# Escape sequences for ASCII display (so \r \n etc. show instead of '.')
_ASCII_ESCAPES = {
0x00: "\\0",
0x07: "\\a",
0x08: "\\b",
0x09: "\\t",
0x0A: "\\n",
0x0B: "\\v",
0x0C: "\\f",
0x0D: "\\r",
0x1B: "\\e",
}
def _format_bytes(data: bytes) -> dict[str, str]:
"""Return hex, dec, and ascii representations of data for logging."""
hex_str = " ".join(f"{b:02X}" for b in data) if len(data) <= 32 else " ".join(f"{b:02X}" for b in data[:32]) + " ..."
dec_str = " ".join(str(b) for b in data) if len(data) <= 48 else " ".join(str(b) for b in data[:48]) + " ..."
# Printable ASCII; common controls as \r \n \t etc.; other non-printable as \xNN
parts = []
for b in data:
if b in _ASCII_ESCAPES:
parts.append(_ASCII_ESCAPES[b])
elif 32 <= b <= 126:
parts.append(chr(b))
else:
parts.append(f"\\x{b:02X}")
ascii_str = "".join(parts)
return {"hex": hex_str, "dec": dec_str, "ascii": ascii_str}
def relay_loop( def relay_loop(
name: str, name: str,
read_ser: serial.Serial, read_ser: serial.Serial,
@ -78,6 +109,7 @@ def relay_loop(
direction: str, direction: str,
log_file, log_file,
log_console: bool, log_console: bool,
log_lock: threading.Lock,
): ):
"""Read from read_ser, write to write_ser, and log each chunk. Does not close ports.""" """Read from read_ser, write to write_ser, and log each chunk. Does not close ports."""
try: try:
@ -90,17 +122,19 @@ def relay_loop(
write_ser.flush() write_ser.flush()
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3] ts = datetime.now().strftime("%H:%M:%S.%f")[:-3]
line = f"[{ts}] {direction} ({len(data)} bytes)\n" line = f"[{ts}] {direction} ({len(data)} bytes)\n"
fmts = _format_bytes(data)
if log_file: if log_file:
log_file.write(line) with log_lock:
log_file.write(" hex: " + data.hex() + "\n") log_file.write(line)
try: log_file.write(f" hex: {fmts['hex']}\n")
log_file.write(" raw: " + repr(data) + "\n") log_file.write(f" dec: {fmts['dec']}\n")
except Exception: log_file.write(f" ascii: {fmts['ascii']}\n")
pass log_file.flush()
log_file.flush()
if log_console: if log_console:
print(line.strip()) print(line.strip())
print(" hex:", data.hex()) print(f" hex: {fmts['hex']}")
print(f" dec: {fmts['dec']}")
print(f" ascii: {fmts['ascii']}")
except (serial.SerialException, OSError) as e: except (serial.SerialException, OSError) as e:
if read_ser.is_open or write_ser.is_open: if read_ser.is_open or write_ser.is_open:
print(f"[{name}] {e}") print(f"[{name}] {e}")
@ -204,22 +238,41 @@ def main():
else: else:
print(f"\n>>> In Arduino IDE, select the OTHER end of the pair (not {virtual_port_ours}) <<<\n") print(f"\n>>> In Arduino IDE, select the OTHER end of the pair (not {virtual_port_ours}) <<<\n")
print("Relaying and recording. Press Ctrl+C to stop.\n") print("Relaying and recording. Press Ctrl+C to stop.")
print("Type a note and press Enter to annotate the log before testing an action.\n")
log_lock = threading.Lock()
def note_logger():
"""Read lines from stdin and write them to the log as [NOTE] annotations."""
while True:
try:
line = input()
except EOFError:
break
if line.strip() and log_file:
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3]
with log_lock:
log_file.write(f"[{ts}] NOTE: {line}\n")
log_file.flush()
print(" (note logged)")
# IDE -> device # IDE -> device
t1 = threading.Thread( t1 = threading.Thread(
target=relay_loop, target=relay_loop,
args=("IDE->device", ser_virtual, ser_real, "IDE->device", log_file, log_console), args=("IDE->device", ser_virtual, ser_real, "IDE->device", log_file, log_console, log_lock),
daemon=True, daemon=True,
) )
# device -> IDE # device -> IDE
t2 = threading.Thread( t2 = threading.Thread(
target=relay_loop, target=relay_loop,
args=("device->IDE", ser_real, ser_virtual, "device->IDE", log_file, log_console), args=("device->IDE", ser_real, ser_virtual, "device->IDE", log_file, log_console, log_lock),
daemon=True, daemon=True,
) )
t_note = threading.Thread(target=note_logger, daemon=True)
t1.start() t1.start()
t2.start() t2.start()
t_note.start()
try: try:
while t1.is_alive() or t2.is_alive(): while t1.is_alive() or t2.is_alive():
@ -234,7 +287,8 @@ def main():
except Exception: except Exception:
pass pass
if log_file: if log_file:
log_file.write("--- session ended ---\n") with log_lock:
log_file.write("--- session ended ---\n")
log_file.close() log_file.close()
print("Done.") print("Done.")