#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Control Board Monitor - v0.1 A lightweight Windows-friendly Python GUI to monitor server variables exposed via HTTP GET as "/?variable=VARNAME". - Defaults: host="localhost", port=8945 (edit globals below or use the GUI). - Refreshes automatically every minute. - Fetches variables sequentially with a configurable Requests-Per-Second (RPS) cap and automatic backoff on HTTP 429/5xx to avoid overwhelming the server. - Attempts to discover the variable list via WEBSERVER_LIST_VARIABLES_JSON, then WEBSERVER_LIST_VARIABLES. You can also paste/import a list manually in the GUI. - First version focuses on read-only monitoring. Next versions can add POST controls to set variables and turn some values into indicator widgets. No external GUI frameworks are required (uses Tkinter from the standard library). No non-stdlib HTTP client is required (uses urllib). """ import json import queue import re import threading import time import urllib.parse import urllib.request from dataclasses import dataclass, field from datetime import datetime from typing import Dict, List, Optional, Tuple try: import tkinter as tk from tkinter import ttk, messagebox, simpledialog except Exception as e: raise SystemExit("Tkinter is required to run this app.") # ===================== # Global configuration # ===================== SERVER_HOST: str = "localhost" # default host SERVER_PORT: int = 8785 # default port (as requested) REFRESH_INTERVAL_S: int = 60 # run a full sweep roughly every minute MAX_RPS: float = 5.0 # cap average requests/sec to avoid overload REQUEST_TIMEOUT_S: float = 5.0 # per-request timeout USER_AGENT: str = "ControlBoardMonitor/0.1 (+tkinter)" CONNECT_TIMEOUT_S: float = 5.0 # Backoff BACKOFF_BASE_S: float = 1.0 # initial backoff BACKOFF_MAX_S: float = 15.0 # max backoff between retries MAX_RETRIES: int = 2 # retries per variable within a cycle # ===================== # Helper utilities # ===================== def http_get(base_url: str, params: Dict[str, str]) -> Tuple[int, str, Dict[str, str]]: """Perform a GET with urllib. Returns (status, text, headers).""" url = base_url + "?" + urllib.parse.urlencode(params) req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) try: with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT_S) as resp: status = resp.getcode() body_bytes = resp.read() try: body = body_bytes.decode("utf-8", errors="replace") except Exception: body = body_bytes.decode("latin-1", errors="replace") headers = {k.lower(): v for k, v in resp.getheaders()} return status, body, headers except urllib.error.HTTPError as e: try: body = e.read().decode("utf-8", errors="replace") except Exception: body = str(e) return e.code, body, dict(e.headers or {}) except Exception as e: return 0, str(e), {} def build_base_url(host: str, port: int) -> str: scheme = "http" return f"{scheme}://{host}:{port}/" def parse_variable_names_from_html_index(html_text: str) -> List[str]: """Best-effort extraction of variable names from the simple HTML index page style.""" vars_found: List[str] = [] for m in re.finditer(r"[?&]variable=([A-Z0-9_]+)", html_text): vars_found.append(m.group(1)) # De-duplicate preserving order seen = set() uniq = [] for v in vars_found: if v not in seen: uniq.append(v) seen.add(v) return uniq def coerce_preview(value: str, maxlen: int = 80) -> str: v = value.strip().replace("\r", " ").replace("\n", " ") if len(v) > maxlen: return v[:maxlen - 1] + "…" return v def infer_status_color(value: str) -> str: """Very simple status → tag mapping to mimic control-board feel.""" val = value.strip().lower() if val in {"1", "true", "on", "open", "active", "ok", "running", "enabled"}: return "ok" if val in {"0", "false", "off", "closed", "inactive", "error", "fault", "disabled"}: return "bad" # Warn if it's empty or has obvious error keywords if val == "" or "exception" in val or "error" in val: return "bad" return "neutral" # ===================== # Data model # ===================== @dataclass class VarInfo: name: str last_value: str = "" last_updated: float = 0.0 last_status: int = 0 error: Optional[str] = None # ===================== # Poller thread # ===================== class Poller(threading.Thread): def __init__( self, host: str, port: int, variables: List[str], ui_queue: queue.Queue, refresh_interval: int, max_rps: float, stop_event: threading.Event, paused_event: threading.Event, ) -> None: super().__init__(daemon=True) self.host = host self.port = port self.variables = variables self.ui_queue = ui_queue self.refresh_interval = int(max(5, refresh_interval)) self.max_rps = max(0.5, max_rps) self.stop_event = stop_event self.paused_event = paused_event def run(self) -> None: base_url = build_base_url(self.host, self.port) min_spacing = 1.0 / self.max_rps # seconds between requests while not self.stop_event.is_set(): cycle_start = time.time() self.ui_queue.put(("cycle_start", datetime.now().strftime("%H:%M:%S"))) for name in list(self.variables): # Allow pausing mid-cycle while self.paused_event.is_set() and not self.stop_event.is_set(): time.sleep(0.1) if self.stop_event.is_set(): break attempt = 0 backoff = BACKOFF_BASE_S while attempt <= MAX_RETRIES and not self.stop_event.is_set(): attempt += 1 t0 = time.time() status, body, headers = http_get(base_url, {"variable": name}) # Success path if status == 200: self.ui_queue.put(("update", name, body, status)) break # Backoff on overload or server error if status in (429, 500, 502, 503, 504, 0): self.ui_queue.put(("error", name, status, coerce_preview(body, 200))) time.sleep(backoff) backoff = min(BACKOFF_MAX_S, backoff * 2.0) else: # Other HTTP errors: report once and stop retrying self.ui_queue.put(("error", name, status, coerce_preview(body, 200))) break # Respect RPS even with errors dt = time.time() - t0 if dt < min_spacing: time.sleep(min_spacing - dt) # Spacing between requests dt = time.time() - t0 if dt < min_spacing: time.sleep(min_spacing - dt) # Sleep to align roughly to the refresh interval cycle_dt = time.time() - cycle_start remaining = self.refresh_interval - cycle_dt if remaining > 0: for _ in range(int(remaining * 10)): if self.stop_event.is_set(): break time.sleep(0.1) self.ui_queue.put(("stopped", "")) # ===================== # GUI # ===================== class App(tk.Tk): def __init__(self) -> None: super().__init__() self.title("Control Board Monitor") self.geometry("1100x700") try: self.iconbitmap(default="") # no icon file bundled except Exception: pass # State self.vars: Dict[str, VarInfo] = {} self.stop_event = threading.Event() self.paused_event = threading.Event() self.ui_queue: queue.Queue = queue.Queue() self.poller: Optional[Poller] = None self.variables_list: List[str] = [] # Top controls controls = ttk.Frame(self, padding=10) controls.pack(side=tk.TOP, fill=tk.X) self.host_var = tk.StringVar(value=SERVER_HOST) self.port_var = tk.IntVar(value=SERVER_PORT) self.refresh_var = tk.IntVar(value=REFRESH_INTERVAL_S) self.rps_var = tk.DoubleVar(value=MAX_RPS) ttk.Label(controls, text="Host:").grid(row=0, column=0, sticky="w") ttk.Entry(controls, textvariable=self.host_var, width=16).grid(row=0, column=1, padx=(0, 12)) ttk.Label(controls, text="Port:").grid(row=0, column=2, sticky="w") ttk.Entry(controls, textvariable=self.port_var, width=8).grid(row=0, column=3, padx=(0, 12)) ttk.Label(controls, text="Refresh (s):").grid(row=0, column=4, sticky="w") ttk.Entry(controls, textvariable=self.refresh_var, width=8).grid(row=0, column=5, padx=(0, 12)) ttk.Label(controls, text="Max RPS:").grid(row=0, column=6, sticky="w") ttk.Entry(controls, textvariable=self.rps_var, width=8).grid(row=0, column=7, padx=(0, 12)) self.status_lbl = ttk.Label(controls, text="Idle", foreground="#666") self.status_lbl.grid(row=0, column=8, padx=(10, 0)) btns = ttk.Frame(controls) btns.grid(row=0, column=9, padx=10, sticky="e") self.start_btn = ttk.Button(btns, text="Start", command=self.start_polling) self.start_btn.grid(row=0, column=0, padx=4) self.pause_btn = ttk.Button(btns, text="Pause", command=self.toggle_pause, state="disabled") self.pause_btn.grid(row=0, column=1, padx=4) self.reload_btn = ttk.Button(btns, text="Reload Vars", command=self.reload_variables) self.reload_btn.grid(row=0, column=2, padx=4) self.import_btn = ttk.Button(btns, text="Import List", command=self.import_list_dialog) self.import_btn.grid(row=0, column=3, padx=4) # Filter/Search search_frame = ttk.Frame(self, padding=(10, 0, 10, 10)) search_frame.pack(side=tk.TOP, fill=tk.X) ttk.Label(search_frame, text="Filter:").pack(side=tk.LEFT) self.filter_var = tk.StringVar(value="") self.filter_var.trace_add("write", lambda *_: self.refresh_tree()) ttk.Entry(search_frame, textvariable=self.filter_var, width=50).pack(side=tk.LEFT, padx=8) # Tree (control-board style with groups) columns = ("value", "updated", "status") self.tree = ttk.Treeview(self, columns=columns, show="tree headings", height=26) self.tree.pack(side=tk.TOP, fill=tk.BOTH, expand=True, padx=10, pady=10) self.tree.heading("#0", text="Variable") self.tree.heading("value", text="Value") self.tree.heading("updated", text="Updated") self.tree.heading("status", text="HTTP") self.tree.column("#0", width=420, anchor="w") self.tree.column("value", width=420, anchor="w") self.tree.column("updated", width=120, anchor="center") self.tree.column("status", width=60, anchor="center") # Style tags for a control-board feel style = ttk.Style(self) try: style.theme_use("clam") except Exception: pass self.tree.tag_configure("ok", background="#eaffea") # soft green self.tree.tag_configure("bad", background="#ffecec") # soft red self.tree.tag_configure("neutral", background="#f5f7fb") # soft gray-blue # Footer footer = ttk.Frame(self, padding=10) footer.pack(side=tk.BOTTOM, fill=tk.X) self.cycle_lbl = ttk.Label(footer, text="Last cycle: —") self.cycle_lbl.pack(side=tk.LEFT) self.protocol("WM_DELETE_WINDOW", self.on_close) # Initial discovery self.after(200, self.reload_variables) # ------------- Variable discovery ------------- def reload_variables(self) -> None: host = self.host_var.get().strip() port = int(self.port_var.get()) base = build_base_url(host, port) discovered: List[str] = [] # Try JSON list first status, body, _ = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"}) if status == 200: try: data = json.loads(body) if isinstance(data, list) and all(isinstance(x, str) for x in data): discovered = list(dict.fromkeys(data)) # preserve order, unique except Exception: discovered = [] # Fallback: parse the human/HTML list if not discovered: status, body, _ = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES"}) if status == 200 and body: cand = parse_variable_names_from_html_index(body) if cand: discovered = cand # If still empty, offer a minimal starter set if not discovered: discovered = [ "TIME", "TIME_STAMP", "CORE_TEMP", "CORE_PRESSURE", "CORE_STATE", "COOLANT_CORE_PRESSURE", "COOLANT_CORE_FLOW_IN", "COOLANT_CORE_FLOW_OUT", "STEAM_TURBINE_0_RPM", "POWER_FROM_TURBINE_KW", "POWER_DEMAND_MW", ] # Update local state self.variables_list = discovered self.vars.clear() for v in discovered: self.vars[v] = VarInfo(name=v) self.status_lbl.configure(text=f"Loaded {len(discovered)} vars") self.refresh_tree() def import_list_dialog(self) -> None: txt = simpledialog.askstring( "Import variable names", "Paste variable names separated by spaces, commas, or newlines:", parent=self, ) if not txt: return names = re.findall(r"[A-Z0-9_]+", txt.upper()) if not names: messagebox.showwarning("Nothing found", "No variable-like tokens found.") return # Deduplicate while preserving order uniq = [] seen = set() for n in names: if n not in seen: uniq.append(n); seen.add(n) self.variables_list = uniq self.vars.clear() for v in uniq: self.vars[v] = VarInfo(name=v) self.status_lbl.configure(text=f"Imported {len(uniq)} vars") self.refresh_tree() # ------------- Poller control ------------- def start_polling(self) -> None: if self.poller and self.poller.is_alive(): messagebox.showinfo("Already running", "Polling is already active.") return if not self.variables_list: messagebox.showwarning("No variables", "Variable list is empty. Reload or import a list first.") return self.stop_event.clear() self.paused_event.clear() self.poller = Poller( host=self.host_var.get().strip(), port=int(self.port_var.get()), variables=self.variables_list, ui_queue=self.ui_queue, refresh_interval=int(self.refresh_var.get()), max_rps=float(self.rps_var.get()), stop_event=self.stop_event, paused_event=self.paused_event, ) self.poller.start() self.start_btn.configure(state="disabled") self.pause_btn.configure(state="normal", text="Pause") self.status_lbl.configure(text="Running…") self.after(100, self.drain_queue) def toggle_pause(self) -> None: if not (self.poller and self.poller.is_alive()): return if self.paused_event.is_set(): self.paused_event.clear() self.pause_btn.configure(text="Pause") self.status_lbl.configure(text="Running…") else: self.paused_event.set() self.pause_btn.configure(text="Resume") self.status_lbl.configure(text="Paused") def drain_queue(self) -> None: """Receive updates from the poller and apply to the UI safely from the main thread.""" try: while True: item = self.ui_queue.get_nowait() kind = item[0] if kind == "update": _, name, value, status = item vi = self.vars.get(name) or VarInfo(name=name) vi.last_value = value vi.last_updated = time.time() vi.last_status = status vi.error = None self.vars[name] = vi elif kind == "error": _, name, status, msg = item vi = self.vars.get(name) or VarInfo(name=name) vi.last_status = status vi.error = msg vi.last_updated = time.time() self.vars[name] = vi elif kind == "cycle_start": _, hhmmss = item self.cycle_lbl.configure(text=f"Last cycle start: {hhmmss}") elif kind == "stopped": self.status_lbl.configure(text="Stopped") self.ui_queue.task_done() except queue.Empty: pass self.refresh_tree() if self.poller and self.poller.is_alive(): self.after(250, self.drain_queue) # continue pumping def stop_polling(self) -> None: if self.poller and self.poller.is_alive(): self.stop_event.set() # give it a moment to end gracefully for _ in range(20): if not self.poller.is_alive(): break time.sleep(0.05) self.start_btn.configure(state="normal") self.pause_btn.configure(state="disabled", text="Pause") # ------------- Tree rendering ------------- def refresh_tree(self) -> None: # Clear existing self.tree.delete(*self.tree.get_children()) # Group by prefix (e.g. CORE_, COOLANT_, STEAM_) groups: Dict[str, List[VarInfo]] = {} filt = self.filter_var.get().strip().upper() for name, vi in self.vars.items(): if filt and filt not in name.upper(): # Also allow filtering by value preview if vi.last_value and filt not in vi.last_value.upper(): continue prefix = name.split("_", 1)[0] if "_" in name else "MISC" groups.setdefault(prefix, []).append(vi) # Sort groups alphabetically, vars within group by name for g in sorted(groups.keys()): parent = self.tree.insert("", "end", text=g, values=("", "", ""), open=True) for vi in sorted(groups[g], key=lambda x: x.name): updated_str = "—" if vi.last_updated: updated_str = datetime.fromtimestamp(vi.last_updated).strftime("%H:%M:%S") if vi.error: value_preview = f"[ERR] {vi.error}" else: # Compact preview of first line value_preview = coerce_preview(vi.last_value, 120) tag = infer_status_color(vi.last_value if not vi.error else "error") self.tree.insert( parent, "end", text=vi.name, values=(value_preview, updated_str, vi.last_status or "—"), tags=(tag,), ) # ------------- Shutdown ------------- def on_close(self) -> None: self.stop_polling() self.destroy() def main() -> None: app = App() app.mainloop() if __name__ == "__main__": main()