diff --git a/nucleares_monitor/control_board_monitor.py b/nucleares_monitor/control_board_monitor.py
index 06c4269..e54cc26 100644
--- a/nucleares_monitor/control_board_monitor.py
+++ b/nucleares_monitor/control_board_monitor.py
@@ -1,12 +1,25 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
-Control Board Monitor - v0.2
+Control Board Monitor - v0.3 (proposal build)
+Features added per request:
+ 1) Variable Selector dialog: after discovery, shows all variables with checkboxes.
+ Pre-checks currently-monitored set; user can add/remove before starting.
+ 2) Functions (POST) Runner + Scheduler:
+ - Discovers writable variables/functions from the HTML POST section and optional JSON.
+ - "Run Once" with a value parameter.
+ - "Schedule" to run at an interval; can pause/resume/remove at any time.
+ 3) Thresholds & Alarms per monitored value:
+ - Configure LOW/HIGH and optional DEAD_LOW/EXTREME_HIGH per variable.
+ - Detect state transitions (into LOW/HIGH/DEAD_LOW/EXTREME_HIGH from previous state).
+ - Optional alarm sound on entering a state.
+ - Optional function to invoke on entering a state (select any discovered function + value).
+ 4) Tree-structured GUI:
+ - Top: Variable monitor tree (grouped by prefix, as before).
+ - Bottom: Actions/Scheduler tree.
+ - Menu/Buttons to open Variable Selector and Thresholds dialogs.
-- Robust discovery: merges WEBSERVER_LIST_VARIABLES_JSON + WEBSERVER_LIST_VARIABLES (HTML).
-- Better HTML parsing: extracts from hrefs and plain "variable=NAME" tokens; filters placeholders.
-- New: "Load HTML" button to parse a local HTML file (e.g., the documentation page you shared).
-- Same lightweight Tkinter GUI, request throttling, and backoff.
+Stdlib-only: Tkinter, threading, urllib. Tested on Windows with Python 3.10+.
"""
import json
@@ -16,7 +29,7 @@ import threading
import time
import urllib.parse
import urllib.request
-from dataclasses import dataclass
+from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Tuple
@@ -26,6 +39,23 @@ try:
except Exception as e:
raise SystemExit("Tkinter is required to run this app.")
+# ======= Windows beep support =======
+def play_alarm_tone(level: str = "LOW") -> None:
+ """Best-effort alarm sound; uses winsound on Windows or tk.bell() otherwise."""
+ try:
+ import winsound # type: ignore
+ freq = {"DEAD_LOW": 300, "LOW": 550, "HIGH": 800, "EXTREME_HIGH": 1000}.get(level, 600)
+ dur = 350 if level != "EXTREME_HIGH" else 600
+ winsound.Beep(freq, dur)
+ except Exception:
+ try:
+ # Fallback to Tk bell if available (non-blocking/chip)
+ root = tk._default_root
+ if root is not None:
+ root.bell()
+ except Exception:
+ pass
+
# =====================
# Global configuration
@@ -35,7 +65,7 @@ SERVER_PORT: int = 8785
REFRESH_INTERVAL_S: int = 1
MAX_RPS: float = 5.0
REQUEST_TIMEOUT_S: float = 5.0
-USER_AGENT: str = "ControlBoardMonitor/0.2 (+tkinter)"
+USER_AGENT: str = "ControlBoardMonitor/0.3 (+tkinter)"
CONNECT_TIMEOUT_S: float = 5.0
BACKOFF_BASE_S: float = 1.0
@@ -69,23 +99,39 @@ def http_get(base_url: str, params: Dict[str, str]) -> Tuple[int, str, Dict[str,
return 0, str(e), {}
+def http_post(base_url: str, data: Dict[str, str]) -> Tuple[int, str, Dict[str, str]]:
+ url = base_url
+ post_data = urllib.parse.urlencode(data).encode("utf-8")
+ req = urllib.request.Request(url, data=post_data, 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:
return f"http://{host}:{port}/"
def parse_variable_names_from_html_index(html_text: str) -> List[str]:
- """Extract variable names from an index-like HTML page.
-
- Strategy:
- 1) Pull from all hrefs' query strings: ?variable=NAME (case-insensitive).
- 2) Also scan plain text "variable=NAME" tokens.
- 3) Normalize, de-dup, keep original order.
- 4) Filter placeholders like "VARNAME".
- """
+ """Extract readable variables (GET) from an index-like HTML page."""
import urllib.parse
tokens: List[str] = []
- # HREFs with any query string containing "variable"
for m in re.finditer(r'href\s*=\s*["\']([^"\']+)["\']', html_text, flags=re.I):
href = m.group(1)
parsed = urllib.parse.urlparse(href)
@@ -93,31 +139,37 @@ def parse_variable_names_from_html_index(html_text: str) -> List[str]:
for v in qs.get("variable", []):
tokens.append(v)
- # Plain text occurrences like variable=FOO (allow quotes, hyphens, periods)
for m in re.finditer(r'variable\s*=\s*["\']?([A-Za-z0-9_.:-]+)', html_text, flags=re.I):
tokens.append(m.group(1))
- # De-duplicate while preserving order; uppercase for consistency
- seen = set()
- uniq: List[str] = []
+ seen = set(); uniq: List[str] = []
for v in tokens:
vv = v.strip().strip(".,;)]}").strip().upper()
if vv and vv not in seen:
- seen.add(vv)
- uniq.append(vv)
+ seen.add(vv); uniq.append(vv)
- # Filter obvious placeholders
placeholders = {"VARNAME", "VARIABLE", "NAME"}
- uniq = [v for v in uniq if v not in placeholders]
+ return [v for v in uniq if v not in placeholders]
- return uniq
+
+def parse_function_names_from_html_index(html_text: str) -> List[str]:
+ """Extract writable variables / functions from the POST section of the HTML."""
+ start = html_text.find("==== POST ====")
+ if start == -1:
+ return []
+ post_html = html_text[start:]
+ names = re.findall(r"([A-Z0-9_]+)", post_html)
+ seen = set(); uniq: List[str] = []
+ for n in names:
+ if n not in seen:
+ seen.add(n); uniq.append(n)
+ placeholders = {"VARNAME", "VARIABLE", "NAME"}
+ return [v for v in uniq if v not in placeholders]
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
+ return v if len(v) <= maxlen else (v[:maxlen - 1] + "…")
def infer_status_color(value: str) -> str:
@@ -131,9 +183,40 @@ def infer_status_color(value: str) -> str:
return "neutral"
+def parse_first_float(value: str) -> Optional[float]:
+ m = re.search(r"[-+]?\d+(?:\.\d+)?", value.replace(",", "."))
+ if m:
+ try:
+ return float(m.group(0))
+ except Exception:
+ return None
+ return None
+
+
# =====================
# Data model
# =====================
+@dataclass
+class Thresholds:
+ dead_low: Optional[float] = None
+ low: Optional[float] = None
+ high: Optional[float] = None
+ extreme_high: Optional[float] = None
+ alarm_dead_low: bool = False
+ alarm_low: bool = False
+ alarm_high: bool = False
+ alarm_extreme_high: bool = True
+ # Optional function triggers on entering each state
+ action_dead_low: Optional[str] = None
+ value_dead_low: str = "1"
+ action_low: Optional[str] = None
+ value_low: str = "1"
+ action_high: Optional[str] = None
+ value_high: str = "1"
+ action_extreme_high: Optional[str] = None
+ value_extreme_high: str = "1"
+
+
@dataclass
class VarInfo:
name: str
@@ -141,10 +224,737 @@ class VarInfo:
last_updated: float = 0.0
last_status: int = 0
error: Optional[str] = None
+ thresholds: Thresholds = field(default_factory=Thresholds)
+ last_state: str = "UNKNOWN" # one of UNKNOWN/NORMAL/DEAD_LOW/LOW/HIGH/EXTREME_HIGH
+
+
+@dataclass
+class ActionTask:
+ name: str # function/variable name to POST
+ value: str # value to send
+ interval_s: int # 0 for one-shot
+ enabled: bool = True
+ last_run: float = 0.0
+ next_run: float = 0.0
+ task_id: int = 0 # assigned by scheduler
# =====================
-# Poller thread
+# Scheduler thread
+# =====================
+class ActionScheduler(threading.Thread):
+ def __init__(self, get_base_url_cb):
+ super().__init__(daemon=True)
+ self.get_base_url_cb = get_base_url_cb
+ self.tasks: Dict[int, ActionTask] = {}
+ self._lock = threading.Lock()
+ self._stop = threading.Event()
+ self._counter = 1
+
+ def add_task(self, task: ActionTask) -> int:
+ with self._lock:
+ tid = self._counter
+ self._counter += 1
+ task.task_id = tid
+ now = time.time()
+ task.next_run = now if task.interval_s == 0 else now + task.interval_s
+ self.tasks[tid] = task
+ return tid
+
+ def remove_task(self, task_id: int) -> None:
+ with self._lock:
+ self.tasks.pop(task_id, None)
+
+ def list_tasks(self) -> List[ActionTask]:
+ with self._lock:
+ return list(self.tasks.values())
+
+ def set_enabled(self, task_id: int, enabled: bool) -> None:
+ with self._lock:
+ t = self.tasks.get(task_id)
+ if t:
+ t.enabled = enabled
+ if enabled and t.interval_s > 0:
+ t.next_run = time.time() + t.interval_s
+
+ def run_once(self, name: str, value: str) -> Tuple[int, str, Dict[str, str]]:
+ base = self.get_base_url_cb()
+ status, body, headers = http_post(base, {"variable": name, "value": value})
+ return status, body, headers
+
+ def run(self) -> None:
+ while not self._stop.is_set():
+ to_run: List[ActionTask] = []
+ now = time.time()
+ with self._lock:
+ for t in self.tasks.values():
+ if t.enabled and t.next_run <= now:
+ to_run.append(t)
+ # Execute outside lock
+ for t in to_run:
+ try:
+ status, body, _ = self.run_once(t.name, t.value)
+ t.last_run = now
+ if t.interval_s > 0:
+ t.next_run = now + t.interval_s
+ else:
+ # one-shot: remove after run
+ self.remove_task(t.task_id)
+ except Exception:
+ # swallow to keep scheduler alive
+ if t.interval_s > 0:
+ t.next_run = now + t.interval_s
+ else:
+ self.remove_task(t.task_id)
+ time.sleep(0.25)
+
+ def stop(self) -> None:
+ self._stop.set()
+
+
+# =====================
+# GUI
+# =====================
+class App(tk.Tk):
+ def __init__(self) -> None:
+ super().__init__()
+ self.title("Control Board Monitor — v0.3")
+ self.geometry("1250x800")
+ try:
+ self.iconbitmap(default="")
+ 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] = []
+ self.functions_list: List[str] = []
+
+ # Scheduler
+ self.scheduler = ActionScheduler(lambda: build_base_url(self.host_var.get().strip(), int(self.port_var.get())))
+ self.scheduler.start()
+
+ # ---- 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, 10))
+ 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, 10))
+ 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, 10))
+ 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, 10))
+
+ 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")
+ ttk.Button(btns, text="Select Vars…", command=self.open_selector).grid(row=0, column=0, padx=4)
+ ttk.Button(btns, text="Reload (disc.)", command=self.reload_discovery).grid(row=0, column=1, padx=4)
+ self.start_btn = ttk.Button(btns, text="Start", command=self.start_polling)
+ self.start_btn.grid(row=0, column=2, padx=4)
+ self.pause_btn = ttk.Button(btns, text="Pause", command=self.toggle_pause, state="disabled")
+ self.pause_btn.grid(row=0, column=3, padx=4)
+
+ # ---- Split panes ----
+ main = ttk.Panedwindow(self, orient=tk.VERTICAL)
+ main.pack(side=tk.TOP, fill=tk.BOTH, expand=True, padx=10, pady=(0,10))
+
+ # Filter/Search
+ search_frame = ttk.Frame(self, padding=(0, 6, 0, 6))
+ 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)
+
+ # Top: Variable monitor tree
+ top_frame = ttk.Frame(main)
+ search_frame.pack(in_=top_frame, side=tk.TOP, fill=tk.X)
+ columns = ("value", "updated", "status")
+ self.tree = ttk.Treeview(top_frame, columns=columns, show="tree headings", height=18)
+ self.tree.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
+
+ 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=480, anchor="w")
+ self.tree.column("value", width=480, anchor="w")
+ self.tree.column("updated", width=120, anchor="center")
+ self.tree.column("status", width=60, anchor="center")
+
+ style = ttk.Style(self)
+ try:
+ style.theme_use("clam")
+ except Exception:
+ pass
+ self.tree.tag_configure("ok", background="#eaffea")
+ self.tree.tag_configure("bad", background="#ffecec")
+ self.tree.tag_configure("neutral", background="#f5f7fb")
+
+ # Context menu to configure thresholds
+ self.tree_menu = tk.Menu(self, tearoff=False)
+ self.tree_menu.add_command(label="Set thresholds & alarms…", command=self.open_thresholds_dialog)
+ self.tree.bind("", self.on_tree_right_click)
+
+ # Bottom: Actions/Scheduler tree + controls
+ bottom = ttk.Frame(main)
+
+ act_controls = ttk.Frame(bottom, padding=(0,6,0,6))
+ ttk.Label(act_controls, text="Function:").pack(side=tk.LEFT)
+ self.func_name_var = tk.StringVar(value="")
+ self.func_combo = ttk.Combobox(act_controls, textvariable=self.func_name_var, width=40, state="readonly")
+ self.func_combo.pack(side=tk.LEFT, padx=6)
+ ttk.Label(act_controls, text="Value:").pack(side=tk.LEFT)
+ self.func_value_var = tk.StringVar(value="1")
+ ttk.Entry(act_controls, textvariable=self.func_value_var, width=10).pack(side=tk.LEFT, padx=6)
+ ttk.Button(act_controls, text="Run Once", command=self.run_func_once).pack(side=tk.LEFT, padx=4)
+ ttk.Button(act_controls, text="Schedule…", command=self.add_schedule_dialog).pack(side=tk.LEFT, padx=4)
+
+ columns2 = ("value", "mode", "interval", "next", "enabled")
+ self.actions_tree = ttk.Treeview(bottom, columns=columns2, show="headings", height=10)
+ for col, hdr, w in zip(columns2, ["Value", "Mode", "Interval(s)", "Next Run", "Enabled"], [160, 100, 100, 180, 80]):
+ self.actions_tree.heading(col, text=hdr)
+ self.actions_tree.column(col, width=w, anchor="center")
+
+ self.actions_tree.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
+ act_controls.pack(side=tk.TOP, fill=tk.X)
+
+ # Actions tree context menu
+ self.actions_menu = tk.Menu(self, tearoff=False)
+ self.actions_menu.add_command(label="Toggle Enable/Disable", command=self.toggle_selected_task)
+ self.actions_menu.add_command(label="Run Now", command=self.run_selected_task_now)
+ self.actions_menu.add_command(label="Remove", command=self.remove_selected_task)
+ self.actions_tree.bind("", self.on_actions_right_click)
+
+ main.add(top_frame, weight=3)
+ main.add(bottom, weight=2)
+
+ # 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 (does not auto-start polling)
+ self.after(200, self.reload_discovery)
+
+ # ------------- Discovery & selector -------------
+ def reload_discovery(self) -> None:
+ host = self.host_var.get().strip()
+ port = int(self.port_var.get())
+ base = build_base_url(host, port)
+
+ discovered_vars: List[str] = []
+ discovered_funcs: List[str] = []
+
+ # JSON for variables
+ status, body, _ = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"})
+ if status == 200:
+ try:
+ data = json.loads(body)
+ if isinstance(data, list):
+ for x in data:
+ if isinstance(x, str):
+ v = x.strip().upper()
+ if v and v not in discovered_vars:
+ discovered_vars.append(v)
+ except Exception:
+ pass
+
+ # HTML
+ status_h, body_h, _h = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES"})
+ if status_h == 200 and body_h:
+ cand = parse_variable_names_from_html_index(body_h)
+ for v in cand:
+ if v not in discovered_vars:
+ discovered_vars.append(v)
+ # Functions from POST section
+ f_cand = parse_function_names_from_html_index(body_h)
+ for f in f_cand:
+ if f not in discovered_funcs:
+ discovered_funcs.append(f)
+
+ # Optional JSON for functions
+ status_f, body_f, _f = http_get(base, {"variable": "WEBSERVER_LIST_FUNCTIONS_JSON"})
+ if status_f == 200:
+ try:
+ dataf = json.loads(body_f)
+ if isinstance(dataf, list):
+ for x in dataf:
+ if isinstance(x, str):
+ fx = x.strip().upper()
+ if fx and fx not in discovered_funcs:
+ discovered_funcs.append(fx)
+ except Exception:
+ pass
+
+ # If empty, keep previous; else present selector dialog
+ if discovered_vars:
+ self.show_selector_dialog(discovered_vars)
+ if discovered_funcs:
+ self.functions_list = discovered_funcs
+ self.func_combo["values"] = self.functions_list
+
+ self.status_lbl.configure(text=f"Discovered vars: {len(discovered_vars)} | functions: {len(self.functions_list)}")
+
+ def open_selector(self) -> None:
+ # Manual open (uses last discovered set = union of current + variables_list)
+ union = sorted(set(self.variables_list) | set(self.vars.keys()))
+ if not union:
+ messagebox.showinfo("No variables", "No discovered variables yet. Use Reload (disc.) first.")
+ return
+ self.show_selector_dialog(union)
+
+ def show_selector_dialog(self, discovered: List[str]) -> None:
+ dlg = tk.Toplevel(self)
+ dlg.title("Select Variables to Monitor")
+ dlg.geometry("600x520")
+ dlg.transient(self)
+ dlg.grab_set()
+
+ # Filter entry
+ top = ttk.Frame(dlg, padding=8)
+ top.pack(side=tk.TOP, fill=tk.X)
+ ttk.Label(top, text="Filter:").pack(side=tk.LEFT)
+ filt_var = tk.StringVar(value="")
+ ttk.Entry(top, textvariable=filt_var, width=40).pack(side=tk.LEFT, padx=6)
+ sel_all = ttk.Button(top, text="Select All")
+ sel_none = ttk.Button(top, text="Clear")
+ sel_all.pack(side=tk.LEFT, padx=4); sel_none.pack(side=tk.LEFT, padx=4)
+
+ # Scrollable checkbox list
+ frame = ttk.Frame(dlg)
+ frame.pack(side=tk.TOP, fill=tk.BOTH, expand=True, padx=8, pady=8)
+ canvas = tk.Canvas(frame, borderwidth=0)
+ vsb = ttk.Scrollbar(frame, orient="vertical", command=canvas.yview)
+ inner = ttk.Frame(canvas)
+ inner.bind("", lambda e: canvas.configure(scrollregion=canvas.bbox("all")))
+ canvas.create_window((0,0), window=inner, anchor="nw")
+ canvas.configure(yscrollcommand=vsb.set)
+ canvas.pack(side="left", fill="both", expand=True); vsb.pack(side="right", fill="y")
+
+ # Build vars
+ current_set = set(self.variables_list or list(self.vars.keys()))
+ rows: List[Tuple[str, tk.BooleanVar, ttk.Checkbutton]] = []
+ def rebuild():
+ for child in inner.winfo_children():
+ child.destroy()
+ rows.clear()
+ f = filt_var.get().strip().upper()
+ for name in discovered:
+ if f and f not in name.upper():
+ continue
+ var = tk.BooleanVar(value=name in current_set)
+ cb = ttk.Checkbutton(inner, text=name, variable=var)
+ cb.pack(anchor="w", pady=2, padx=4)
+ rows.append((name, var, cb))
+ rebuild()
+
+ # Wire filter/select-all
+ def on_filter(*_):
+ rebuild()
+ filt_var.trace_add("write", on_filter)
+
+ sel_all.configure(command=lambda: [v.set(True) for _, v, _ in rows])
+ sel_none.configure(command=lambda: [v.set(False) for _, v, _ in rows])
+
+ # Buttons
+ bot = ttk.Frame(dlg, padding=8)
+ bot.pack(side=tk.BOTTOM, fill=tk.X)
+ def apply_and_close():
+ selected = [name for name, var, _ in rows if var.get()]
+ if not selected:
+ if not messagebox.askyesno("Empty selection", "No variables selected. Continue?"):
+ return
+ self.variables_list = selected
+ # reset var info dictionary to selected
+ self.vars = {v: self.vars.get(v, VarInfo(name=v)) for v in self.variables_list}
+ self.refresh_tree()
+ dlg.destroy()
+ ttk.Button(bot, text="Apply", command=apply_and_close).pack(side=tk.RIGHT, padx=4)
+ ttk.Button(bot, text="Cancel", command=dlg.destroy).pack(side=tk.RIGHT, padx=4)
+
+ # ------------- Thresholds dialog -------------
+ def on_tree_right_click(self, event):
+ try:
+ iid = self.tree.identify_row(event.y)
+ if iid:
+ self.tree.selection_set(iid)
+ finally:
+ self.tree_menu.tk_popup(event.x_root, event.y_root)
+
+ def open_thresholds_dialog(self):
+ sel = self.tree.selection()
+ if not sel:
+ messagebox.showinfo("No selection", "Select a variable first.")
+ return
+ # Resolve to a variable item (skip group headers)
+ item = sel[0]
+ name = self.tree.item(item, "text")
+ if not name or name.upper() in {"MISC","CORE","COOLANT","STEAM","CHEM","POWER","VALVE","RESISTOR"}:
+ messagebox.showinfo("Select variable", "Select a specific variable row (not a group).")
+ return
+ vi = self.vars.get(name) or VarInfo(name=name)
+
+ dlg = tk.Toplevel(self)
+ dlg.title(f"Thresholds & Alarms — {name}")
+ dlg.geometry("560x520")
+ dlg.transient(self); dlg.grab_set()
+
+ # Entries
+ t = vi.thresholds
+ grid = ttk.Frame(dlg, padding=10)
+ grid.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
+
+ entries = {}
+ def add_row(r, label, key, default):
+ ttk.Label(grid, text=label).grid(row=r, column=0, sticky="w", pady=4)
+ sv = tk.StringVar(value="" if default is None else str(default))
+ e = ttk.Entry(grid, textvariable=sv, width=12)
+ e.grid(row=r, column=1, sticky="w", padx=6)
+ entries[key] = sv
+
+ add_row(0, "DEAD LOW", "dead_low", t.dead_low)
+ add_row(1, "LOW", "low", t.low)
+ add_row(2, "HIGH", "high", t.high)
+ add_row(3, "EXTREME HIGH", "extreme_high", t.extreme_high)
+
+ # Alarm checkboxes
+ ttk.Label(grid, text="Alarm on enter:").grid(row=4, column=0, sticky="w", pady=(12,4))
+ alarm_vars = {}
+ for i, (lbl, key, val) in enumerate([("Dead Low", "alarm_dead_low", t.alarm_dead_low),
+ ("Low", "alarm_low", t.alarm_low),
+ ("High", "alarm_high", t.alarm_high),
+ ("Extreme High", "alarm_extreme_high", t.alarm_extreme_high)]):
+ var = tk.BooleanVar(value=val)
+ ttk.Checkbutton(grid, text=lbl, variable=var).grid(row=5+i//2, column=i%2+0, sticky="w", padx=6, pady=2)
+ alarm_vars[key] = var
+
+ # Action bindings
+ ttk.Label(grid, text="On enter, run function:").grid(row=7, column=0, sticky="w", pady=(12,4))
+ # Function selectors + value entries for each state
+ rows = [("Dead Low", "action_dead_low", "value_dead_low"),
+ ("Low", "action_low", "value_low"),
+ ("High", "action_high", "value_high"),
+ ("Extreme High", "action_extreme_high", "value_extreme_high")]
+ action_vars = {}
+ value_vars = {}
+ for i, (label, action_key, value_key) in enumerate(rows):
+ ttk.Label(grid, text=label).grid(row=8+i, column=0, sticky="w", pady=2)
+ s = ttk.Combobox(grid, values=self.functions_list, width=38, state="readonly")
+ s.set(getattr(t, action_key) or "")
+ s.grid(row=8+i, column=1, sticky="w", padx=6)
+ action_vars[action_key] = s
+ sv = tk.StringVar(value=getattr(t, value_key))
+ ttk.Entry(grid, textvariable=sv, width=12).grid(row=8+i, column=2, sticky="w", padx=6)
+ value_vars[value_key] = sv
+
+ def save_and_close():
+ # Parse floats or None
+ def _f(v):
+ v = v.strip()
+ return float(v) if v else None
+ nt = Thresholds(
+ dead_low=_f(entries["dead_low"].get()),
+ low=_f(entries["low"].get()),
+ high=_f(entries["high"].get()),
+ extreme_high=_f(entries["extreme_high"].get()),
+ alarm_dead_low=alarm_vars["alarm_dead_low"].get(),
+ alarm_low=alarm_vars["alarm_low"].get(),
+ alarm_high=alarm_vars["alarm_high"].get(),
+ alarm_extreme_high=alarm_vars["alarm_extreme_high"].get(),
+ action_dead_low=action_vars["action_dead_low"].get() or None,
+ value_dead_low=value_vars["value_dead_low"].get(),
+ action_low=action_vars["action_low"].get() or None,
+ value_low=value_vars["value_low"].get(),
+ action_high=action_vars["action_high"].get() or None,
+ value_high=value_vars["value_high"].get(),
+ action_extreme_high=action_vars["action_extreme_high"].get() or None,
+ value_extreme_high=value_vars["value_extreme_high"].get(),
+ )
+ vi.thresholds = nt
+ self.vars[name] = vi
+ dlg.destroy()
+
+ btns = ttk.Frame(dlg, padding=10)
+ btns.pack(side=tk.BOTTOM, fill=tk.X)
+ ttk.Button(btns, text="Save", command=save_and_close).pack(side=tk.RIGHT, padx=4)
+ ttk.Button(btns, text="Cancel", command=dlg.destroy).pack(side=tk.RIGHT, padx=4)
+
+ # ------------- Actions (POST) -------------
+ def run_func_once(self):
+ fname = self.func_name_var.get().strip()
+ if not fname:
+ messagebox.showinfo("No function", "Choose a function first.")
+ return
+ val = self.func_value_var.get()
+ status, body, _ = self.scheduler.run_once(fname, val)
+ messagebox.showinfo("Run Once", f"{fname}={val}\nHTTP {status}\n{coerce_preview(body, 300)}")
+
+ def add_schedule_dialog(self):
+ fname = self.func_name_var.get().strip()
+ if not fname:
+ messagebox.showinfo("No function", "Choose a function first.")
+ return
+ top = tk.Toplevel(self); top.title("Schedule Function"); top.geometry("360x160"); top.transient(self); top.grab_set()
+ ttk.Label(top, text=f"Function: {fname}").pack(pady=(10,4))
+ val_var = tk.StringVar(value=self.func_value_var.get())
+ ttk.Label(top, text="Value:").pack()
+ ttk.Entry(top, textvariable=val_var, width=16).pack()
+ int_var = tk.IntVar(value=30)
+ ttk.Label(top, text="Interval (seconds):").pack(pady=(6,0))
+ ttk.Entry(top, textvariable=int_var, width=10).pack()
+
+ def add_it():
+ task = ActionTask(name=fname, value=val_var.get(), interval_s=max(1, int_var.get()))
+ tid = self.scheduler.add_task(task)
+ self.refresh_actions_tree()
+ top.destroy()
+ ttk.Button(top, text="Add", command=add_it).pack(pady=10)
+
+ def on_actions_right_click(self, event):
+ try:
+ iid = self.actions_tree.identify_row(event.y)
+ if iid:
+ self.actions_tree.selection_set(iid)
+ finally:
+ self.actions_menu.tk_popup(event.x_root, event.y_root)
+
+ def toggle_selected_task(self):
+ sel = self.actions_tree.selection()
+ if not sel: return
+ tid = int(sel[0])
+ tasks = {t.task_id: t for t in self.scheduler.list_tasks()}
+ t = tasks.get(tid)
+ if not t: return
+ self.scheduler.set_enabled(tid, not t.enabled)
+ self.refresh_actions_tree()
+
+ def run_selected_task_now(self):
+ sel = self.actions_tree.selection()
+ if not sel: return
+ tid = int(sel[0])
+ tasks = {t.task_id: t for t in self.scheduler.list_tasks()}
+ t = tasks.get(tid)
+ if not t: return
+ status, body, _ = self.scheduler.run_once(t.name, t.value)
+ messagebox.showinfo("Run Now", f"{t.name}={t.value}\nHTTP {status}\n{coerce_preview(body, 300)}")
+
+ def remove_selected_task(self):
+ sel = self.actions_tree.selection()
+ if not sel: return
+ tid = int(sel[0])
+ self.scheduler.remove_task(tid)
+ self.refresh_actions_tree()
+
+ def refresh_actions_tree(self):
+ self.actions_tree.delete(*self.actions_tree.get_children())
+ for t in self.scheduler.list_tasks():
+ mode = "Interval" if t.interval_s > 0 else "Once"
+ next_s = ("—" if t.interval_s == 0 else datetime.fromtimestamp(t.next_run).strftime("%H:%M:%S"))
+ self.actions_tree.insert("", "end", iid=str(t.task_id), values=(t.value, mode, t.interval_s, next_s, "Yes" if t.enabled else "No"))
+
+ # ------------- 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. Use Select Vars… 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:
+ 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
+ # Evaluate thresholds & trigger events if needed
+ self.evaluate_thresholds(name, value)
+ 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()
+ self.refresh_actions_tree()
+ if self.poller and self.poller.is_alive():
+ self.after(250, self.drain_queue)
+
+ def stop_polling(self) -> None:
+ if self.poller and self.poller.is_alive():
+ self.stop_event.set()
+ 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")
+
+ # ------------- Threshold logic & rendering -------------
+ def evaluate_thresholds(self, name: str, value: str) -> None:
+ vi = self.vars.get(name)
+ if not vi:
+ return
+ t = vi.thresholds
+ x = parse_first_float(value)
+ if x is None:
+ # can't evaluate; treat as NORMAL unless error
+ new_state = "NORMAL"
+ else:
+ new_state = "NORMAL"
+ # Order of precedence: dead_low < low < high < extreme_high
+ if t.dead_low is not None and x <= t.dead_low:
+ new_state = "DEAD_LOW"
+ elif t.low is not None and x <= t.low:
+ new_state = "LOW"
+ elif t.extreme_high is not None and x >= t.extreme_high:
+ new_state = "EXTREME_HIGH"
+ elif t.high is not None and x >= t.high:
+ new_state = "HIGH"
+
+ if new_state != vi.last_state:
+ # Entering a state: run actions / alarms
+ self.on_enter_state(name, new_state, vi)
+ vi.last_state = new_state
+
+ def on_enter_state(self, name: str, state: str, vi: VarInfo) -> None:
+ t = vi.thresholds
+ # Alarm
+ alarm_map = {
+ "DEAD_LOW": t.alarm_dead_low,
+ "LOW": t.alarm_low,
+ "HIGH": t.alarm_high,
+ "EXTREME_HIGH": t.alarm_extreme_high,
+ }
+ if alarm_map.get(state):
+ play_alarm_tone(state)
+
+ # Function trigger
+ act_map = {
+ "DEAD_LOW": (t.action_dead_low, t.value_dead_low),
+ "LOW": (t.action_low, t.value_low),
+ "HIGH": (t.action_high, t.value_high),
+ "EXTREME_HIGH": (t.action_extreme_high, t.value_extreme_high),
+ }
+ act, val = act_map.get(state, (None, "1"))
+ if act:
+ # fire-and-forget one-shot via scheduler
+ self.scheduler.add_task(ActionTask(name=act, value=val, interval_s=0))
+
+ # ------------- Tree rendering -------------
+ def refresh_tree(self) -> None:
+ self.tree.delete(*self.tree.get_children())
+ 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():
+ 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)
+
+ 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:
+ value_preview = coerce_preview(vi.last_value, 120)
+
+ tag = infer_status_color(vi.last_value if not vi.error else "error")
+ # Append state suffix if thresholds defined
+ suffix = "" if vi.last_state in {"UNKNOWN", "NORMAL"} else f" [{vi.last_state}]"
+ self.tree.insert(
+ parent,
+ "end",
+ text=vi.name + suffix,
+ values=(value_preview, updated_str, vi.last_status or "—"),
+ tags=(tag,),
+ )
+
+ # ------------- Shutdown -------------
+ def on_close(self) -> None:
+ try:
+ self.stop_polling()
+ except Exception:
+ pass
+ try:
+ self.scheduler.stop()
+ except Exception:
+ pass
+ self.destroy()
+
+
+# =====================
+# Poller thread (after App to use its type hints without quotes)
# =====================
class Poller(threading.Thread):
def __init__(
@@ -170,7 +980,7 @@ class Poller(threading.Thread):
def run(self) -> None:
base_url = build_base_url(self.host, self.port)
- min_spacing = 1.0 / self.max_rps # seconds between requests
+ min_spacing = 1.0 / self.max_rps
while not self.stop_event.is_set():
cycle_start = time.time()
self.ui_queue.put(("cycle_start", datetime.now().strftime("%H:%M:%S")))
@@ -217,321 +1027,6 @@ class Poller(threading.Thread):
self.ui_queue.put(("stopped", ""))
-# =====================
-# GUI
-# =====================
-class App(tk.Tk):
- def __init__(self) -> None:
- super().__init__()
- self.title("Control Board Monitor")
- self.geometry("1150x720")
- try:
- self.iconbitmap(default="")
- 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)
- self.loadhtml_btn = ttk.Button(btns, text="Load HTML", command=self.load_html_dialog)
- self.loadhtml_btn.grid(row=0, column=4, 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=450, anchor="w")
- self.tree.column("value", width=450, anchor="w")
- self.tree.column("updated", width=120, anchor="center")
- self.tree.column("status", width=60, anchor="center")
-
- # Style tags
- style = ttk.Style(self)
- try:
- style.theme_use("clam")
- except Exception:
- pass
- self.tree.tag_configure("ok", background="#eaffea")
- self.tree.tag_configure("bad", background="#ffecec")
- self.tree.tag_configure("neutral", background="#f5f7fb")
-
- # 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 and accumulate
- status, body, _ = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"})
- if status == 200:
- try:
- data = json.loads(body)
- if isinstance(data, list):
- for x in data:
- if isinstance(x, str):
- v = x.strip().upper()
- if v and v not in discovered:
- discovered.append(v)
- except Exception:
- pass # ignore malformed JSON
-
- # Also parse the human HTML and merge (covers cases where JSON is incomplete)
- status_h, body_h, _h = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES"})
- if status_h == 200 and body_h:
- cand = parse_variable_names_from_html_index(body_h)
- for v in cand:
- if v not in discovered:
- discovered.append(v)
-
- # If still empty, provide a small 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",
- "POWER_FROM_TURBINE_KW", "POWER_DEMAND_MW", "CORE_IODINE_GENERATION"
- ]
-
- # Drop placeholders just in case
- discovered = [v for v in discovered if v not in {"VARNAME", "VARIABLE"}]
-
- 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-Za-z0-9_:-]+", txt.upper())
- if not names:
- messagebox.showwarning("Nothing found", "No variable-like tokens found.")
- return
- 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()
-
- def load_html_dialog(self) -> None:
- """Load a local HTML index file and parse variables (handy for testing)."""
- path = filedialog.askopenfilename(
- title="Select HTML file",
- filetypes=[("HTML files", "*.html *.htm"), ("All files", "*.*")],
- )
- if not path:
- return
- try:
- with open(path, "r", encoding="utf-8", errors="replace") as f:
- html = f.read()
- except Exception as e:
- messagebox.showerror("Read error", f"Could not read file:\n{e}")
- return
- cand = parse_variable_names_from_html_index(html)
- if not cand:
- messagebox.showwarning("No variables", "No variables recognized in that HTML.")
- return
- self.variables_list = cand
- self.vars.clear()
- for v in cand:
- self.vars[v] = VarInfo(name=v)
- self.status_lbl.configure(text=f"Loaded {len(cand)} from HTML")
- 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:
- 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)
-
- def stop_polling(self) -> None:
- if self.poller and self.poller.is_alive():
- self.stop_event.set()
- 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:
- self.tree.delete(*self.tree.get_children())
- 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():
- 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)
-
- 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:
- 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()