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
Files
311cb768b9c85235ea87b3c0ea5b25d5bdb734ff
vtt_work/nucleares_monitor/control_board_monitor.py
T
gitea 311cb768b9 0.4
2025-10-20 18:25:50 +02:00

1022 lines
44 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Control Board Monitor - v0.4
Fixes & additions:
- Default port changed to 8785.
- Discovery now shows clear error/fallback messages and *never* leaves you empty.
- If discovery fails, it loads a built-in default list (from your provided HTML).
- "Select Vars…" opens with defaults if nothing was discovered yet.
Everything else from v0.3 (selectors, functions/scheduler, thresholds/alarms) is included.
"""
import json
import queue
import re
import threading
import time
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Tuple
try:
import tkinter as tk
from tkinter import ttk, messagebox, simpledialog, filedialog
except Exception as e:
raise SystemExit("Tkinter is required to run this app.")
# ======= Windows beep support =======
def play_alarm_tone(level: str = "LOW") -> None:
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:
root = tk._default_root
if root is not None:
root.bell()
except Exception:
pass
# =====================
# Global configuration
# =====================
SERVER_HOST: str = "localhost"
SERVER_PORT: int = 8785 # changed as requested
REFRESH_INTERVAL_S: int = 60
MAX_RPS: float = 5.0
REQUEST_TIMEOUT_S: float = 5.0
USER_AGENT: str = "ControlBoardMonitor/0.4 (+tkinter)"
CONNECT_TIMEOUT_S: float = 5.0
BACKOFF_BASE_S: float = 1.0
BACKOFF_MAX_S: float = 15.0
MAX_RETRIES: int = 2
# Embedded defaults (from provided HTML)
DEFAULT_VARS: List[str] = ['ALARMS_ACTIVE', 'AMBIENT_TEMPERATURE', 'CHEM_BORON_DOSAGE_ACTUAL', 'CHEM_BORON_DOSAGE_ORDERED', 'CHEM_BORON_FILTER_ACTUAL', 'CHEM_BORON_FILTER_ORDERED', 'CHEM_BORON_PPM', 'CHEM_TRUCK_CONNECTED', 'CHEM_TRUCK_IN_ZONE', 'CHEM_TRUCK_PUMP_SPEED', 'CHEM_TRUCK_SAFETY_VALVE_CLOSED', 'CHEM_TRUCK_STATION_VALVE_CLOSED', 'COOLANT_CORE_CIRCULATION_PUMP_0_ACTUAL_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_0_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_0_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_1_ACTUAL_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_1_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_1_STATUS', 'COOLANT_CORE_FLOW_IN', 'COOLANT_CORE_FLOW_OUT', 'COOLANT_CORE_PRESSURE', 'COOLANT_PRIMARY_PRESSURE', 'COOLANT_SECONDARY_PRESSURE', 'COOLANT_TEMPERATURE_IN', 'COOLANT_TEMPERATURE_OUT', 'CONDENSER_PRESSURE', 'CONDENSER_TEMPERATURE', 'CONDENSER_VACUUM', 'CONDENSER_VACUUM_PUMP_0_STATUS', 'CONDENSER_VACUUM_PUMP_1_STATUS', 'CORE_BAY_1_FUEL_LOADING', 'CORE_BAY_1_HATCH', 'CORE_BAY_2_FUEL_LOADING', 'CORE_BAY_2_HATCH', 'CORE_BAY_3_FUEL_LOADING', 'CORE_BAY_3_HATCH', 'CORE_BAY_4_FUEL_LOADING', 'CORE_BAY_4_HATCH', 'CORE_BAY_5_FUEL_LOADING', 'CORE_BAY_5_HATCH', 'CORE_BAY_6_FUEL_LOADING', 'CORE_BAY_6_HATCH', 'CORE_BAY_7_FUEL_LOADING', 'CORE_BAY_7_HATCH', 'CORE_BAY_8_FUEL_LOADING', 'CORE_BAY_8_HATCH', 'CORE_BAY_9_FUEL_LOADING', 'CORE_BAY_9_HATCH', 'CORE_CONTROL_RODS_INSERTION_PERCENT', 'CORE_INTEGRITY', 'CORE_PRESSURE', 'CORE_STATE', 'CORE_TEMP', 'CRANE_ARM_POSITION', 'CRANE_GRABBER_CLOSED', 'CRANE_GRABBER_POSITION', 'CRANE_GRABBER_WEIGHT_KG', 'CRANE_LOCATION', 'CRANE_TORCH_ON', 'CRANE_TORCH_POWER', 'EMERGENCY_STOP', 'MAIN_POWER_SWITCH', 'POWER_DEMAND_MW', 'POWER_FROM_TURBINE_KW', 'RADIATION_BUILDING_CONTAINMENT', 'RADIATION_BUILDING_TURBINE', 'RADIATION_CONTROL_ROOM', 'RADIATION_OUTSIDE', 'RESISTOR_0_CURRENT_A', 'RESISTOR_0_ENABLED', 'RESISTOR_0_TEMP', 'RESISTOR_0_VOLTAGE_V', 'RESISTOR_1_CURRENT_A', 'RESISTOR_1_ENABLED', 'RESISTOR_1_TEMP', 'RESISTOR_1_VOLTAGE_V', 'RESISTOR_2_CURRENT_A', 'RESISTOR_2_ENABLED', 'RESISTOR_2_TEMP', 'RESISTOR_2_VOLTAGE_V', 'RESISTOR_3_CURRENT_A', 'RESISTOR_3_ENABLED', 'RESISTOR_3_TEMP', 'RESISTOR_3_VOLTAGE_V', 'STEAM_PRESSURE_AFTER_CONDENSER', 'STEAM_PRESSURE_AFTER_TURBINE', 'STEAM_PRESSURE_BEFORE_CONDENSER', 'STEAM_PRESSURE_BEFORE_TURBINE', 'STEAM_TEMPERATURE_AFTER_CONDENSER', 'STEAM_TEMPERATURE_AFTER_TURBINE', 'STEAM_TEMPERATURE_BEFORE_CONDENSER', 'STEAM_TEMPERATURE_BEFORE_TURBINE', 'STEAM_TURBINE_0_RPM', 'STEAM_TURBINE_1_RPM', 'STEAM_TURBINE_2_RPM', 'STEAM_TURBINE_VALVE_CLOSED', 'TIME', 'TIME_STAMP', 'VALVE_CLOSED', 'VALVE_OPEN', 'WEBSERVER_LIST_FUNCTIONS', 'WEBSERVER_LIST_FUNCTIONS_JSON', 'WEBSERVER_LIST_VARIABLES', 'WEBSERVER_LIST_VARIABLES_JSON']
DEFAULT_FUNCTIONS: List[str] = ['CHEM_BORON_DOSAGE_ORDERED_RATE', 'CHEM_BORON_FILTER_ORDERED_SPEED', 'CONDENSER_CIRCULATION_PUMP_ORDERED_SPEED', 'CONDENSER_CIRCULATION_PUMP_SWITCH', 'CONDENSER_VACUUM_PUMP_MODE', 'COOLANT_CORE_CIRCULATION_PUMP_0_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_0_SWITCH', 'COOLANT_CORE_CIRCULATION_PUMP_1_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_1_SWITCH', 'CORE_BAY_1_FUEL_LOADING', 'CORE_BAY_1_HATCH', 'CORE_BAY_2_FUEL_LOADING', 'CORE_BAY_2_HATCH', 'CORE_BAY_3_FUEL_LOADING', 'CORE_BAY_3_HATCH', 'CORE_BAY_4_FUEL_LOADING', 'CORE_BAY_4_HATCH', 'CORE_BAY_5_FUEL_LOADING', 'CORE_BAY_5_HATCH', 'CORE_BAY_6_FUEL_LOADING', 'CORE_BAY_6_HATCH', 'CORE_BAY_7_FUEL_LOADING', 'CORE_BAY_7_HATCH', 'CORE_BAY_8_FUEL_LOADING', 'CORE_BAY_8_HATCH', 'CORE_BAY_9_FUEL_LOADING', 'CORE_BAY_9_HATCH', 'CRANE_GRABBER_CLOSED', 'CRANE_TORCH_POWER', 'FUN_ALARM', 'FUN_BREAKER_TRIP', 'FUN_DECREASE_INTEGRITY', 'FUN_FIRE_DRILL', 'FUN_INCREASE_INTEGRITY', 'FUN_RESET_ALARMS', 'MAIN_POWER_SWITCH', 'RESISTOR_0_ENABLED', 'RESISTOR_1_ENABLED', 'RESISTOR_2_ENABLED', 'RESISTOR_3_ENABLED', 'STEAM_TURBINE_VALVE_CLOSED', 'VALVE_OFF', 'VALVE_OPEN']
# =====================
# Helper utilities
# =====================
def http_get(base_url: str, params: Dict[str, str]) -> Tuple[int, str, Dict[str, str]]:
url = base_url + "?" + urllib.parse.urlencode(params)
req = urllib.request.Request(url, 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 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]:
import urllib.parse
tokens: List[str] = []
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)
for m in re.finditer(r'variable\s*=\s*["\']?([A-Za-z0-9_.:-]+)', html_text, flags=re.I):
tokens.append(m.group(1))
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)
placeholders = {"VARNAME", "VARIABLE", "NAME"}
return [v for v in uniq if v not in placeholders]
def parse_function_names_from_html_index(html_text: str) -> List[str]:
start = html_text.find("==== POST ====")
if start == -1:
return []
post_html = html_text[start:]
names = re.findall(r"<b>([A-Z0-9_]+)</b>", 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", " ")
return v if len(v) <= maxlen else (v[:maxlen - 1] + "…")
def infer_status_color(value: str) -> str:
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"
if val == "" or "exception" in val or "error" in val:
return "bad"
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
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
last_value: str = ""
last_updated: float = 0.0
last_status: int = 0
error: Optional[str] = None
thresholds: Thresholds = field(default_factory=Thresholds)
last_state: str = "UNKNOWN"
@dataclass
class ActionTask:
name: str
value: str
interval_s: int
enabled: bool = True
last_run: float = 0.0
next_run: float = 0.0
task_id: int = 0
# =====================
# 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)
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:
self.remove_task(t.task_id)
except Exception:
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.4")
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("<Button-3>", 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("<Button-3>", 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: don't assume server works; just preload defaults into selector
self.variables_list = []
self.functions_list = list(DEFAULT_FUNCTIONS)
self.func_combo["values"] = self.functions_list
# optionally, we could auto-open selector; but keep consistent with your flow.
# ------------- 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] = []
msgs = []
# Try JSON vars
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 as e:
msgs.append(f"Vars JSON parse error: {e}")
else:
msgs.append(f"Vars JSON HTTP {status}: {coerce_preview(body, 120)}")
# Try HTML vars + functions
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)
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)
else:
msgs.append(f"Vars HTML HTTP {status_h}: {coerce_preview(body_h, 120)}")
# Optional functions JSON
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 as e:
msgs.append(f"Funcs JSON parse error: {e}")
elif status_f != 0:
msgs.append(f"Funcs JSON HTTP {status_f}: {coerce_preview(body_f, 120)}")
# Fallbacks
used_defaults = False
if not discovered_vars:
discovered_vars = list(DEFAULT_VARS)
used_defaults = True
if not discovered_funcs:
discovered_funcs = list(DEFAULT_FUNCTIONS)
# Apply
self.functions_list = discovered_funcs
self.func_combo["values"] = self.functions_list
self.show_selector_dialog(discovered_vars)
# User-visible status
if used_defaults:
messagebox.showwarning(
"Discovery failed — loaded defaults",
"Could not discover variables from the server.\n"
"Loaded a built-in list from your spec so you can proceed.\n\n"
+ ("\n".join(msgs[:6]) if msgs else "No server responses."))
self.status_lbl.configure(text=f"Loaded defaults ({len(discovered_vars)} vars, {len(discovered_funcs)} funcs)")
else:
self.status_lbl.configure(text=f"Discovered vars: {len(discovered_vars)} | functions: {len(discovered_funcs)}")
def open_selector(self) -> None:
union = sorted(set(self.variables_list) | set(self.vars.keys()))
if not union:
# use defaults if nothing yet
union = list(DEFAULT_VARS)
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()
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)
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("<Configure>", 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")
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()
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])
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
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
item = sel[0]
name = self.tree.item(item, "text")
# If user right-clicked a group header, bail
if not name or "_" not in name:
messagebox.showinfo("Select variable", "Select a specific variable row (not a group).")
return
# remove any state suffix in brackets
if name.endswith("]") and "[" in name:
name = name[:name.rfind("[")].strip()
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()
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)
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
ttk.Label(grid, text="On enter, run function:").grid(row=7, column=0, sticky="w", pady=(12,4))
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 or DEFAULT_FUNCTIONS, 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():
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
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:
new_state = "NORMAL"
else:
new_state = "NORMAL"
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:
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_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)
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:
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")
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
# =====================
class Poller(threading.Thread):
def __init__(
self,
host: str,
port: int,
variables: List[str],
ui_queue: queue.Queue,
refresh_interval: int,
max_rps: float,
stop_event: threading.Event,
paused_event: threading.Event,
) -> None:
super().__init__(daemon=True)
self.host = host
self.port = port
self.variables = variables
self.ui_queue = ui_queue
self.refresh_interval = int(max(5, refresh_interval))
self.max_rps = max(0.5, max_rps)
self.stop_event = stop_event
self.paused_event = paused_event
def run(self) -> None:
base_url = build_base_url(self.host, self.port)
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")))
for name in list(self.variables):
while self.paused_event.is_set() and not self.stop_event.is_set():
time.sleep(0.1)
if self.stop_event.is_set():
break
attempt = 0
backoff = BACKOFF_BASE_S
while attempt <= MAX_RETRIES and not self.stop_event.is_set():
attempt += 1
t0 = time.time()
status, body, headers = http_get(base_url, {"variable": name})
if status == 200:
self.ui_queue.put(("update", name, body, status))
break
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:
self.ui_queue.put(("error", name, status, coerce_preview(body, 200)))
break
dt = time.time() - t0
if dt < min_spacing:
time.sleep(min_spacing - dt)
dt = time.time() - t0
if dt < min_spacing:
time.sleep(min_spacing - dt)
cycle_dt = time.time() - cycle_start
remaining = self.refresh_interval - cycle_dt
if remaining > 0:
for _ in range(int(remaining * 10)):
if self.stop_event.is_set():
break
time.sleep(0.1)
self.ui_queue.put(("stopped", ""))
def main() -> None:
app = App()
app.mainloop()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink