From 61a8cb40a4e5040ea91758b1fee016d2f0044271 Mon Sep 17 00:00:00 2001 From: Michal Tuszowski Date: Mon, 20 Oct 2025 18:25:24 +0200 Subject: [PATCH] 0.3 --- nucleares_monitor/control_board_monitor.py | 1185 ++++++++++++++------ 1 file changed, 840 insertions(+), 345 deletions(-) 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()