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
+73 -77
View File
@@ -1,18 +1,15 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Control Board Monitor — v0.7
Control Board Monitor — v0.8
Fixes & upgrades:
- ✅ Fixed AttributeError (missing run_func_once / add_schedule_dialog).
- ✅ Discovery now calls WEBSERVER_LIST_VARIABLES_JSON and parses "GET:" and "POST:" for variable/function names.
- ✅ Poller now fetches ALL values in one call via WEBSERVER_LIST_VARIABLES_JSON and only displays chosen ones.
- ✅ Threshold actions: besides calling a function, you can provide an EXPRESSION/snippet using x (current numeric value).
Notes:
- 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.
Spec alignment:
- Discovery of functions (POST actions) via WEBSERVER_LIST_VARIABLES_JSON → "POST:"
- Discovery of variables & polling via WEBSERVER_BATCH_GET (bulk values)
- Fallbacks: main page parse (/, GET links / POST section) → embedded defaults
- Expressions/snippets on thresholds with x = current value (safe eval)
- Functions executed via POST (variable=<FUNC>, value=<VAL>)
- Default host/port: localhost:8785
"""
import json
@@ -40,7 +37,7 @@ SERVER_HOST: str = "localhost"
SERVER_PORT: int = 8785
REFRESH_INTERVAL_S: int = 60
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_MAX_S: float = 15.0
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]
# ---- Parse WEBSERVER_LIST_VARIABLES_JSON ----
def parse_weblist_json(body: str) -> Tuple[List[str], List[str], Dict[str,str]]:
"""Return (get_list, post_list, values_map)."""
def parse_weblist_names(body: str) -> Tuple[List[str], List[str]]:
"""Return (get_list, post_list)."""
try:
data = json.loads(body)
except Exception:
return [], [], {}
return [], []
get_names: List[str] = []
post_names: List[str] = []
values_map: Dict[str, str] = {}
if isinstance(data, dict):
# Detect "GET"/"POST" blocks
for k, v in data.items():
key = str(k).strip().upper().rstrip(":")
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)
else:
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):
# Rare: a list of names only
get_names.extend(extract_names(data))
# Deduplicate preserving order
def dedup(seq):
seen=set(); out=[]
for s in seq:
if s not in seen:
seen.add(s); out.append(s)
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]:
out: List[str] = []
@@ -228,6 +210,24 @@ def extract_names(value_obj) -> List[str]:
add(m)
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
# =====================
@@ -317,7 +317,7 @@ class ActionScheduler(threading.Thread):
def run_once(self, name: str, value: str) -> Tuple[int, str, Dict[str, str]]:
base = self.get_base_url_cb()
# Functions should be POST per your spec
# Functions use POST
return http_post(base, {"variable": name, "value": value})
def run(self) -> None:
@@ -347,7 +347,7 @@ class ActionScheduler(threading.Thread):
self._stop.set()
# =====================
# Poller thread (aggregated)
# Poller thread (uses BATCH_GET)
# =====================
class Poller(threading.Thread):
def __init__(
@@ -375,30 +375,19 @@ class Poller(threading.Thread):
cycle_start = time.time()
self.ui_queue.put(("cycle_start", datetime.now().strftime("%H:%M:%S")))
# Try aggregated fetch
status, body, _ = http_get(base_url, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"})
status, body, _ = http_get(base_url, {"variable": "WEBSERVER_BATCH_GET"})
if status == 200:
try:
_get, _post, values_map = parse_weblist_json(body)
if values_map:
# Update only selected variables
upper_map = {k.upper(): v for k,v in values_map.items()}
for name in list(self.variables):
val = upper_map.get(name.upper())
if val is not None:
self.ui_queue.put(("update", name, str(val), 200))
else:
self.ui_queue.put(("error", name, 206, "Not in bulk payload"))
else:
# No map -> maybe it's list-only; fallback 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
values_map = parse_batch_values(body)
if values_map:
upper_map = {k.upper(): v for k, v in values_map.items()}
for name in list(self.variables):
val = upper_map.get(name.upper())
if val is not None:
self.ui_queue.put(("update", name, str(val), 200))
else:
self.ui_queue.put(("error", name, 206, "Not in BATCH_GET payload"))
else:
# fallback to per-variable
for name in list(self.variables):
st, b, _h = http_get(base_url, {"variable": name})
if st == 200:
@@ -406,7 +395,7 @@ class Poller(threading.Thread):
else:
self.ui_queue.put(("error", name, st, coerce_preview(b, 200)))
else:
# Bulk endpoint failed -> per-variable fallback
# fallback to per-variable
for name in list(self.variables):
st, b, _h = http_get(base_url, {"variable": name})
if st == 200:
@@ -433,7 +422,7 @@ class Poller(threading.Thread):
class App(tk.Tk):
def __init__(self) -> None:
super().__init__()
self.title("Control Board Monitor — v0.7")
self.title("Control Board Monitor — v0.8")
self.geometry("1250x820")
self.vars: Dict[str, VarInfo] = {}
@@ -560,30 +549,37 @@ class App(tk.Tk):
discovered_funcs: List[str] = []
msgs = []
# 1) WEBSERVER_LIST_VARIABLES_JSON — source of truth for GET/POST lists or values map
status, body, _ = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"})
if status == 200 and body:
v_get, v_post, values_map = parse_weblist_json(body)
if v_get: discovered_vars.extend(v_get)
if v_post: discovered_funcs.extend(v_post)
# 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()))
# 1) Functions via WEBSERVER_LIST_VARIABLES_JSON (POST list)
s1, b1, _1 = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"})
if s1 == 200 and b1:
_get_list, post_list = parse_weblist_names(b1)
if post_list:
discovered_funcs.extend(post_list)
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
s2, b2, _2 = http_get(base, {"variable": "WEBSERVER_BATCH_GET"})
if s2 == 200 and b2:
values_map = parse_batch_values(b2)
if values_map:
discovered_vars.extend(sorted(values_map.keys()))
else:
msgs.append(f"BATCH_GET HTTP {s2}: {coerce_preview(b2,120)}")
# 3) Fallbacks
# 3a: parse main page for vars/funcs
if not discovered_vars or not discovered_funcs:
s2, b2, _2 = http_get_root(base)
if s2 == 200 and b2:
s3, b3, _3 = http_get_root(base)
if s3 == 200 and b3:
if not discovered_vars:
discovered_vars = parse_variable_names_from_html_index(b2) or 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(b2) or discovered_funcs
discovered_funcs = parse_function_names_from_html_index(b3) or discovered_funcs
else:
msgs.append(f"Main page HTTP {s2}: {coerce_preview(b2, 120)}")
msgs.append(f"Main page HTTP {s3}: {coerce_preview(b3,120)}")
# 3) Final fallback to embedded defaults
# 3b: embedded defaults
used_defaults = False
if not discovered_vars:
discovered_vars = list(DEFAULT_VARS)
@@ -600,7 +596,7 @@ class App(tk.Tk):
if used_defaults:
messagebox.showwarning(
"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 ""))
self.status_lbl.configure(text=f"Loaded defaults ({len(discovered_vars)} vars, {len(discovered_funcs)} funcs)")
else:
@@ -843,7 +839,7 @@ class App(tk.Tk):
dlg = tk.Toplevel(self)
dlg.title(f"Thresholds, actions & alarms — {name}")
dlg.geometry("780x580")
dlg.geometry("780x600")
dlg.transient(self); dlg.grab_set()
t = vi.thresholds