215 lines
8.2 KiB
Python
215 lines
8.2 KiB
Python
#!/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 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]
|
|
|
|
raw_bw = self.raw_bw_var.get().strip()
|
|
raw_s3 = self.raw_s3_var.get().strip()
|
|
if raw_bw:
|
|
args += ["--raw-bw", raw_bw]
|
|
if 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("<<process-exit>>")
|
|
|
|
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 == "<<process-exit>>":
|
|
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())
|