#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Control Board Monitor — v0.7 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=, value=). - Discovery gracefully falls back to parsing main page "/" and then embedded defaults. """ import json import math 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 except Exception: raise SystemExit("Tkinter is required to run this app.") # ===================== # Global configuration # ===================== 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)" BACKOFF_BASE_S: float = 1.0 BACKOFF_MAX_S: float = 15.0 MAX_RETRIES: int = 2 # Fallback defaults (extracted from your 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', 'CHEMICAL_CLEANING_PUMP_DRY_STATUS', 'CHEMICAL_CLEANING_PUMP_OVERLOAD_STATUS', 'CHEMICAL_CLEANING_PUMP_STATUS', 'CHEMICAL_DOSING_PUMP_DRY_STATUS', 'CHEMICAL_DOSING_PUMP_OVERLOAD_STATUS', 'CHEMICAL_DOSING_PUMP_STATUS', 'CHEMICAL_FILTER_PUMP_DRY_STATUS', 'CHEMICAL_FILTER_PUMP_OVERLOAD_STATUS', 'CHEMICAL_FILTER_PUMP_STATUS', 'CONDENSER_CIRCULATION_PUMP_ACTIVE', 'CONDENSER_CIRCULATION_PUMP_ORDERED_SPEED', 'CONDENSER_CIRCULATION_PUMP_OVERLOAD_STATUS', 'CONDENSER_CIRCULATION_PUMP_SPEED', 'CONDENSER_CIRCULATION_PUMP_SWITCH', 'CONDENSER_CONDENSATE_FLOW_RATE', 'CONDENSER_COOLANT_EVAPORATED', 'CONDENSER_EXTRACTION_FLOW_RATE', 'CONDENSER_PRESSURE', 'CONDENSER_TEMPERATURE', 'CONDENSER_VACUUM', 'CONDENSER_VACUUM_PUMP_ACTIVE', 'CONDENSER_VACUUM_PUMP_MODE', 'CONDENSER_VACUUM_PUMP_POWER', 'CONDENSER_VACUUM_RELIEF_VALVE_OPENING', 'CONDENSER_VAPOR_VOLUME', 'CONDENSER_VOLUME', 'COOLANT_CORE_CIRCULATION_PUMP_0_CAPACITY', 'COOLANT_CORE_CIRCULATION_PUMP_0_DRY_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_0_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_0_OVERLOAD_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_0_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_0_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_1_CAPACITY', 'COOLANT_CORE_CIRCULATION_PUMP_1_DRY_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_1_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_1_OVERLOAD_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_1_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_1_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_2_CAPACITY', 'COOLANT_CORE_CIRCULATION_PUMP_2_DRY_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_2_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_2_OVERLOAD_STATUS', 'COOLANT_CORE_CIRCULATION_PUMP_2_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_2_STATUS', 'COOLANT_CORE_FLOW_IN', 'COOLANT_CORE_FLOW_ORDERED_SPEED', 'COOLANT_CORE_FLOW_OUT', 'COOLANT_CORE_FLOW_REACHED_SPEED', 'COOLANT_CORE_FLOW_SPEED', 'COOLANT_CORE_MAX_PRESSURE', 'COOLANT_CORE_PRESSURE', 'COOLANT_CORE_PRIMARY_LOOP_LEVEL', 'COOLANT_CORE_QUANTITY_CIRCULATION_PUMPS_PRESENT', 'COOLANT_CORE_QUANTITY_FREIGHT_PUMPS_PRESENT', 'COOLANT_CORE_QUANTITY_IN_VESSEL', 'COOLANT_CORE_STATE', 'COOLANT_CORE_VESSEL_TEMPERATURE', 'COOLANT_SEC_0_LIQUID_VOLUME', 'COOLANT_SEC_0_PRESSURE', 'COOLANT_SEC_0_TEMPERATURE', 'COOLANT_SEC_0_VOLUME', 'COOLANT_SEC_1_LIQUID_VOLUME', 'COOLANT_SEC_1_PRESSURE', 'COOLANT_SEC_1_TEMPERATURE', 'COOLANT_SEC_1_VOLUME', 'COOLANT_SEC_2_LIQUID_VOLUME', 'COOLANT_SEC_2_PRESSURE', 'COOLANT_SEC_2_TEMPERATURE', 'COOLANT_SEC_2_VOLUME', 'COOLANT_SEC_CIRCULATION_PUMP_0_CAPACITY', 'COOLANT_SEC_CIRCULATION_PUMP_0_DRY_STATUS', 'COOLANT_SEC_CIRCULATION_PUMP_0_ORDERED_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_0_OVERLOAD_STATUS', 'COOLANT_SEC_CIRCULATION_PUMP_0_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_0_STATUS', 'COOLANT_SEC_CIRCULATION_PUMP_1_CAPACITY', 'COOLANT_SEC_CIRCULATION_PUMP_1_DRY_STATUS', 'COOLANT_SEC_CIRCULATION_PUMP_1_ORDERED_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_1_OVERLOAD_STATUS', 'COOLANT_SEC_CIRCULATION_PUMP_1_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_1_STATUS', 'COOLANT_SEC_CIRCULATION_PUMP_2_CAPACITY', 'COOLANT_SEC_CIRCULATION_PUMP_2_DRY_STATUS', 'COOLANT_SEC_CIRCULATION_PUMP_2_ORDERED_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_2_OVERLOAD_STATUS', 'COOLANT_SEC_CIRCULATION_PUMP_2_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_2_STATUS', 'CORE_BAY_1_HATCH_OPEN', 'CORE_BAY_1_STATE', 'CORE_BAY_2_HATCH_OPEN', 'CORE_BAY_2_STATE', 'CORE_BAY_3_HATCH_OPEN', 'CORE_BAY_3_STATE', 'CORE_BAY_4_HATCH_OPEN', 'CORE_BAY_4_STATE', 'CORE_BAY_5_HATCH_OPEN', 'CORE_BAY_5_STATE', 'CORE_BAY_6_HATCH_OPEN', 'CORE_BAY_6_STATE', 'CORE_BAY_7_HATCH_OPEN', 'CORE_BAY_7_STATE', 'CORE_BAY_8_HATCH_OPEN', 'CORE_BAY_8_STATE', 'CORE_BAY_9_HATCH_OPEN', 'CORE_BAY_9_STATE', 'CORE_CRITICAL_MASS_REACHED', 'CORE_CRITICAL_MASS_REACHED_COUNTER', 'CORE_EXTERNAL_COOLANT_RESERVOIR_VOLUME', 'CORE_FACTOR', 'CORE_FACTOR_CHANGE', 'CORE_FUEL_1_FISSIONABLE', 'CORE_FUEL_1_POWER_FACTOR', 'CORE_FUEL_1_TEMPERATURE', 'CORE_FUEL_2_FISSIONABLE', 'CORE_FUEL_2_POWER_FACTOR', 'CORE_FUEL_2_TEMPERATURE', 'CORE_FUEL_3_FISSIONABLE', 'CORE_FUEL_3_POWER_FACTOR', 'CORE_FUEL_3_TEMPERATURE', 'CORE_FUEL_4_FISSIONABLE', 'CORE_FUEL_4_POWER_FACTOR', 'CORE_FUEL_4_TEMPERATURE', 'CORE_FUEL_5_FISSIONABLE', 'CORE_FUEL_5_POWER_FACTOR', 'CORE_FUEL_5_TEMPERATURE', 'CORE_FUEL_6_FISSIONABLE', 'CORE_FUEL_6_POWER_FACTOR', 'CORE_FUEL_6_TEMPERATURE', 'CORE_FUEL_7_FISSIONABLE', 'CORE_FUEL_7_POWER_FACTOR', 'CORE_FUEL_7_TEMPERATURE', 'CORE_FUEL_8_FISSIONABLE', 'CORE_FUEL_8_POWER_FACTOR', 'CORE_FUEL_8_TEMPERATURE', 'CORE_FUEL_9_FISSIONABLE', 'CORE_FUEL_9_POWER_FACTOR', 'CORE_FUEL_9_TEMPERATURE', 'CORE_FUEL_AVG_FISSIONABLE', 'CORE_FUEL_AVG_POWER_FACTOR', 'CORE_FUEL_AVG_TEMPERATURE', 'CORE_HIGH_STEAM_PRESENT', 'CORE_IMMINENT_FUSION', 'CORE_INTEGRITY', 'CORE_IODINE_CUMULATIVE', 'CORE_IODINE_GENERATION', 'CORE_OPERATION_MODE', 'CORE_POOL_COOLANT_TANK_VOLUME', 'CORE_POOL_PUMP', 'CORE_PRESSURE', 'CORE_PRESSURE_MAX', 'CORE_PRESSURE_OPERATIVE', 'CORE_PRIMARY_CIRCUIT_COOLING_TANK_VOLUME', 'CORE_READY_FOR_START', 'CORE_STATE', 'CORE_STATE_CRITICALITY', 'CORE_STEAM_PRESENT', 'CORE_TEMP', 'CORE_TEMP_MAX', 'CORE_TEMP_MIN', 'CORE_TEMP_OPERATIVE', 'CORE_TEMP_RESIDUAL', 'CORE_WEAR', 'CORE_XENON_CUMULATIVE', 'CORE_XENON_GENERATION', 'EMERGENCY_BATTERIES_MODE', 'EMERGENCY_BATTERIES_POWER_OUTPUT_KW', 'EMERGENCY_GENERATOR_1_FUEL', 'EMERGENCY_GENERATOR_1_MAINTENANCE_NEEDED', 'EMERGENCY_GENERATOR_1_MODE', 'EMERGENCY_GENERATOR_1_PRESSURIZER', 'EMERGENCY_GENERATOR_1_STATUS', 'EMERGENCY_GENERATOR_2_FUEL', 'EMERGENCY_GENERATOR_2_MAINTENANCE_NEEDED', 'EMERGENCY_GENERATOR_2_MODE', 'EMERGENCY_GENERATOR_2_PRESSURIZER', 'EMERGENCY_GENERATOR_2_STATUS', 'EMERGENCY_GENERATOR_POWER_OUTPUT_KW', 'FREIGHT_PUMP_CONDENSER_ACTIVE', 'FREIGHT_PUMP_CONDENSER_SWITCH', 'FREIGHT_PUMP_EXTERNAL_ACTIVE', 'FREIGHT_PUMP_EXTERNAL_SWITCH', 'FREIGHT_PUMP_FEEDWATER_ACTIVE', 'FREIGHT_PUMP_FEEDWATER_SWITCH', 'FREIGHT_PUMP_INTERNAL_ACTIVE', 'FREIGHT_PUMP_INTERNAL_SWITCH', 'FUN_IS_ENABLED', 'GAME_DIFFICULTY', 'GAME_SIM_SPEED', 'GAME_VERSION', 'GENERATOR_0_A', 'GENERATOR_0_BREAKER', 'GENERATOR_0_HERTZ', 'GENERATOR_0_KW', 'GENERATOR_0_V', 'GENERATOR_1_A', 'GENERATOR_1_BREAKER', 'GENERATOR_1_HERTZ', 'GENERATOR_1_KW', 'GENERATOR_1_V', 'GENERATOR_2_A', 'GENERATOR_2_BREAKER', 'GENERATOR_2_HERTZ', 'GENERATOR_2_KW', 'GENERATOR_2_V', 'INSTALLED_LOOPS_JSON', 'INVENTORY_HTML', 'MAINTENANCE_REPORT_HTML', 'MSCV_0_OPENING_ACTUAL', 'MSCV_1_OPENING_ACTUAL', 'MSCV_2_OPENING_ACTUAL', 'POWER_DEMAND_MW', 'POWER_FROM_EXTERNAL_KW', 'POWER_FROM_TURBINE_KW', 'POWER_MAX_THEORETICAL_FINAL_PLANT_OUTPUT_MW', 'POWER_MAX_THEORETICAL_PLANT_OUTPUT_MW', 'RES_ABSORPTION_CAPACITY_MW', 'RES_DIVERT_SURPLUS_FROM_MW', 'RES_EFFECTIVELY_DERIVED_ENERGY_MW', 'RESISTOR_BANK_01_SWITCH', 'RESISTOR_BANK_02_SWITCH', 'RESISTOR_BANK_03_SWITCH', 'RESISTOR_BANK_04_SWITCH', 'RESISTOR_BANKS_JSON', 'RESISTOR_BANKS_MAIN_SWITCH', 'ROD_BANK_POS_0_ACTUAL', 'ROD_BANK_POS_0_ORDERED', 'ROD_BANK_POS_1_ACTUAL', 'ROD_BANK_POS_1_ORDERED', 'ROD_BANK_POS_2_ACTUAL', 'ROD_BANK_POS_2_ORDERED', 'ROD_BANK_POS_3_ACTUAL', 'ROD_BANK_POS_3_ORDERED', 'ROD_BANK_POS_4_ACTUAL', 'ROD_BANK_POS_4_ORDERED', 'ROD_BANK_POS_5_ACTUAL', 'ROD_BANK_POS_5_ORDERED', 'ROD_BANK_POS_6_ACTUAL', 'ROD_BANK_POS_6_ORDERED', 'ROD_BANK_POS_7_ACTUAL', 'ROD_BANK_POS_7_ORDERED', 'ROD_BANK_POS_8_ACTUAL', 'ROD_BANK_POS_8_ORDERED', 'RODS_ALIGNED', 'RODS_DEFORMED', 'RODS_MAX_TEMPERATURE', 'RODS_MOVEMENT_SPEED', 'RODS_MOVEMENT_SPEED_DECREASED_HIGH_TEMPERATURE', 'RODS_POS_ACTUAL', 'RODS_POS_ORDERED', 'RODS_POS_REACHED', 'RODS_QUANTITY', 'RODS_STATUS', 'RODS_TEMPERATURE', 'STEAM_EJECTOR_CONDENSER_RETURN_VALVE_ACTUAL', 'STEAM_EJECTOR_CONDENSER_RETURN_VALVE_ORDERED', 'STEAM_EJECTOR_MOTIVE', 'STEAM_EJECTOR_OPERATIONAL_MOTIVE_VALVE_ACTUAL', 'STEAM_EJECTOR_OPERATIONAL_MOTIVE_VALVE_ORDERED', 'STEAM_EJECTOR_STARTUP_MOTIVE_VALVE_ACTUAL', 'STEAM_EJECTOR_STARTUP_MOTIVE_VALVE_ORDERED', 'STEAM_GEN_0_BOILING_POINT', 'STEAM_GEN_0_EVAPORATED', 'STEAM_GEN_0_INLET', 'STEAM_GEN_0_OUTLET', 'STEAM_GEN_0_RETURN_FLOW_PLUS_CONDENSED', 'STEAM_GEN_0_STATUS', 'STEAM_GEN_0_VENT_SWITCH', 'STEAM_GEN_1_BOILING_POINT', 'STEAM_GEN_1_EVAPORATED', 'STEAM_GEN_1_INLET', 'STEAM_GEN_1_OUTLET', 'STEAM_GEN_1_RETURN_FLOW_PLUS_CONDENSED', 'STEAM_GEN_1_STATUS', 'STEAM_GEN_1_VENT_SWITCH', 'STEAM_GEN_2_BOILING_POINT', 'STEAM_GEN_2_EVAPORATED', 'STEAM_GEN_2_INLET', 'STEAM_GEN_2_OUTLET', 'STEAM_GEN_2_RETURN_FLOW_PLUS_CONDENSED', 'STEAM_GEN_2_STATUS', 'STEAM_GEN_2_VENT_SWITCH', 'STEAM_TURBINE_0_BYPASS_ACTUAL', 'STEAM_TURBINE_0_INSTALLED', 'STEAM_TURBINE_0_PRESSURE', 'STEAM_TURBINE_0_RPM', 'STEAM_TURBINE_0_TEMPERATURE', 'STEAM_TURBINE_0_TORQUE', 'STEAM_TURBINE_1_BYPASS_ACTUAL', 'STEAM_TURBINE_1_INSTALLED', 'STEAM_TURBINE_1_PRESSURE', 'STEAM_TURBINE_1_RPM', 'STEAM_TURBINE_1_TEMPERATURE', 'STEAM_TURBINE_1_TORQUE', 'STEAM_TURBINE_2_BYPASS_ACTUAL', 'STEAM_TURBINE_2_INSTALLED', 'STEAM_TURBINE_2_PRESSURE', 'STEAM_TURBINE_2_RPM', 'STEAM_TURBINE_2_TEMPERATURE', 'STEAM_TURBINE_2_TORQUE', 'TIME', 'TIME_DAY', 'TIME_STAMP', 'VACUUM_RETENTION_TANK_PRESSURE', 'VACUUM_RETENTION_TANK_VOLUME', 'VALVE_M01_OPEN', 'VALVE_M02_OPEN', 'VALVE_M03_OPEN', 'VALVE_PANEL_JSON', 'WEATHER_FORECAST_JSON', 'WEBSERVER_BATCH_GET', 'WEBSERVER_LIST_VARIABLES', 'WEBSERVER_LIST_VARIABLES_JSON', 'WEBSERVER_VIEW_VARIABLES'] 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', 'CONDENSER_VACUUM_PUMP_START_STOP', 'COOLANT_CORE_CIRCULATION_PUMP_0_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_1_ORDERED_SPEED', 'COOLANT_CORE_CIRCULATION_PUMP_2_ORDERED_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_0_ORDERED_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_1_ORDERED_SPEED', 'COOLANT_SEC_CIRCULATION_PUMP_2_ORDERED_SPEED', '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_EMERGENCY_STOP', 'CORE_END_EMERGENCY_STOP', 'CORE_OPERATION_MODE', 'CORE_POOL_PUMP', 'CORE_SCRAM_BUTTON', 'EMERGENCY_BATTERIES_MODE', 'EMERGENCY_GENERATOR_1_MODE', 'EMERGENCY_GENERATOR_1_START_STOP', 'EMERGENCY_GENERATOR_2_MODE', 'EMERGENCY_GENERATOR_2_START_STOP', 'FREIGHT_PUMP_CONDENSER_SWITCH', 'FREIGHT_PUMP_EXTERNAL_SWITCH', 'FREIGHT_PUMP_FEEDWATER_SWITCH', 'FREIGHT_PUMP_INTERNAL_SWITCH', 'FUN_AO_SABOTAGE_ONCE', 'FUN_AO_SABOTAGE_TIME', 'FUN_BANK_ROBBERY', 'FUN_BREAKER_TRIP', 'FUN_DECREASE_INTEGRITY', 'FUN_FIRE_DRILL', 'FUN_IODINE_SPILL', 'FUN_OIL_SPILL', 'FUN_PUMP_JAM', 'FUN_REQUEST_ENABLE', 'FUN_SHOW_MESSAGE', 'FUN_TOGGLE_RANDOM_SWITCH', 'FUN_TRIGGER_AUDIT', 'FUN_WEATHER_CONTROL', 'FUN_XENON_SPILL', 'MSCV_0_OPENING_ORDERED', 'MSCV_1_OPENING_ORDERED', 'MSCV_2_OPENING_ORDERED', 'RESET_AO', 'RESISTOR_BANK_01_SWITCH', 'RESISTOR_BANK_02_SWITCH', 'RESISTOR_BANK_03_SWITCH', 'RESISTOR_BANK_04_SWITCH', 'RESISTOR_BANKS_MAIN_SWITCH', 'ROD_BANK_POS_0_ORDERED', 'ROD_BANK_POS_1_ORDERED', 'ROD_BANK_POS_2_ORDERED', 'ROD_BANK_POS_3_ORDERED', 'ROD_BANK_POS_4_ORDERED', 'ROD_BANK_POS_5_ORDERED', 'ROD_BANK_POS_6_ORDERED', 'ROD_BANK_POS_7_ORDERED', 'ROD_BANK_POS_8_ORDERED', 'RODS_ALL_POS_ORDERED', 'STEAM_EJECTOR_CONDENSER_RETURN_VALVE', 'STEAM_EJECTOR_OPERATIONAL_MOTIVE_VALVE', 'STEAM_EJECTOR_STARTUP_MOTIVE_VALVE', 'STEAM_GEN_0_VENT_SWITCH', 'STEAM_GEN_1_VENT_SWITCH', 'STEAM_GEN_2_VENT_SWITCH', 'STEAM_TURBINE_0_BYPASS_ORDERED', 'STEAM_TURBINE_1_BYPASS_ORDERED', 'STEAM_TURBINE_2_BYPASS_ORDERED', 'STEAM_TURBINE_TRIP', 'VALVE_CLOSE', 'VALVE_OFF', 'VALVE_OPEN'] # ===================== # Helpers # ===================== def build_base_url(host: str, port: int) -> str: return f"http://{host}:{port}/" 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 http_get_root(base_url: str) -> Tuple[int, str, Dict[str, str]]: req = urllib.request.Request(base_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 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 parse_variable_names_from_html_index(html_text: str) -> List[str]: 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"([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] # ---- 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).""" try: data = json.loads(body) except Exception: 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"): names = extract_names(v) if key == "GET": 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 def extract_names(value_obj) -> List[str]: out: List[str] = [] def add(n): if isinstance(n, str): u = n.strip().upper() if u and u not in out: out.append(u) if isinstance(value_obj, list): for it in value_obj: if isinstance(it, str): add(it) elif isinstance(it, dict): for k in ("name","variable","var","id","func","function"): if k in it and isinstance(it[k], str): add(it[k]) for k in list(it.keys()): if isinstance(k, str) and k.isupper(): add(k) elif isinstance(value_obj, dict): for k in value_obj.keys(): if isinstance(k, str): add(k) elif isinstance(value_obj, str): for m in re.findall(r"[A-Z0-9_]+", value_obj.upper()): add(m) return out # ===================== # Data model # ===================== @dataclass class Thresholds: dead_low: Optional[float] = None low: Optional[float] = None high: Optional[float] = None extreme_high: Optional[float] = None # alarms alarm_dead_low: bool = False alarm_low: bool = False alarm_high: bool = False alarm_extreme_high: bool = True # function actions 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" # expression actions (NEW): evaluated with x = current numeric value expr_dead_low: Optional[str] = None expr_low: Optional[str] = None expr_high: Optional[str] = None expr_extreme_high: Optional[str] = None @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() # Functions should be POST per your spec return http_post(base, {"variable": name, "value": value}) def run(self) -> None: while not self._stop.is_set(): now = time.time() to_run: List[ActionTask] = [] 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() # ===================== # Poller thread (aggregated) # ===================== class Poller(threading.Thread): def __init__( self, host: str, port: int, variables: List[str], ui_queue: queue.Queue, refresh_interval: int, 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.stop_event = stop_event self.paused_event = paused_event def run(self) -> None: base_url = build_base_url(self.host, self.port) while not self.stop_event.is_set(): 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"}) 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 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))) else: # Bulk endpoint failed -> per-variable fallback 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))) # sleep until next cycle, unless paused/stopped 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 while self.paused_event.is_set() and not self.stop_event.is_set(): time.sleep(0.1) time.sleep(0.1) self.ui_queue.put(("stopped", "")) # ===================== # GUI # ===================== class App(tk.Tk): def __init__(self) -> None: super().__init__() self.title("Control Board Monitor — v0.7") self.geometry("1250x820") 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] = [] self.scheduler = ActionScheduler(lambda: build_base_url(self.host_var.get().strip(), int(self.port_var.get()))) self.scheduler.start() 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) 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)) self.status_lbl = ttk.Label(controls, text="Idle", foreground="#666") self.status_lbl.grid(row=0, column=6, padx=(10, 0)) btns = ttk.Frame(controls) btns.grid(row=0, column=7, 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) main = ttk.Panedwindow(self, orient=tk.VERTICAL) main.pack(side=tk.TOP, fill=tk.BOTH, expand=True, padx=10, pady=(0,10)) 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_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") self.tree_menu = tk.Menu(self, tearoff=False) self.tree_menu.add_command(label="Set thresholds, actions & alarms…", command=self.open_thresholds_dialog) self.tree.bind("", self.on_tree_right_click) 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 = ("func", "value", "mode", "interval", "next", "enabled") self.actions_tree = ttk.Treeview(bottom, columns=columns2, show="headings", height=10) for col, hdr, w in zip(columns2, ["Function", "Value", "Mode", "Interval(s)", "Next Run", "Enabled"], [260, 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) 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 = 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) self.functions_list = list(DEFAULT_FUNCTIONS) self.func_combo["values"] = self.functions_list # --- Discovery --- 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 = [] # 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())) else: msgs.append(f"LIST_JSON HTTP {status}: {coerce_preview(body, 120)}") # 2) Fallback to main page parse if not discovered_vars or not discovered_funcs: s2, b2, _2 = http_get_root(base) if s2 == 200 and b2: if not discovered_vars: discovered_vars = parse_variable_names_from_html_index(b2) or discovered_vars if not discovered_funcs: discovered_funcs = parse_function_names_from_html_index(b2) or discovered_funcs else: msgs.append(f"Main page HTTP {s2}: {coerce_preview(b2, 120)}") # 3) Final fallback to embedded defaults 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) # Status if used_defaults: messagebox.showwarning( "Discovery fallback", "Could not discover variables/functions from WEBSERVER_LIST_VARIABLES_JSON or main page.\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: self.status_lbl.configure(text=f"Discovered: vars {len(discovered_vars)} | funcs {len(discovered_funcs)}") # --- Select Vars dialog --- def open_selector(self) -> None: union = sorted(set(self.variables_list) | set(self.vars.keys())) if not union: 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("", 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) # --- Actions UI --- 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 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"POST: {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("360x170"); 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 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"POST: {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.name, t.value, mode, t.interval_s, next_s, "Yes" if t.enabled else "No")) # --- Poller control / rendering --- def start_polling(self) -> None: if hasattr(self, "poller") and 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()), 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) # --- Threshold logic & rendering --- 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 not name or "_" not in name: messagebox.showinfo("Select variable", "Select a specific variable row (not a group).") return 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, actions & alarms — {name}") dlg.geometry("780x580") 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) # numeric thresholds 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 # function actions ttk.Label(grid, text="On enter: call function (optional)").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 # expression actions ttk.Label(grid, text="On enter: evaluate expression/snippet with x = current numeric value (optional)").grid(row=12, column=0, sticky="w", pady=(14,4), columnspan=3) expr_vars = { "expr_dead_low": tk.StringVar(value=t.expr_dead_low or ""), "expr_low": tk.StringVar(value=t.expr_low or ""), "expr_high": tk.StringVar(value=t.expr_high or ""), "expr_extreme_high": tk.StringVar(value=t.expr_extreme_high or ""), } for i, (label, key) in enumerate([("Dead Low expr", "expr_dead_low"), ("Low expr", "expr_low"), ("High expr", "expr_high"), ("Extreme High expr", "expr_extreme_high")]): ttk.Label(grid, text=label).grid(row=13+i, column=0, sticky="w", pady=2) ttk.Entry(grid, textvariable=expr_vars[key], width=56).grid(row=13+i, column=1, sticky="w", padx=6, columnspan=2) 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(), expr_dead_low=expr_vars["expr_dead_low"].get().strip() or None, expr_low=expr_vars["expr_low"].get().strip() or None, expr_high=expr_vars["expr_high"].get().strip() or None, expr_extreme_high=expr_vars["expr_extreme_high"].get().strip() or None, ) 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) 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, x) vi.last_state = new_state def on_enter_state(self, name: str, state: str, vi: VarInfo, x: Optional[float]) -> None: t = vi.thresholds # Alarms 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): try: import winsound freq = {"DEAD_LOW": 300, "LOW": 550, "HIGH": 800, "EXTREME_HIGH": 1000}.get(state, 600) dur = 350 if state != "EXTREME_HIGH" else 600 winsound.Beep(freq, dur) except Exception: try: self.bell() except Exception: pass # Expression takes precedence if provided expr_key = { "DEAD_LOW": "expr_dead_low", "LOW": "expr_low", "HIGH": "expr_high", "EXTREME_HIGH": "expr_extreme_high" }.get(state) expr = getattr(t, expr_key) if expr_key else None if expr and x is not None: try: computed = eval_user_expression(expr, x) # Send result as new value for this variable base = build_base_url(self.host_var.get().strip(), int(self.port_var.get())) http_post(base, {"variable": name, "value": str(computed)}) except Exception as e: messagebox.showwarning("Expression error", f"{name}: failed to evaluate expression on enter {state}: {e}") # fall through to function action # Function action 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: # one-shot task 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 = "neutral" if vi.error: tag = "bad" else: tag = "ok" if value_preview not in ("",) else "neutral" 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,), ) def on_close(self) -> None: try: if self.poller and self.poller.is_alive(): self.stop_event.set() except Exception: pass try: self.scheduler.stop() except Exception: pass self.destroy() # --- Utils --- 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 def eval_user_expression(expr: str, x: float): """Evaluate user expression/snippet with x available. Allowed: math.*, min, max, abs, round, int, float, clamp (custom). Returns the computed result. """ def clamp(v, lo, hi): return max(lo, min(hi, v)) safe_globals = {"__builtins__": {}, "math": math, "min": min, "max": max, "abs": abs, "round": round, "int": int, "float": float, "clamp": clamp} safe_locals = {"x": x} # If it's multiline or contains statements, try exec and read 'result' if "\n" in expr or ";" in expr: code = compile(expr, "", "exec") exec(code, safe_globals, safe_locals) if "result" not in safe_locals: raise ValueError("Snippet must assign to 'result', e.g., result = x * 1.1") return safe_locals["result"] else: code = compile(expr, "", "eval") return eval(code, safe_globals, safe_locals) def main() -> None: app = App() app.mainloop() if __name__ == "__main__": main()