#!/usr/bin/env python3 """ gui_bridge.py — simple Tk GUI wrapper for s3_bridge.py (Windows-friendly). Features: - Select BW and S3 COM ports, baud, log directory. - Optional raw taps (BW->S3, S3->BW). - Start/Stop buttons spawn/terminate s3_bridge as a subprocess. - Live stdout view from the bridge process. Requires only the stdlib (Tkinter is bundled on Windows/Python). """ from __future__ import annotations import datetime import os import queue import subprocess import sys import threading import tkinter as tk from tkinter import filedialog, messagebox, scrolledtext, simpledialog SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) BRIDGE_PATH = os.path.join(SCRIPT_DIR, "s3-bridge", "s3_bridge.py") class BridgeGUI(tk.Tk): def __init__(self) -> None: super().__init__() self.title("S3 Bridge GUI") self.process: subprocess.Popen | None = None self.stdout_q: queue.Queue[str] = queue.Queue() self._build_widgets() self._poll_stdout() def _build_widgets(self) -> None: pad = {"padx": 6, "pady": 4} # Row 0: Ports tk.Label(self, text="BW COM:").grid(row=0, column=0, sticky="e", **pad) self.bw_var = tk.StringVar(value="COM4") tk.Entry(self, textvariable=self.bw_var, width=10).grid(row=0, column=1, sticky="w", **pad) tk.Label(self, text="S3 COM:").grid(row=0, column=2, sticky="e", **pad) self.s3_var = tk.StringVar(value="COM5") tk.Entry(self, textvariable=self.s3_var, width=10).grid(row=0, column=3, sticky="w", **pad) # Row 1: Baud tk.Label(self, text="Baud:").grid(row=1, column=0, sticky="e", **pad) self.baud_var = tk.StringVar(value="38400") tk.Entry(self, textvariable=self.baud_var, width=10).grid(row=1, column=1, sticky="w", **pad) # Row 1: Logdir chooser tk.Label(self, text="Log dir:").grid(row=1, column=2, sticky="e", **pad) self.logdir_var = tk.StringVar(value=".") tk.Entry(self, textvariable=self.logdir_var, width=24).grid(row=1, column=3, sticky="we", **pad) tk.Button(self, text="Browse", command=self._choose_dir).grid(row=1, column=4, sticky="w", **pad) # Row 2: Raw taps self.raw_bw_var = tk.StringVar(value="") self.raw_s3_var = tk.StringVar(value="") tk.Checkbutton(self, text="Save BW->S3 raw", command=self._toggle_raw_bw, onvalue="1", offvalue="").grid(row=2, column=0, sticky="w", **pad) tk.Entry(self, textvariable=self.raw_bw_var, width=28).grid(row=2, column=1, columnspan=3, sticky="we", **pad) tk.Button(self, text="...", command=lambda: self._choose_file(self.raw_bw_var, "bw")).grid(row=2, column=4, **pad) tk.Checkbutton(self, text="Save S3->BW raw", command=self._toggle_raw_s3, onvalue="1", offvalue="").grid(row=3, column=0, sticky="w", **pad) tk.Entry(self, textvariable=self.raw_s3_var, width=28).grid(row=3, column=1, columnspan=3, sticky="we", **pad) tk.Button(self, text="...", command=lambda: self._choose_file(self.raw_s3_var, "s3")).grid(row=3, column=4, **pad) # Row 4: Status + buttons self.status_var = tk.StringVar(value="Idle") tk.Label(self, textvariable=self.status_var, anchor="w").grid(row=4, column=0, columnspan=5, sticky="we", **pad) tk.Button(self, text="Start", command=self.start_bridge, width=12).grid(row=5, column=0, columnspan=2, **pad) tk.Button(self, text="Stop", command=self.stop_bridge, width=12).grid(row=5, column=2, columnspan=2, **pad) self.mark_btn = tk.Button(self, text="Add Mark", command=self.add_mark, width=12, state="disabled") self.mark_btn.grid(row=5, column=4, **pad) # Row 6: Log view self.log_view = scrolledtext.ScrolledText(self, height=20, width=90, state="disabled") self.log_view.grid(row=6, column=0, columnspan=5, sticky="nsew", **pad) # Grid weights for c in range(5): self.grid_columnconfigure(c, weight=1) self.grid_rowconfigure(6, weight=1) def _choose_dir(self) -> None: path = filedialog.askdirectory() if path: self.logdir_var.set(path) def _choose_file(self, var: tk.StringVar, direction: str) -> None: filename = filedialog.asksaveasfilename( title=f"Raw tap file for {direction}", defaultextension=".bin", filetypes=[("Binary", "*.bin"), ("All files", "*.*")] ) if filename: var.set(filename) def _toggle_raw_bw(self) -> None: if not self.raw_bw_var.get(): # default name self.raw_bw_var.set(os.path.join(self.logdir_var.get(), "raw_bw.bin")) def _toggle_raw_s3(self) -> None: if not self.raw_s3_var.get(): self.raw_s3_var.set(os.path.join(self.logdir_var.get(), "raw_s3.bin")) def start_bridge(self) -> None: if self.process and self.process.poll() is None: messagebox.showinfo("Bridge", "Bridge is already running.") return bw = self.bw_var.get().strip() s3 = self.s3_var.get().strip() baud = self.baud_var.get().strip() logdir = self.logdir_var.get().strip() or "." if not bw or not s3: messagebox.showerror("Error", "Please enter both BW and S3 COM ports.") return args = [sys.executable, BRIDGE_PATH, "--bw", bw, "--s3", s3, "--baud", baud, "--logdir", logdir] ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") raw_bw = self.raw_bw_var.get().strip() raw_s3 = self.raw_s3_var.get().strip() # If the user left the default generic name, replace with a timestamped one # so each session gets its own file. if raw_bw: if os.path.basename(raw_bw) in ("raw_bw.bin", "raw_bw"): raw_bw = os.path.join(os.path.dirname(raw_bw) or logdir, f"raw_bw_{ts}.bin") self.raw_bw_var.set(raw_bw) args += ["--raw-bw", raw_bw] if raw_s3: if os.path.basename(raw_s3) in ("raw_s3.bin", "raw_s3"): raw_s3 = os.path.join(os.path.dirname(raw_s3) or logdir, f"raw_s3_{ts}.bin") self.raw_s3_var.set(raw_s3) args += ["--raw-s3", raw_s3] try: self.process = subprocess.Popen( args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.PIPE, text=True, bufsize=1, ) except Exception as e: messagebox.showerror("Error", f"Failed to start bridge: {e}") return threading.Thread(target=self._reader_thread, daemon=True).start() self.status_var.set("Running...") self._append_log("== Bridge started ==\n") self.mark_btn.configure(state="normal") def stop_bridge(self) -> None: if self.process and self.process.poll() is None: self.process.terminate() try: self.process.wait(timeout=3) except subprocess.TimeoutExpired: self.process.kill() self.status_var.set("Stopped") self._append_log("== Bridge stopped ==\n") self.mark_btn.configure(state="disabled") def _reader_thread(self) -> None: if not self.process or not self.process.stdout: return for line in self.process.stdout: self.stdout_q.put(line) self.stdout_q.put("<>") def add_mark(self) -> None: if not self.process or not self.process.stdin or self.process.poll() is not None: return label = simpledialog.askstring("Mark", "Enter label for mark:", parent=self) if label is None or label.strip() == "": return try: # Mimic CLI behavior: send 'm' + Enter, then label + Enter self.process.stdin.write("m\n") self.process.stdin.write(label.strip() + "\n") self.process.stdin.flush() self._append_log(f"[GUI] Mark sent: {label.strip()}\n") except Exception as e: messagebox.showerror("Error", f"Failed to send mark: {e}") def _poll_stdout(self) -> None: try: while True: line = self.stdout_q.get_nowait() if line == "<>": self.status_var.set("Stopped") self.mark_btn.configure(state="disabled") break self._append_log(line) except queue.Empty: pass finally: self.after(100, self._poll_stdout) def _append_log(self, text: str) -> None: self.log_view.configure(state="normal") self.log_view.insert(tk.END, text) self.log_view.see(tk.END) self.log_view.configure(state="disabled") def main() -> int: app = BridgeGUI() app.mainloop() return 0 if __name__ == "__main__": raise SystemExit(main())