#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Control Board Monitor - v0.3 (proposal build)
Features added per request:
1) Variable Selector dialog: after discovery, shows all variables with checkboxes.
Pre-checks currently-monitored set; user can add/remove before starting.
2) Functions (POST) Runner + Scheduler:
- Discovers writable variables/functions from the HTML POST section and optional JSON.
- "Run Once" with a value parameter.
- "Schedule" to run at an interval; can pause/resume/remove at any time.
3) Thresholds & Alarms per monitored value:
- Configure LOW/HIGH and optional DEAD_LOW/EXTREME_HIGH per variable.
- Detect state transitions (into LOW/HIGH/DEAD_LOW/EXTREME_HIGH from previous state).
- Optional alarm sound on entering a state.
- Optional function to invoke on entering a state (select any discovered function + value).
4) Tree-structured GUI:
- Top: Variable monitor tree (grouped by prefix, as before).
- Bottom: Actions/Scheduler tree.
- Menu/Buttons to open Variable Selector and Thresholds dialogs.
Stdlib-only: Tkinter, threading, urllib. Tested on Windows with Python 3.10+.
"""
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:
"""Best-effort alarm sound; uses winsound on Windows or tk.bell() otherwise."""
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:
# Fallback to Tk bell if available (non-blocking/chip)
root = tk._default_root
if root is not None:
root.bell()
except Exception:
pass
# =====================
# Global configuration
# =====================
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.3 (+tkinter)"
CONNECT_TIMEOUT_S: float = 5.0
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]]:
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]:
"""Extract readable variables (GET) from an index-like HTML page."""
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]:
"""Extract writable variables / functions from the POST section of the HTML."""
start = html_text.find("==== POST ====")
if start == -1:
return []
post_html = html_text[start:]
names = re.findall(r"([A-Z0-9_]+)", 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
# Optional function triggers on entering each state
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" # one of UNKNOWN/NORMAL/DEAD_LOW/LOW/HIGH/EXTREME_HIGH
@dataclass
class ActionTask:
name: str # function/variable name to POST
value: str # value to send
interval_s: int # 0 for one-shot
enabled: bool = True
last_run: float = 0.0
next_run: float = 0.0
task_id: int = 0 # assigned by scheduler
# =====================
# 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)
# Execute outside lock
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:
# one-shot: remove after run
self.remove_task(t.task_id)
except Exception:
# swallow to keep scheduler alive
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.3")
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("", 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("", 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 discovery (does not auto-start polling)
self.after(200, self.reload_discovery)
# ------------- 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] = []
# JSON for variables
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:
pass
# HTML
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)
# Functions from POST section
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)
# Optional JSON for functions
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:
pass
# If empty, keep previous; else present selector dialog
if discovered_vars:
self.show_selector_dialog(discovered_vars)
if discovered_funcs:
self.functions_list = discovered_funcs
self.func_combo["values"] = self.functions_list
self.status_lbl.configure(text=f"Discovered vars: {len(discovered_vars)} | functions: {len(self.functions_list)}")
def open_selector(self) -> None:
# Manual open (uses last discovered set = union of current + variables_list)
union = sorted(set(self.variables_list) | set(self.vars.keys()))
if not union:
messagebox.showinfo("No variables", "No discovered variables yet. Use Reload (disc.) first.")
return
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()
# Filter entry
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)
# Scrollable checkbox list
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("", 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")
# Build vars
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()
# Wire filter/select-all
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])
# Buttons
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
# reset var info dictionary to 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
# Resolve to a variable item (skip group headers)
item = sel[0]
name = self.tree.item(item, "text")
if not name or name.upper() in {"MISC","CORE","COOLANT","STEAM","CHEM","POWER","VALVE","RESISTOR"}:
messagebox.showinfo("Select variable", "Select a specific variable row (not a group).")
return
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()
# Entries
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)
# Alarm checkboxes
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
# Action bindings
ttk.Label(grid, text="On enter, run function:").grid(row=7, column=0, sticky="w", pady=(12,4))
# Function selectors + value entries for each state
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, 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():
# Parse floats or None
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
# Evaluate thresholds & trigger events if needed
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:
# can't evaluate; treat as NORMAL unless error
new_state = "NORMAL"
else:
new_state = "NORMAL"
# Order of precedence: dead_low < low < high < extreme_high
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:
# Entering a state: run actions / alarms
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
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)
# Function trigger
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:
# fire-and-forget one-shot via scheduler
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")
# Append state suffix if thresholds defined
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 (after App to use its type hints without quotes)
# =====================
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()