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.8

Browse Source
This commit is contained in:
Michal Tuszowski
2025-10-20 18:26:49 +02:00
parent a73459ea87
commit 65b3989099
1 changed files with 73 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nucleares_monitor/control_board_monitor.py
+68 -72
View File
@@ -1,18 +1,15 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
Control Board Monitor — v0.7 Control Board Monitor — v0.8
Fixes & upgrades: Spec alignment:
- ✅ Fixed AttributeError (missing run_func_once / add_schedule_dialog). - Discovery of functions (POST actions) via WEBSERVER_LIST_VARIABLES_JSON → "POST:"
- ✅ Discovery now calls WEBSERVER_LIST_VARIABLES_JSON and parses "GET:" and "POST:" for variable/function names. - Discovery of variables & polling via WEBSERVER_BATCH_GET (bulk values)
- ✅ Poller now fetches ALL values in one call via WEBSERVER_LIST_VARIABLES_JSON and only displays chosen ones. - Fallbacks: main page parse (/, GET links / POST section) → embedded defaults
- ✅ Threshold actions: besides calling a function, you can provide an EXPRESSION/snippet using x (current numeric value). - Expressions/snippets on thresholds with x = current value (safe eval)
- Functions executed via POST (variable=<FUNC>, value=<VAL>)
Notes: - Default host/port: localhost:8785
- Default host/port = localhost:8785
- Functions are executed via POST (variable=<FUNC>, value=<VAL>).
- Discovery gracefully falls back to parsing main page "/" and then embedded defaults.
""" """
import json import json
@@ -40,7 +37,7 @@ SERVER_HOST: str = "localhost"
SERVER_PORT: int = 8785 SERVER_PORT: int = 8785
REFRESH_INTERVAL_S: int = 60 REFRESH_INTERVAL_S: int = 60
REQUEST_TIMEOUT_S: float = 5.0 REQUEST_TIMEOUT_S: float = 5.0
USER_AGENT: str = "ControlBoardMonitor/0.7 (+tkinter)" USER_AGENT: str = "ControlBoardMonitor/0.8 (+tkinter)"
BACKOFF_BASE_S: float = 1.0 BACKOFF_BASE_S: float = 1.0
BACKOFF_MAX_S: float = 15.0 BACKOFF_MAX_S: float = 15.0
MAX_RETRIES: int = 2 MAX_RETRIES: int = 2
@@ -157,18 +154,16 @@ def parse_function_names_from_html_index(html_text: str) -> List[str]:
return [v for v in uniq if v not in placeholders] return [v for v in uniq if v not in placeholders]
# ---- Parse WEBSERVER_LIST_VARIABLES_JSON ---- # ---- Parse WEBSERVER_LIST_VARIABLES_JSON ----
def parse_weblist_json(body: str) -> Tuple[List[str], List[str], Dict[str,str]]: def parse_weblist_names(body: str) -> Tuple[List[str], List[str]]:
"""Return (get_list, post_list, values_map).""" """Return (get_list, post_list)."""
try: try:
data = json.loads(body) data = json.loads(body)
except Exception: except Exception:
return [], [], {} return [], []
get_names: List[str] = [] get_names: List[str] = []
post_names: List[str] = [] post_names: List[str] = []
values_map: Dict[str, str] = {}
if isinstance(data, dict): if isinstance(data, dict):
# Detect "GET"/"POST" blocks
for k, v in data.items(): for k, v in data.items():
key = str(k).strip().upper().rstrip(":") key = str(k).strip().upper().rstrip(":")
if key in ("GET", "POST"): if key in ("GET", "POST"):
@@ -177,29 +172,16 @@ def parse_weblist_json(body: str) -> Tuple[List[str], List[str], Dict[str,str]]:
get_names.extend(names) get_names.extend(names)
else: else:
post_names.extend(names) post_names.extend(names)
# If dict looks like NAME->value mapping, collect as values_map
scalar_like = any(isinstance(v, (int, float, str)) for v in data.values())
if scalar_like and not get_names and not post_names:
for k, v in data.items():
name = str(k).strip().upper()
try:
values_map[name] = v if isinstance(v, str) else json.dumps(v)
except Exception:
values_map[name] = str(v)
elif isinstance(data, list): elif isinstance(data, list):
# Rare: a list of names only
get_names.extend(extract_names(data)) get_names.extend(extract_names(data))
# Deduplicate preserving order
def dedup(seq): def dedup(seq):
seen=set(); out=[] seen=set(); out=[]
for s in seq: for s in seq:
if s not in seen: if s not in seen:
seen.add(s); out.append(s) seen.add(s); out.append(s)
return out return out
return dedup(get_names), dedup(post_names), values_map return dedup(get_names), dedup(post_names)
def extract_names(value_obj) -> List[str]: def extract_names(value_obj) -> List[str]:
out: List[str] = [] out: List[str] = []
@@ -228,6 +210,24 @@ def extract_names(value_obj) -> List[str]:
add(m) add(m)
return out return out
# ---- Parse WEBSERVER_BATCH_GET ----
def parse_batch_values(body: str) -> Dict[str, str]:
try:
data = json.loads(body)
except Exception:
return {}
values: Dict[str, str] = {}
if isinstance(data, dict):
for k, v in data.items():
if not isinstance(k, str):
k = str(k)
key = k.strip().upper()
try:
values[key] = v if isinstance(v, str) else json.dumps(v)
except Exception:
values[key] = str(v)
return values
# ===================== # =====================
# Data model # Data model
# ===================== # =====================
@@ -317,7 +317,7 @@ class ActionScheduler(threading.Thread):
def run_once(self, name: str, value: str) -> Tuple[int, str, Dict[str, str]]: def run_once(self, name: str, value: str) -> Tuple[int, str, Dict[str, str]]:
base = self.get_base_url_cb() base = self.get_base_url_cb()
# Functions should be POST per your spec # Functions use POST
return http_post(base, {"variable": name, "value": value}) return http_post(base, {"variable": name, "value": value})
def run(self) -> None: def run(self) -> None:
@@ -347,7 +347,7 @@ class ActionScheduler(threading.Thread):
self._stop.set() self._stop.set()
# ===================== # =====================
# Poller thread (aggregated) # Poller thread (uses BATCH_GET)
# ===================== # =====================
class Poller(threading.Thread): class Poller(threading.Thread):
def __init__( def __init__(
@@ -375,30 +375,19 @@ class Poller(threading.Thread):
cycle_start = time.time() cycle_start = time.time()
self.ui_queue.put(("cycle_start", datetime.now().strftime("%H:%M:%S"))) self.ui_queue.put(("cycle_start", datetime.now().strftime("%H:%M:%S")))
# Try aggregated fetch status, body, _ = http_get(base_url, {"variable": "WEBSERVER_BATCH_GET"})
status, body, _ = http_get(base_url, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"})
if status == 200: if status == 200:
try: values_map = parse_batch_values(body)
_get, _post, values_map = parse_weblist_json(body)
if values_map: if values_map:
# Update only selected variables upper_map = {k.upper(): v for k, v in values_map.items()}
upper_map = {k.upper(): v for k,v in values_map.items()}
for name in list(self.variables): for name in list(self.variables):
val = upper_map.get(name.upper()) val = upper_map.get(name.upper())
if val is not None: if val is not None:
self.ui_queue.put(("update", name, str(val), 200)) self.ui_queue.put(("update", name, str(val), 200))
else: else:
self.ui_queue.put(("error", name, 206, "Not in bulk payload")) self.ui_queue.put(("error", name, 206, "Not in BATCH_GET payload"))
else: else:
# No map -> maybe it's list-only; fallback per variable # fallback to per-variable
for name in list(self.variables):
st, b, _h = http_get(base_url, {"variable": name})
if st == 200:
self.ui_queue.put(("update", name, b, st))
else:
self.ui_queue.put(("error", name, st, coerce_preview(b, 200)))
except Exception as e:
# Parsing error -> fallback to per-variable
for name in list(self.variables): for name in list(self.variables):
st, b, _h = http_get(base_url, {"variable": name}) st, b, _h = http_get(base_url, {"variable": name})
if st == 200: if st == 200:
@@ -406,7 +395,7 @@ class Poller(threading.Thread):
else: else:
self.ui_queue.put(("error", name, st, coerce_preview(b, 200))) self.ui_queue.put(("error", name, st, coerce_preview(b, 200)))
else: else:
# Bulk endpoint failed -> per-variable fallback # fallback to per-variable
for name in list(self.variables): for name in list(self.variables):
st, b, _h = http_get(base_url, {"variable": name}) st, b, _h = http_get(base_url, {"variable": name})
if st == 200: if st == 200:
@@ -433,7 +422,7 @@ class Poller(threading.Thread):
class App(tk.Tk): class App(tk.Tk):
def __init__(self) -> None: def __init__(self) -> None:
super().__init__() super().__init__()
self.title("Control Board Monitor — v0.7") self.title("Control Board Monitor — v0.8")
self.geometry("1250x820") self.geometry("1250x820")
self.vars: Dict[str, VarInfo] = {} self.vars: Dict[str, VarInfo] = {}
@@ -560,30 +549,37 @@ class App(tk.Tk):
discovered_funcs: List[str] = [] discovered_funcs: List[str] = []
msgs = [] msgs = []
# 1) WEBSERVER_LIST_VARIABLES_JSON — source of truth for GET/POST lists or values map # 1) Functions via WEBSERVER_LIST_VARIABLES_JSON (POST list)
status, body, _ = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"}) s1, b1, _1 = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"})
if status == 200 and body: if s1 == 200 and b1:
v_get, v_post, values_map = parse_weblist_json(body) _get_list, post_list = parse_weblist_names(b1)
if v_get: discovered_vars.extend(v_get) if post_list:
if v_post: discovered_funcs.extend(v_post) discovered_funcs.extend(post_list)
# If it's a values map only, use keys as variables
if (not v_get) and values_map:
discovered_vars.extend(list(values_map.keys()))
else: else:
msgs.append(f"LIST_JSON HTTP {status}: {coerce_preview(body, 120)}") msgs.append(f"LIST_VARIABLES_JSON HTTP {s1}: {coerce_preview(b1,120)}")
# 2) Fallback to main page parse # 2) Variables via WEBSERVER_BATCH_GET keys
if not discovered_vars or not discovered_funcs: s2, b2, _2 = http_get(base, {"variable": "WEBSERVER_BATCH_GET"})
s2, b2, _2 = http_get_root(base)
if s2 == 200 and b2: if s2 == 200 and b2:
if not discovered_vars: values_map = parse_batch_values(b2)
discovered_vars = parse_variable_names_from_html_index(b2) or discovered_vars if values_map:
if not discovered_funcs: discovered_vars.extend(sorted(values_map.keys()))
discovered_funcs = parse_function_names_from_html_index(b2) or discovered_funcs
else: else:
msgs.append(f"Main page HTTP {s2}: {coerce_preview(b2, 120)}") msgs.append(f"BATCH_GET HTTP {s2}: {coerce_preview(b2,120)}")
# 3) Final fallback to embedded defaults # 3) Fallbacks
# 3a: parse main page for vars/funcs
if not discovered_vars or not discovered_funcs:
s3, b3, _3 = http_get_root(base)
if s3 == 200 and b3:
if not discovered_vars:
discovered_vars = parse_variable_names_from_html_index(b3) or discovered_vars
if not discovered_funcs:
discovered_funcs = parse_function_names_from_html_index(b3) or discovered_funcs
else:
msgs.append(f"Main page HTTP {s3}: {coerce_preview(b3,120)}")
# 3b: embedded defaults
used_defaults = False used_defaults = False
if not discovered_vars: if not discovered_vars:
discovered_vars = list(DEFAULT_VARS) discovered_vars = list(DEFAULT_VARS)
@@ -600,7 +596,7 @@ class App(tk.Tk):
if used_defaults: if used_defaults:
messagebox.showwarning( messagebox.showwarning(
"Discovery fallback", "Discovery fallback",
"Could not discover variables/functions from WEBSERVER_LIST_VARIABLES_JSON or main page.\n" "Could not fully discover via LIST_VARIABLES_JSON / BATCH_GET.\n"
"Loaded embedded defaults so you can proceed.\n\n" + ("\n".join(msgs[:6]) if msgs else "")) "Loaded embedded defaults so you can proceed.\n\n" + ("\n".join(msgs[:6]) if msgs else ""))
self.status_lbl.configure(text=f"Loaded defaults ({len(discovered_vars)} vars, {len(discovered_funcs)} funcs)") self.status_lbl.configure(text=f"Loaded defaults ({len(discovered_vars)} vars, {len(discovered_funcs)} funcs)")
else: else:
@@ -843,7 +839,7 @@ class App(tk.Tk):
dlg = tk.Toplevel(self) dlg = tk.Toplevel(self)
dlg.title(f"Thresholds, actions & alarms — {name}") dlg.title(f"Thresholds, actions & alarms — {name}")
dlg.geometry("780x580") dlg.geometry("780x600")
dlg.transient(self); dlg.grab_set() dlg.transient(self); dlg.grab_set()
t = vi.thresholds t = vi.thresholds