gitea/vtt_work
1
0
Fork 0
mirror of https://github.com/migatu/vtt_work.git synced 2026-07-14 13:34:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

0.1

Browse Source
This commit is contained in:
Michal Tuszowski
2025-10-20 18:24:38 +02:00
parent ec8b34044b
commit ca4a3b6523
1 changed files with 110 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nucleares_monitor/control_board_monitor.py
+110 -85
View File
@@ -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")