From ca4a3b65234075e3aafd9b43d4277a69dd3cc977 Mon Sep 17 00:00:00 2001 From: Michal Tuszowski Date: Mon, 20 Oct 2025 18:24:38 +0200 Subject: [PATCH] 0.1 --- nucleares_monitor/control_board_monitor.py | 195 ++++++++++++--------- 1 file changed, 110 insertions(+), 85 deletions(-) diff --git a/nucleares_monitor/control_board_monitor.py b/nucleares_monitor/control_board_monitor.py index 5da46a5..06c4269 100644 --- a/nucleares_monitor/control_board_monitor.py +++ b/nucleares_monitor/control_board_monitor.py @@ -1,22 +1,12 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -Control Board Monitor - v0.1 +Control Board Monitor - v0.2 -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). +- 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. """ import json @@ -26,13 +16,13 @@ import threading import time import urllib.parse import urllib.request -from dataclasses import dataclass, field +from dataclasses import dataclass from datetime import datetime from typing import Dict, List, Optional, Tuple try: import tkinter as tk - from tkinter import ttk, messagebox, simpledialog + from tkinter import ttk, messagebox, simpledialog, filedialog except Exception as e: raise SystemExit("Tkinter is required to run this app.") @@ -40,25 +30,23 @@ except Exception as e: # ===================== # 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)" +SERVER_HOST: str = "localhost" +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)" 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 +BACKOFF_BASE_S: float = 1.0 +BACKOFF_MAX_S: float = 15.0 +MAX_RETRIES: int = 2 # ===================== # 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: @@ -82,22 +70,46 @@ def http_get(base_url: str, params: Dict[str, str]) -> Tuple[int, str, Dict[str, def build_base_url(host: str, port: int) -> str: - scheme = "http" - return f"{scheme}://{host}:{port}/" + return f"http://{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 + """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". + """ + 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) + qs = urllib.parse.parse_qs(parsed.query) + 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 = [] - for v in vars_found: - if v not in seen: - uniq.append(v) - seen.add(v) + 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) + + # Filter obvious placeholders + placeholders = {"VARNAME", "VARIABLE", "NAME"} + uniq = [v for v in uniq if v not in placeholders] + return uniq @@ -109,13 +121,11 @@ def coerce_preview(value: str, maxlen: int = 80) -> str: 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" @@ -165,7 +175,6 @@ class Poller(threading.Thread): 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(): @@ -177,33 +186,26 @@ class Poller(threading.Thread): 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: @@ -222,9 +224,9 @@ class App(tk.Tk): def __init__(self) -> None: super().__init__() self.title("Control Board Monitor") - self.geometry("1100x700") + self.geometry("1150x720") try: - self.iconbitmap(default="") # no icon file bundled + self.iconbitmap(default="") except Exception: pass @@ -267,6 +269,8 @@ class App(tk.Tk): 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)) @@ -286,20 +290,20 @@ class App(tk.Tk): 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("#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 for a control-board feel + # Style tags 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 + 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) @@ -319,33 +323,39 @@ class App(tk.Tk): base = build_base_url(host, port) discovered: List[str] = [] - # Try JSON list first + # 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) and all(isinstance(x, str) for x in data): - discovered = list(dict.fromkeys(data)) # preserve order, unique + 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: - discovered = [] + pass # ignore malformed JSON - # 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 + # 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, offer a minimal starter set + # 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", - "STEAM_TURBINE_0_RPM", "POWER_FROM_TURBINE_KW", "POWER_DEMAND_MW", + "POWER_FROM_TURBINE_KW", "POWER_DEMAND_MW", "CORE_IODINE_GENERATION" ] - # Update local state + # 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: @@ -362,13 +372,11 @@ class App(tk.Tk): ) if not txt: return - names = re.findall(r"[A-Z0-9_]+", txt.upper()) + names = re.findall(r"[A-Za-z0-9_:-]+", txt.upper()) if not names: messagebox.showwarning("Nothing found", "No variable-like tokens found.") return - # Deduplicate while preserving order - uniq = [] - seen = set() + uniq, seen = [], set() for n in names: if n not in seen: uniq.append(n); seen.add(n) @@ -379,6 +387,31 @@ class App(tk.Tk): 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(): @@ -418,7 +451,6 @@ class App(tk.Tk): 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() @@ -448,12 +480,11 @@ class App(tk.Tk): pass self.refresh_tree() if self.poller and self.poller.is_alive(): - self.after(250, self.drain_queue) # continue pumping + self.after(250, self.drain_queue) 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 @@ -463,22 +494,17 @@ class App(tk.Tk): # ------------- 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): @@ -489,7 +515,6 @@ class App(tk.Tk): 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")