serversdown/seismo-relay
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add ACH TCP bridge, serial tap tool, and Serial Watch tab

Browse Source 
- bridges/ach_bridge.py: transparent TCP bridge that MITMs the MiniMate Plus
  call-home connection — forwards to real ACH server while logging all frames
  to raw_client/raw_server .bin files compatible with parse_capture.py;
  standalone capture mode for lab use without a real server

- bridges/serial_watch.py: RS-232 serial monitor with live S3 frame parsing;
  taps the line between MiniMate and modem (RV50/RV55); captures raw bytes,
  .log and .jsonl; --ack-ok mode auto-replies to AT commands; fixed fatal
  indentation bug in the original that silently prevented any data capture

- seismo_lab.py: new "Serial Watch" fourth tab (SerialWatchPanel) wrapping
  serial_watch.py functionality; COM port picker with refresh, baud config,
  ack-ok toggle, colour-coded live frame log (teal frames / yellow ctrl /
  blue AT), raw .bin capture auto-fed into Analyzer tab on stop

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Brian Harrison
2026-04-09 12:10:52 -04:00
committed by serversdown
parent 2db565ff9c
commit 37d32077a4
3 changed files with 1240 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seismo_lab.py
+404
View File
@@ -1071,6 +1071,398 @@ class AnalyzerPanel(tk.Frame):
# ─────────────────────────────────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# Serial Watch panel — tap the RS-232 line between device and modem
# ─────────────────────────────────────────────────────────────────────────────
try:
import serial as _serial
from serial.tools import list_ports as _list_ports
_SERIAL_OK = True
except ImportError:
_SERIAL_OK = False
from minimateplus.framing import S3FrameParser as _S3FrameParser # noqa: E402
_SW_KNOWN_SUBS = {
0xA4: "POLL_RSP", 0xA5: "BULK_WAVEFORM_RSP", 0xE0: "ADV_EVENT_RSP",
0xE1: "EVT_IDX_FIRST_RSP", 0xE3: "MONITOR_STATUS_RSP", 0xEA: "SERIAL_NUM_RSP",
0xF3: "WAVEFORM_REC_RSP", 0xF5: "WAVEFORM_HDR_RSP", 0xF7: "EVENT_INDEX_RSP",
0xF9: "UNK_06_RSP", 0xFE: "DEVICE_INFO_RSP",
0x69: "START_MON_ACK", 0x68: "STOP_MON_ACK",
}
class SerialWatchPanel(tk.Frame):
"""
Tap the RS-232 line between the MiniMate Plus and its modem (RV50/RV55).
Runs the serial reader in a background thread; surfaces parsed S3 frames
live in the log view. Writes raw_s3_<ts>.bin compatible with Analyzer.
Typical use for call-home capture:
1. Connect a USB-to-serial tap to the RS-232 line.
2. Pick that COM port here, click Start.
3. Wait for the unit to trigger / call home.
4. Click Stop, then 'Open in Analyzer' to inspect the frames.
"""
_COL_FRAME = "#4ec9b0" # teal — parsed S3 frame
_COL_CTRL = "#dcdcaa" # yellow — control-line change
_COL_AT = "#9cdcfe" # blue — AT command / ASCII noise
_COL_ERR = "#f44747" # red — error
def __init__(self, parent: tk.Widget, on_capture_ready=None, **kw):
"""
on_capture_ready(raw_s3_path: str) — called when capture stops,
so the parent can inject the file into the Analyzer.
"""
super().__init__(parent, bg=BG2, **kw)
self._on_capture_ready = on_capture_ready
self._serial: Optional[object] = None # serial.Serial instance
self._reader_thread: Optional[threading.Thread] = None
self._stop_evt = threading.Event()
self._log_q: queue.Queue[tuple[str, str]] = queue.Queue() # (text, colour)
self._raw_fh = None # open binary file handle
self._raw_path: Optional[str] = None
self._frame_count = 0
self._build()
self._poll_log_queue()
# ── build ─────────────────────────────────────────────────────────────
def _build(self) -> None:
pad = {"padx": 6, "pady": 4}
cfg = tk.Frame(self, bg=BG2)
cfg.pack(side=tk.TOP, fill=tk.X, padx=4, pady=4)
# Row 0 — port picker
tk.Label(cfg, text="COM port:", bg=BG2, fg=FG, font=MONO
).grid(row=0, column=0, sticky="e", **pad)
self._port_var = tk.StringVar()
self._port_cb = ttk.Combobox(cfg, textvariable=self._port_var,
width=12, font=MONO, state="normal")
self._port_cb.grid(row=0, column=1, sticky="w", **pad)
tk.Button(cfg, text="↺", bg=BG3, fg=FG, relief="flat", cursor="hand2",
font=MONO, command=self._refresh_ports
).grid(row=0, column=2, **pad)
tk.Label(cfg, text=" Baud:", bg=BG2, fg=FG, font=MONO
).grid(row=0, column=3, sticky="e", **pad)
self._baud_var = tk.StringVar(value="38400")
tk.Entry(cfg, textvariable=self._baud_var, width=8,
bg=BG3, fg=FG, insertbackground=FG, relief="flat", font=MONO
).grid(row=0, column=4, sticky="w", **pad)
self._ack_ok_var = tk.BooleanVar(value=False)
tk.Checkbutton(cfg, text="Ack OK to AT commands",
variable=self._ack_ok_var,
bg=BG2, fg=FG, selectcolor=BG3, activebackground=BG2,
font=MONO).grid(row=0, column=5, sticky="w", **pad)
# Row 1 — capture dir
tk.Label(cfg, text="Save to:", bg=BG2, fg=FG, font=MONO
).grid(row=1, column=0, sticky="e", **pad)
self._dir_var = tk.StringVar(
value=str(SCRIPT_DIR / "bridges" / "captures"))
tk.Entry(cfg, textvariable=self._dir_var, width=40,
bg=BG3, fg=FG, insertbackground=FG, relief="flat", font=MONO
).grid(row=1, column=1, columnspan=4, sticky="we", **pad)
tk.Button(cfg, text="Browse", bg=BG3, fg=FG, relief="flat",
cursor="hand2", font=MONO, command=self._choose_dir
).grid(row=1, column=5, **pad)
# Button row
btn_row = tk.Frame(self, bg=BG2)
btn_row.pack(side=tk.TOP, fill=tk.X, padx=4, pady=2)
self._start_btn = tk.Button(
btn_row, text="Start Watch", bg=GREEN, fg="#000000",
relief="flat", padx=12, cursor="hand2", font=MONO_B,
command=self._start)
self._start_btn.pack(side=tk.LEFT, padx=6)
self._stop_btn = tk.Button(
btn_row, text="Stop", bg=BG3, fg=FG,
relief="flat", padx=12, cursor="hand2", font=MONO,
command=self._stop, state="disabled")
self._stop_btn.pack(side=tk.LEFT, padx=4)
self._analyzer_btn = tk.Button(
btn_row, text="Open in Analyzer", bg=BG3, fg=FG,
relief="flat", padx=10, cursor="hand2", font=MONO,
command=self._send_to_analyzer, state="disabled")
self._analyzer_btn.pack(side=tk.LEFT, padx=4)
tk.Button(btn_row, text="Clear", bg=BG3, fg=FG,
relief="flat", padx=8, cursor="hand2", font=MONO,
command=self._clear_log).pack(side=tk.LEFT, padx=4)
self._status_var = tk.StringVar(value="Idle")
tk.Label(btn_row, textvariable=self._status_var,
bg=BG2, fg=FG_DIM, font=MONO).pack(side=tk.LEFT, padx=10)
# Log view
self._log = scrolledtext.ScrolledText(
self, height=24, font=MONO_SM,
bg=BG, fg=FG, insertbackground=FG,
relief="flat", state="disabled",
)
self._log.pack(fill=tk.BOTH, expand=True, padx=4, pady=4)
self._log.tag_config("frame", foreground=self._COL_FRAME)
self._log.tag_config("ctrl", foreground=self._COL_CTRL)
self._log.tag_config("at", foreground=self._COL_AT)
self._log.tag_config("err", foreground=self._COL_ERR)
self._log.tag_config("dim", foreground=FG_DIM)
# Populate ports on first load
self._refresh_ports()
# ── port helpers ──────────────────────────────────────────────────────
def _refresh_ports(self) -> None:
if not _SERIAL_OK:
self._port_cb["values"] = ["(pyserial not installed)"]
return
ports = [p.device for p in _list_ports.comports()]
self._port_cb["values"] = ports
if ports and not self._port_var.get():
self._port_var.set(ports[0])
def _choose_dir(self) -> None:
d = filedialog.askdirectory(initialdir=self._dir_var.get())
if d:
self._dir_var.set(d)
# ── start / stop ──────────────────────────────────────────────────────
def _start(self) -> None:
if not _SERIAL_OK:
messagebox.showerror(
"pyserial missing",
"Install pyserial first:\n pip install pyserial")
return
port = self._port_var.get().strip()
if not port or "not installed" in port:
messagebox.showerror("Error", "Select a valid COM port first.")
return
try:
baud = int(self._baud_var.get().strip())
except ValueError:
messagebox.showerror("Error", "Invalid baud rate.")
return
# Open output files
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
out_dir = Path(self._dir_var.get()) / f"serial_{ts}"
out_dir.mkdir(parents=True, exist_ok=True)
self._raw_path = str(out_dir / f"raw_s3_{ts}.bin")
try:
self._raw_fh = open(self._raw_path, "wb")
except OSError as exc:
messagebox.showerror("Error", f"Cannot open capture file:\n{exc}")
return
# Open serial port
try:
ser = _serial.Serial(
port=port, baudrate=baud,
bytesize=8, parity=_serial.PARITY_NONE,
stopbits=_serial.STOPBITS_ONE,
timeout=0.05, write_timeout=0,
)
ser.setDTR(True)
ser.setRTS(True)
except Exception as exc:
self._raw_fh.close()
self._raw_fh = None
messagebox.showerror("Error", f"Cannot open {port}:\n{exc}")
return
self._serial = ser
self._stop_evt.clear()
self._frame_count = 0
self._analyzer_btn.configure(state="disabled")
self._reader_thread = threading.Thread(
target=self._reader_loop,
args=(ser, baud),
daemon=True,
)
self._reader_thread.start()
self._status_var.set(f"Watching {port} @ {baud}")
self._start_btn.configure(state="disabled")
self._stop_btn.configure(state="normal", bg=RED)
self._append(f"── Serial watch started {port} @ {baud} [{ts}] ──\n", "dim")
self._append(f" Capture: {self._raw_path}\n", "dim")
self._append(" Waiting for data…\n\n", "dim")
def _stop(self) -> None:
self._stop_evt.set()
if self._serial:
try:
self._serial.close()
except Exception:
pass
self._serial = None
if self._raw_fh:
self._raw_fh.close()
self._raw_fh = None
self._status_var.set("Stopped")
self._start_btn.configure(state="normal")
self._stop_btn.configure(state="disabled", bg=BG3)
if self._raw_path and Path(self._raw_path).exists():
self._analyzer_btn.configure(state="normal")
self._append("\n── Watch stopped ──\n", "dim")
# ── reader thread ─────────────────────────────────────────────────────
def _reader_loop(self, ser, baud: int) -> None:
parser = _S3FrameParser()
rx_buf = bytearray()
ack_ok = self._ack_ok_var.get()
# Monitor control lines in a sub-thread
ctrl_stop = threading.Event()
ctrl_thread = threading.Thread(
target=self._ctrl_loop, args=(ser, ctrl_stop), daemon=True)
ctrl_thread.start()
try:
while not self._stop_evt.is_set():
try:
data = ser.read(4096)
except Exception as exc:
self._log_q.put((f"Read error: {exc}\n", "err"))
break
if not data:
continue
# Save raw bytes
if self._raw_fh:
try:
self._raw_fh.write(data)
self._raw_fh.flush()
except Exception:
pass
# Parse S3 frames
for byte in data:
result = parser.feed(bytes([byte]))
if result:
frames = result if isinstance(result, list) else [result]
for f in frames:
self._frame_count += 1
name = _SW_KNOWN_SUBS.get(f.sub, f"UNK_0x{f.sub:02X}")
chk = "✓" if f.checksum_valid else "✗ BAD_CHK"
peek = f.data[:32].hex() + ("…" if len(f.data) > 32 else "")
msg = (
f"[{self._frame_count:04d}] "
f"SUB=0x{f.sub:02X} ({name:<22}) "
f"page=0x{f.page_key:04X} "
f"data={len(f.data):4d}B {chk}\n"
f" {peek}\n"
)
self._log_q.put((msg, "frame"))
# AT command handling for --ack-ok mode
if ack_ok:
rx_buf.extend(data)
while b"\r" in rx_buf or b"\n" in rx_buf:
for sep in (b"\r", b"\n"):
idx = rx_buf.find(sep)
if idx != -1:
line_bytes = bytes(rx_buf[:idx])
del rx_buf[:idx + 1]
break
else:
break
line_str = line_bytes.decode("latin1", errors="ignore").strip()
if line_str.upper().startswith("AT"):
self._log_q.put((f"AT: {line_str!r}\n", "at"))
if not line_str.upper().startswith("ATDT"):
try:
ser.write(b"\r\nOK\r\n")
ser.flush()
self._log_q.put((f" → OK\n", "at"))
except Exception:
pass
finally:
ctrl_stop.set()
ctrl_thread.join(timeout=0.5)
# Signal the main thread that the reader ended naturally
if not self._stop_evt.is_set():
self._log_q.put(("<<done>>", ""))
def _ctrl_loop(self, ser, stop: threading.Event) -> None:
prev = {}
try:
prev = dict(CTS=ser.cts, DSR=ser.dsr, DCD=ser.cd)
try:
prev["RI"] = ser.ri
except Exception:
prev["RI"] = None
except Exception:
return
while not stop.is_set():
try:
cur = dict(CTS=ser.cts, DSR=ser.dsr, DCD=ser.cd, RI=None)
try:
cur["RI"] = ser.ri
except Exception:
pass
for name, val in cur.items():
if val != prev.get(name):
self._log_q.put((f"CTRL {name} → {val}\n", "ctrl"))
prev[name] = val
except Exception:
break
stop.wait(0.2)
# ── log view ──────────────────────────────────────────────────────────
def _poll_log_queue(self) -> None:
try:
while True:
text, tag = self._log_q.get_nowait()
if text == "<<done>>":
self._stop()
break
self._append(text, tag)
except queue.Empty:
pass
finally:
self.after(80, self._poll_log_queue)
def _append(self, text: str, tag: str = "") -> None:
self._log.configure(state="normal")
if tag:
self._log.insert(tk.END, text, tag)
else:
self._log.insert(tk.END, text)
self._log.see(tk.END)
self._log.configure(state="disabled")
def _clear_log(self) -> None:
self._log.configure(state="normal")
self._log.delete("1.0", tk.END)
self._log.configure(state="disabled")
# ── send to analyzer ──────────────────────────────────────────────────
def _send_to_analyzer(self) -> None:
if self._raw_path and self._on_capture_ready:
self._on_capture_ready(self._raw_path)
# Console panel (tk.Frame — lives inside a notebook tab)
# ─────────────────────────────────────────────────────────────────────────────
@@ -1504,6 +1896,12 @@ class SeismoLab(tk.Tk):
)
nb.add(self._console_panel, text=" Console ")
self._serial_watch_panel = SerialWatchPanel(
nb,
on_capture_ready=self._on_serial_capture_ready,
)
nb.add(self._serial_watch_panel, text=" Serial Watch ")
self._nb = nb
self.protocol("WM_DELETE_WINDOW", self._on_close)
@@ -1522,8 +1920,14 @@ class SeismoLab(tk.Tk):
self._analyzer_panel.s3_var.set(raw_s3_path)
self._nb.select(1)
def _on_serial_capture_ready(self, raw_s3_path: str) -> None:
"""Serial Watch capture finished → inject into Analyzer and switch tab."""
self._analyzer_panel.s3_var.set(raw_s3_path)
self._nb.select(1)
def _on_close(self) -> None:
self._bridge_panel.stop_bridge()
self._serial_watch_panel._stop()
self.destroy()