#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Control Board Monitor — v1.2 What’s new: - POSTs are now **exactly** like the sample app: POST http://:/?variable=&value= (method=POST, no body) (All previous POST modes removed.) - Select Variables dialog: typing in the filter **does not clear** previously ticked checkboxes. - Mouse wheel scrolling enabled (selector canvas, main variables tree, actions tree). """ 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 = 1 REQUEST_TIMEOUT_S: float = 5.0 USER_AGENT: str = "ControlBoardMonitor/1.2 (+tkinter)" # 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'] # One global lock to serialize every HTTP call _HTTP_LOCK = threading.Lock() # ===================== # Helpers # ===================== def build_base_url(host: str, port: int) -> str: return f"http://{host}:{port}/" def _request(req: urllib.request.Request) -> Tuple[int, str, Dict[str, str]]: with _HTTP_LOCK: 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(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, "Connection": "close"}) return _request(req) def http_post_query(base_url: str, params: Dict[str, str]) -> Tuple[int, str, Dict[str, str]]: # EXACT behavior requested: POST with query string, no body. url = base_url + "?" + urllib.parse.urlencode(params) req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Connection": "close"}, method="POST") return _request(req) def http_get_root(base_url: str) -> Tuple[int, str, Dict[str, str]]: req = urllib.request.Request(base_url, headers={"User-Agent": USER_AGENT, "Connection": "close"}) return _request(req) 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_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) return uniq 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", []) + qs.get("Variable", []): tokens.append(v) for m in re.finditer(r'(?:Variable|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: if v and v.upper() not in {"VARNAME", "VARIABLE", "NAME"} and v not in seen: seen.add(v); uniq.append(v) return uniq # ---- Parse WEBSERVER_LIST_VARIABLES_JSON ---- def parse_weblist_names(body: str) -> Tuple[List[str], List[str]]: """Return (get_list, post_list). Keep discovered case for function names.""" try: data = json.loads(body) except Exception: return [], [] get_names: List[str] = [] post_names: List[str] = [] if isinstance(data, dict): for k, v in data.items(): key = str(k).strip().rstrip(":").upper() if key in ("GET", "POST"): names = extract_names_preserve_case(v) if key == "GET": get_names.extend(names) else: post_names.extend(names) elif isinstance(data, list): get_names.extend(extract_names_preserve_case(data)) 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) def extract_names_preserve_case(value_obj) -> List[str]: out: List[str] = [] def add(n): if isinstance(n, str) and n and n not in out: out.append(n) 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.upper() == k: 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-Za-z0-9_]+", value_obj): add(m) return out # ---- Parse WEBSERVER_BATCH_GET ---- def parse_batch_values(body: str) -> Dict[str, str]: try: data = json.loads(body) except Exception: return {} values: Dict[str, str] = {} if isinstance(data, dict): for k, v in data.items(): if not isinstance(k, str): k = str(k) try: values[k] = v if isinstance(v, str) else json.dumps(v) except Exception: values[k] = str(v) return values # ===================== # Data model # ===================== @dataclass class Thresholds: dead_low: Optional[float] = None low: Optional[float] = None high: Optional[float] = None extreme_high: Optional[float] = None # alarms (defaults: on for dead_low & extreme_high as requested) alarm_dead_low: bool = True 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" # interval while-in-state (0 = once on enter) action_dead_low_interval: float = 1.0 action_low_interval: float = 1.0 action_high_interval: float = 1.0 action_extreme_high_interval: float = 1.0 # expression actions expr_dead_low: Optional[str] = None expr_low: Optional[str] = None expr_high: Optional[str] = None expr_extreme_high: Optional[str] = None # expression target function (per state) expr_target_dead_low: Optional[str] = None expr_target_low: Optional[str] = None expr_target_high: Optional[str] = None expr_target_extreme_high: Optional[str] = None @dataclass class VarInfo: key: str # canonical lower-case key display_name: str # original case as chosen 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: float 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() params = {"variable": name, "value": value} return http_post_query(base, params) 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 (uses BATCH_GET) # ===================== class Poller(threading.Thread): def __init__( self, host: str, port: int, variables_keys: List[str], # canonical lower-case keys ui_queue: queue.Queue, refresh_interval: float, stop_event: threading.Event, paused_event: threading.Event, ) -> None: super().__init__(daemon=True) self.host = host self.port = port self.variables_keys = variables_keys self.ui_queue = ui_queue self.refresh_interval = float(max(0.1, 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"))) status, body, _ = http_get(base_url, {"variable": "WEBSERVER_BATCH_GET"}) if status == 200: values_map = parse_batch_values(body) if values_map: lower_map = {k.lower(): v for k, v in values_map.items()} # push full batch to UI for 'y' lookups self.ui_queue.put(("batch", lower_map)) for key in list(self.variables_keys): if key in lower_map: self.ui_queue.put(("update", key, str(lower_map[key]), 200)) else: self.ui_queue.put(("error", key, 206, "Not in BATCH_GET payload")) else: for key in list(self.variables_keys): st, b, _h = http_get(base_url, {"variable": key}) if st == 200: self.ui_queue.put(("update", key, b, st)) else: self.ui_queue.put(("error", key, st, coerce_preview(b, 200))) else: for key in list(self.variables_keys): st, b, _h = http_get(base_url, {"variable": key}) if st == 200: self.ui_queue.put(("update", key, b, st)) else: self.ui_queue.put(("error", key, st, coerce_preview(b, 200))) 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 — v1.2") self.geometry("1280x860") # Canonical registry: lower-case key -> VarInfo 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_keys: List[str] = [] # list of lower-case keys in selection order self.functions_list: List[str] = [] # preserve original casing 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.DoubleVar(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=20, selectmode="browse") self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) vsb = ttk.Scrollbar(top_frame, orient="vertical", command=self.tree.yview) vsb.pack(side=tk.RIGHT, fill=tk.Y) self.tree.configure(yscrollcommand=vsb.set) self._bind_mousewheel(self.tree) 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=540, 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 # Context menu 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=44, 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, selectmode="browse") for col, hdr, w in zip(columns2, ["Function", "Value", "Mode", "Interval(s)", "Next Run", "Enabled"], [280, 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.LEFT, fill=tk.BOTH, expand=True) vsb2 = ttk.Scrollbar(bottom, orient="vertical", command=self.actions_tree.yview) vsb2.pack(side=tk.RIGHT, fill=tk.Y) self.actions_tree.configure(yscrollcommand=vsb2.set) self._bind_mousewheel(self.actions_tree) 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) # Menubar menubar = tk.Menu(self) filemenu = tk.Menu(menubar, tearoff=False) filemenu.add_command(label="Save Configuration...", command=self.save_config) filemenu.add_command(label="Load Configuration...", command=self.load_config) filemenu.add_separator() filemenu.add_command(label="Exit", command=self.on_close) menubar.add_cascade(label="File", menu=filemenu) funmenu = tk.Menu(menubar, tearoff=False) funmenu.add_command(label="Run Function Once...", command=self.menu_run_once_dialog) funmenu.add_command(label="Add Scheduled Function...", command=self.menu_schedule_dialog) funmenu.add_separator() funmenu.add_command(label="Reload Discovery", command=self.reload_discovery) funmenu.add_command(label="Select Variables…", command=self.open_selector) self.config(menu=menubar) self.functions_list = list(DEFAULT_FUNCTIONS) self.func_combo["values"] = self.functions_list if self.functions_list and not self.func_name_var.get(): self.func_name_var.set(self.functions_list[0]) # Stable IDs self._group_ids: Dict[str, str] = {} self._var_ids: Dict[str, str] = {} # Color tag cache: color_hex -> tag_name self._color_tags: Dict[str, str] = {} # Track repeating actions per (var_key,state) self.state_tasks: Dict[tuple, int] = {} # Latest values from batch (lowercased keys) self.latest_values: Dict[str, str] = {} # Trigger discovery shortly after startup so function list populates self.after(200, self.reload_discovery) # Live watch for refresh interval changes self._last_refresh_val = float(self.refresh_var.get()) self.after(2000, self._watch_refresh_interval) # Actions 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) # --- 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) Functions via WEBSERVER_LIST_VARIABLES_JSON (POST list) s1, b1, _1 = http_get(base, {"variable": "WEBSERVER_LIST_VARIABLES_JSON"}) if s1 == 200 and b1: _get_list, post_list = parse_weblist_names(b1) if post_list: discovered_funcs.extend(post_list) else: msgs.append(f"LIST_VARIABLES_JSON HTTP {s1}: {coerce_preview(b1,120)}") # 2) Variables via WEBSERVER_BATCH_GET keys s2, b2, _2 = http_get(base, {"variable": "WEBSERVER_BATCH_GET"}) if s2 == 200 and b2: values_map = parse_batch_values(b2) if values_map: seen=set() for k in values_map.keys(): kl = k.lower() if kl not in seen: seen.add(kl); discovered_vars.append(k) else: msgs.append(f"BATCH_GET HTTP {s2}: {coerce_preview(b2,120)}") # 3) Fallbacks if not discovered_vars or not discovered_funcs: s3, b3, _3 = http_get_root(base) if s3 == 200 and b3: if not discovered_vars: discovered_vars = parse_variable_names_from_html_index(b3) or discovered_vars if not discovered_funcs: discovered_funcs = parse_function_names_from_html_index(b3) or discovered_funcs else: msgs.append(f"Main page HTTP {s3}: {coerce_preview(b3,120)}") # 4) 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 if self.functions_list and not self.func_name_var.get(): self.func_name_var.set(self.functions_list[0]) self.show_selector_dialog(discovered_vars) # Status if used_defaults: messagebox.showwarning( "Discovery fallback", "Could not fully discover via LIST_VARIABLES_JSON / BATCH_GET.\n" "Loaded embedded defaults so you can proceed.\n\n" + ("\n".join(msgs[:6]) if msgs else "")) self.status_lbl.configure(text=f"Loaded defaults ({len(discovered_vars)} vars, {len(discovered_funcs)} funcs)") else: self.status_lbl.configure(text=f"Discovered: vars {len(discovered_vars)} | funcs {len(discovered_funcs)}") # --- Select Vars dialog --- def open_selector(self) -> None: union = set([vi.display_name for vi in self.vars.values()]) or set(DEFAULT_VARS) self.show_selector_dialog(sorted(union)) def show_selector_dialog(self, discovered: List[str]) -> None: seen=set(); disp=[] for n in discovered: kl = n.lower() if kl not in seen: seen.add(kl); disp.append(n) dlg = tk.Toplevel(self) dlg.title("Select Variables to Monitor") dlg.geometry("600x540") dlg.transient(self) dlg.grab_set() temp_selected = {k: True for k in self.variables_keys} 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") def _mw(event): if event.delta: canvas.yview_scroll(int(-1*(event.delta/120)), "units") return "break" def _mw_up(_e): canvas.yview_scroll(-1, "units"); return "break" def _mw_down(_e): canvas.yview_scroll(1, "units"); return "break" canvas.bind("", _mw) inner.bind("", _mw) canvas.bind("", _mw_up) canvas.bind("", _mw_down) inner.bind("", _mw_up) inner.bind("", _mw_down) rows: List[Tuple[str, str, tk.BooleanVar]] = [] def rebuild(): for child in inner.winfo_children(): child.destroy() rows.clear() f = filt_var.get().strip().lower() for name in disp: if f and f not in name.lower(): continue key = name.lower() var = tk.BooleanVar(value=temp_selected.get(key, False)) def bind_trace(k=key, v=var): def _(*_a): temp_selected[k] = v.get() v.trace_add("write", _) bind_trace() cb = ttk.Checkbutton(inner, text=name, variable=var) cb.pack(anchor="w", pady=2, padx=4) rows.append((name, key, var)) rebuild() filt_var.trace_add("write", lambda *_: rebuild()) def select_all(): for _, key, var in rows: var.set(True); temp_selected[key] = True def clear_all(): for _, key, var in rows: var.set(False); temp_selected[key] = False sel_all.configure(command=select_all) sel_none.configure(command=clear_all) bot = ttk.Frame(dlg, padding=8) bot.pack(side=tk.BOTTOM, fill=tk.X) def apply_and_close(): selected_keys = [k for k, sel in temp_selected.items() if sel] new_order = [k for k in self.variables_keys if k in selected_keys] for name in disp: key = name.lower() if temp_selected.get(key, False) and key not in new_order: new_order.append(key) self.variables_keys = new_order new_vars: Dict[str, VarInfo] = {} for key in self.variables_keys: if key in self.vars: vi = self.vars[key] else: disp_name = next((n for n, k, _ in rows if k == key), key) vi = VarInfo(key=key, display_name=disp_name) new_vars[key] = vi self.vars = new_vars # If poller is running, update its live list if self.poller and self.poller.is_alive(): self.poller.variables_keys = self.variables_keys[:] 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 ?variable={fname}&value={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.DoubleVar(value=1.0) ttk.Label(top, text="Interval (seconds):").pack(pady=(6,0)) ttk.Entry(top, textvariable=int_var, width=10).pack() def add_it(): interval = float(int_var.get()) task = ActionTask(name=fname, value=val_var.get(), interval_s=max(0.0, interval)) 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 ?variable={t.name}&value={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_keys: 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_keys=self.variables_keys, ui_queue=self.ui_queue, refresh_interval=float(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": _, key, value, status = item vi = self.vars.get(key) if not vi: vi = VarInfo(key=key, display_name=key) vi.last_value = value vi.last_updated = time.time() vi.last_status = status vi.error = None self.vars[key] = vi self.evaluate_thresholds(vi, value) elif kind == "error": _, key, status, msg = item vi = self.vars.get(key) or VarInfo(key=key, display_name=key) vi.last_status = status vi.error = msg vi.last_updated = time.time() self.vars[key] = vi elif kind == "cycle_start": _, hhmmss = item self.cycle_lbl.configure(text=f"Last cycle start: {hhmmss}") elif kind == "batch": _, mapping = item if isinstance(mapping, dict): self.latest_values.update(mapping) 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 iid = sel[0] if not iid.startswith("var:"): messagebox.showinfo("Select variable", "Select a specific variable row (not a group).") return key = iid.split("var:",1)[1] vi = self.vars.get(key) if not vi: messagebox.showinfo("Missing", "Selected variable is not available.") return dlg = tk.Toplevel(self) dlg.title(f"Thresholds, actions & alarms — {vi.display_name}") dlg.geometry("980x680") dlg.transient(self); dlg.grab_set() t = vi.thresholds # Auto-defaults based on current value if thresholds are empty cur = parse_first_float(vi.last_value) auto_defaults = {} if cur is not None: auto_defaults = { "dead_low": round(cur * 0.8, 6), "low": round(cur * 0.9, 6), "high": round(cur * 1.1, 6), "extreme_high": round(cur * 1.2, 6), } grid = ttk.Frame(dlg, padding=10) grid.pack(side=tk.TOP, fill=tk.BOTH, expand=True) entries = {} def add_row(r, label, keyname, default): ttk.Label(grid, text=label).grid(row=r, column=0, sticky="w", pady=4) dv = auto_defaults.get(keyname) if default is None else default sv = tk.StringVar(value="" if dv is None else str(dv)) e = ttk.Entry(grid, textvariable=sv, width=12) e.grid(row=r, column=1, sticky="w", padx=6) entries[keyname] = 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, keyname, 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[keyname] = var 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", "action_dead_low_interval"), ("Low", "action_low", "value_low", "action_low_interval"), ("High", "action_high", "value_high", "action_high_interval"), ("Extreme High", "action_extreme_high", "value_extreme_high", "action_extreme_high_interval")] action_vars = {} value_vars = {} interval_vars = {} ttk.Label(grid, text="Interval(s) while in state (0 = once)").grid(row=7, column=3, sticky="w") for i, (label, action_key, value_key, interval_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=44, 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 iv = tk.DoubleVar(value=getattr(t, interval_key) if getattr(t, interval_key) is not None else 1.0) ttk.Entry(grid, textvariable=iv, width=10).grid(row=8+i, column=3, sticky="w", padx=6) interval_vars[interval_key] = iv ttk.Label(grid, text="On enter: evaluate expression/snippet (x = current value) and optionally POST to a function").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 ""), } expr_target_vars = { "expr_target_dead_low": tk.StringVar(value=t.expr_target_dead_low or ""), "expr_target_low": tk.StringVar(value=t.expr_target_low or ""), "expr_target_high": tk.StringVar(value=t.expr_target_high or ""), "expr_target_extreme_high": tk.StringVar(value=t.expr_target_extreme_high or ""), } for i, (label, keyname, tkey) in enumerate([("Dead Low expr", "expr_dead_low", "expr_target_dead_low"), ("Low expr", "expr_low", "expr_target_low"), ("High expr", "expr_high", "expr_target_high"), ("Extreme High expr", "expr_extreme_high", "expr_target_extreme_high")]): ttk.Label(grid, text=label).grid(row=13+i, column=0, sticky="w", pady=2) ttk.Entry(grid, textvariable=expr_vars[keyname], width=64).grid(row=13+i, column=1, sticky="w", padx=6) ttk.Label(grid, text="→ POST to function").grid(row=13+i, column=2, sticky="e") s = ttk.Combobox(grid, values=self.functions_list or DEFAULT_FUNCTIONS, width=32, state="readonly") s.set(expr_target_vars[tkey].get()) s.grid(row=13+i, column=3, sticky="w", padx=6) def _on_combo_select(event, varkey=tkey, combo=s): expr_target_vars[varkey].set(combo.get()) s.bind("<>", _on_combo_select) 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(), action_dead_low_interval=float(interval_vars["action_dead_low_interval"].get() or 1.0), action_low_interval=float(interval_vars["action_low_interval"].get() or 1.0), action_high_interval=float(interval_vars["action_high_interval"].get() or 1.0), action_extreme_high_interval=float(interval_vars["action_extreme_high_interval"].get() or 1.0), 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, expr_target_dead_low=expr_target_vars["expr_target_dead_low"].get().strip() or None, expr_target_low=expr_target_vars["expr_target_low"].get().strip() or None, expr_target_high=expr_target_vars["expr_target_high"].get().strip() or None, expr_target_extreme_high=expr_target_vars["expr_target_extreme_high"].get().strip() or None, ) vi.thresholds = nt self.vars[vi.key] = 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) # --- State evaluation per spec + color gradients --- def evaluate_thresholds(self, vi: VarInfo, value: str) -> None: x = parse_first_float(value) t = vi.thresholds # Auto-defaults based on current value if thresholds are empty cur = parse_first_float(vi.last_value) auto_defaults = {} if cur is not None: auto_defaults = { "dead_low": round(cur * 0.8, 6), "low": round(cur * 0.9, 6), "high": round(cur * 1.1, 6), "extreme_high": round(cur * 1.2, 6), } prev = vi.last_state state = "UNKNOWN" if x is None or (t.low is None and t.high is None and t.dead_low is None and t.extreme_high is None): state = "OPERATING" else: dl = t.dead_low if t.dead_low is not None else -float("inf") lo = t.low if t.low is not None else -float("inf") hi = t.high if t.high is not None else float("inf") ex = t.extreme_high if t.extreme_high is not None else float("inf") if x < dl: state = "DEAD_LOW" elif x < lo: state = "LOW" elif x < hi: state = "OPERATING" elif x < ex: state = "HIGH" else: state = "EXTREME_HIGH" if state != prev: self.on_state_change(vi, prev, state, x) vi.last_state = state def on_state_change(self, vi: VarInfo, prev_state: str, new_state: str, x: Optional[float]) -> None: # stop any repeating task associated with previous state tid = self.state_tasks.pop((vi.key, prev_state), None) if tid is not None: try: self.scheduler.remove_task(tid) except Exception: pass # then handle enter actions for new_state self.on_enter_state(vi, new_state, x) def on_enter_state(self, vi: VarInfo, state: str, x: Optional[float]) -> None: t = vi.thresholds # Auto-defaults based on current value if thresholds are empty cur = parse_first_float(vi.last_value) auto_defaults = {} if cur is not None: auto_defaults = { "dead_low": round(cur * 0.8, 6), "low": round(cur * 0.9, 6), "high": round(cur * 1.1, 6), "extreme_high": round(cur * 1.2, 6), } alarm_map = { "DEAD_LOW": t.alarm_dead_low, "LOW": t.alarm_low, "OPERATING": False, "HIGH": t.alarm_high, "EXTREME_HIGH": t.alarm_extreme_high, } if alarm_map.get(state): try: import winsound freq = {"DEAD_LOW": 350, "LOW": 550, "HIGH": 800, "EXTREME_HIGH": 1000}.get(state, 600) dur = 400 if state in ("DEAD_LOW", "EXTREME_HIGH") else 300 winsound.Beep(freq, dur) except Exception: try: self.bell() except Exception: pass expr_key = { "DEAD_LOW": "expr_dead_low", "LOW": "expr_low", "HIGH": "expr_high", "EXTREME_HIGH": "expr_extreme_high" }.get(state) target_key = { "DEAD_LOW": "expr_target_dead_low", "LOW": "expr_target_low", "HIGH": "expr_target_high", "EXTREME_HIGH": "expr_target_extreme_high" }.get(state) expr = getattr(t, expr_key) if expr_key else None target_func = getattr(t, target_key) if target_key else None # Determine y (current value of the variable we intend to change) y_val = None y_source = None if target_func: y_source = target_func.lower() else: # fallback to action_* target name if set action_map = {"DEAD_LOW": t.action_dead_low, "LOW": t.action_low, "HIGH": t.action_high, "EXTREME_HIGH": t.action_extreme_high} fallback = action_map.get(state) if fallback: y_source = fallback.lower() if y_source: # try latest batch first if hasattr(self, "latest_values") and y_source in self.latest_values: y_val = parse_first_float(self.latest_values.get(y_source)) else: # try direct GET base = build_base_url(self.host_var.get().strip(), int(self.port_var.get())) st, bd, _h = http_get(base, {"variable": y_source}) if st == 200: y_val = parse_first_float(bd) if expr and x is not None: try: computed = eval_user_expression(expr, x, y_val) if target_func: base = build_base_url(self.host_var.get().strip(), int(self.port_var.get())) params = {"variable": target_func, "value": str(computed)} http_post_query(base, params) else: self.status_lbl.configure(text=f"Expr computed for {vi.display_name} [{state}], no target function set") except Exception as e: messagebox.showwarning("Expression error", f"{vi.display_name}: expression failed on enter {state}: {e}") 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: # Once on enter if interval == 0 interval_map = { "DEAD_LOW": t.action_dead_low_interval, "LOW": t.action_low_interval, "HIGH": t.action_high_interval, "EXTREME_HIGH": t.action_extreme_high_interval, } interval = float(interval_map.get(state, 0) or 0.0) if interval <= 0.0: self.scheduler.add_task(ActionTask(name=act, value=val, interval_s=0.0)) else: tid = self.scheduler.add_task(ActionTask(name=act, value=val, interval_s=max(0.0, interval))) self.state_tasks[(vi.key, state)] = tid # --- Tree helpers & rendering --- def _ensure_group(self, group: str) -> str: iid = self._group_ids.get(group) if iid and self.tree.exists(iid): return iid iid = f"grp:{group}" self._group_ids[group] = iid if not self.tree.exists(iid): self.tree.insert("", "end", iid=iid, text=group, values=("", "", ""), open=True) return iid def _ensure_var(self, group_iid: str, key: str, display_name: str) -> str: iid = self._var_ids.get(key) if iid and self.tree.exists(iid): return iid iid = f"var:{key}" self._var_ids[key] = iid if not self.tree.exists(iid): self.tree.insert(group_iid, "end", iid=iid, text=display_name, values=("", "", "")) return iid def _get_color_tag(self, hex_color: str) -> str: tag = self._color_tags.get(hex_color) if tag: return tag tag = f"col_{hex_color[1:]}" self._color_tags[hex_color] = tag try: self.tree.tag_configure(tag, background=hex_color) except Exception: pass return tag def _state_color(self, vi: VarInfo, x: Optional[float]) -> str: t = vi.thresholds # Auto-defaults based on current value if thresholds are empty cur = parse_first_float(vi.last_value) auto_defaults = {} if cur is not None: auto_defaults = { "dead_low": round(cur * 0.8, 6), "low": round(cur * 0.9, 6), "high": round(cur * 1.1, 6), "extreme_high": round(cur * 1.2, 6), } if x is None: return "#f5f7fb" def ci(v): return max(0, min(255, int(v))) if t.low is not None and t.high is not None and t.dead_low is not None and t.extreme_high is not None: if x < t.dead_low: return "#FFC266" elif x < t.low: span = max(1e-9, t.low - t.dead_low) frac = 1.0 - (x - t.dead_low) / span r = 255 g = 248 - frac * (248 - 226) b = 204 - frac * (204 - 138) return f"#{ci(r):02X}{ci(g):02X}{ci(b):02X}" elif x < t.high: return "#E6F5E6" elif x < t.extreme_high: span = max(1e-9, t.extreme_high - t.high) frac = (x - t.high) / span r = 255 g = 214 - frac * (214 - 158) b = 214 - frac * (214 - 158) return f"#{ci(r):02X}{ci(g):02X}{ci(b):02X}" else: return "#FF6B6B" return "#E6F5E6" def refresh_tree(self) -> None: filt = self.filter_var.get().strip().lower() groups: Dict[str, List[VarInfo]] = {} for key, vi in self.vars.items(): text = vi.display_name if filt and filt not in text.lower() and (not vi.last_value or filt not in vi.last_value.lower()): continue prefix = text.split("_", 1)[0] if "_" in text else "MISC" groups.setdefault(prefix, []).append(vi) sel = self.tree.selection() sel_iid = sel[0] if sel else None valid_var_iids = set() valid_group_iids = set() for g in sorted(groups.keys(), key=lambda s: s.lower()): gid = self._ensure_group(g) valid_group_iids.add(gid) for vi in sorted(groups[g], key=lambda x: x.display_name.lower()): vid = self._ensure_var(gid, vi.key, vi.display_name) valid_var_iids.add(vid) 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) x = parse_first_float(vi.last_value) color = self._state_color(vi, x) tag = self._get_color_tag(color) self.tree.item(vid, text=vi.display_name, values=(value_preview, updated_str, vi.last_status or "—"), tags=(tag,)) for key, vid in list(self._var_ids.items()): if vid not in valid_var_iids and self.tree.exists(vid): self.tree.delete(vid); self._var_ids.pop(key, None) for g, gid in list(self._group_ids.items()): if gid not in valid_group_iids and self.tree.exists(gid): self.tree.delete(gid); self._group_ids.pop(g, None) if sel_iid and self.tree.exists(sel_iid): self.tree.selection_set(sel_iid) def _bind_mousewheel(self, widget): def _mw(event): if event.delta: widget.yview_scroll(int(-1*(event.delta/120)), "units") return "break" def _mw_up(_e): widget.yview_scroll(-1, "units"); return "break" def _mw_down(_e): widget.yview_scroll(1, "units"); return "break" widget.bind("", _mw) widget.bind("", _mw_up) widget.bind("", _mw_down) def _watch_refresh_interval(self): try: val = float(self.refresh_var.get()) except Exception: val = self._last_refresh_val if val != self._last_refresh_val: self._last_refresh_val = val if self.poller and self.poller.is_alive(): try: self.poller.refresh_interval = max(0.1, val) self.status_lbl.configure(text=f"Running… (refresh {self.poller.refresh_interval}s)") except Exception: pass self.after(2000, self._watch_refresh_interval) def menu_run_once_dialog(self): top = tk.Toplevel(self); top.title("Run Function Once"); top.geometry("420x180"); top.transient(self); top.grab_set() ttk.Label(top, text="Function:").pack(pady=(10,2)) name = tk.StringVar(value=(self.functions_list[0] if self.functions_list else "")) cmb = ttk.Combobox(top, values=self.functions_list, textvariable=name, width=44, state="readonly") cmb.pack() ttk.Label(top, text="Value:").pack(pady=(6,2)) val = tk.StringVar(value="1") ttk.Entry(top, textvariable=val, width=16).pack() def go(): fname = name.get().strip() if not fname: return status, body, _ = self.scheduler.run_once(fname, val.get()) messagebox.showinfo("Run Once", f"POST ?variable={fname}&value={val.get()}\nHTTP {status}\n{coerce_preview(body, 300)}") top.destroy() ttk.Button(top, text="Run", command=go).pack(pady=10) def menu_schedule_dialog(self): top = tk.Toplevel(self); top.title("Add Scheduled Function"); top.geometry("420x220"); top.transient(self); top.grab_set() ttk.Label(top, text="Function:").pack(pady=(10,2)) name = tk.StringVar(value=(self.functions_list[0] if self.functions_list else "")) cmb = ttk.Combobox(top, values=self.functions_list, textvariable=name, width=44, state="readonly") cmb.pack() ttk.Label(top, text="Value:").pack(pady=(6,2)) val = tk.StringVar(value="1") ttk.Entry(top, textvariable=val, width=16).pack() ttk.Label(top, text="Interval (seconds, can be < 1.0):").pack(pady=(6,2)) interval = tk.DoubleVar(value=1.0) ttk.Entry(top, textvariable=interval, width=12).pack() def go(): fname = name.get().strip() if not fname: return tid = self.scheduler.add_task(ActionTask(name=fname, value=val.get(), interval_s=max(0.0, float(interval.get())))) self.refresh_actions_tree() top.destroy() ttk.Button(top, text="Add", command=go).pack(pady=10) def save_config(self): try: from tkinter import filedialog cfg = { "host": self.host_var.get(), "port": int(self.port_var.get()), "refresh_interval": float(self.refresh_var.get()), "variables_keys": self.variables_keys, "vars": {k: self._serialize_varinfo(v) for k,v in self.vars.items() if k in self.variables_keys}, "scheduled_tasks": [self._serialize_task(t) for t in self.scheduler.list_tasks()], } path = filedialog.asksaveasfilename(defaultextension=".json", filetypes=[("JSON","*.json")], title="Save Configuration") if not path: return import json with open(path, "w", encoding="utf-8") as f: json.dump(cfg, f, indent=2) messagebox.showinfo("Saved", f"Configuration saved to:\n{path}") except Exception as e: messagebox.showerror("Save failed", str(e)) def load_config(self): try: from tkinter import filedialog path = filedialog.askopenfilename(filetypes=[("JSON","*.json")], title="Load Configuration") if not path: return import json with open(path, "r", encoding="utf-8") as f: cfg = json.load(f) self.host_var.set(cfg.get("host", self.host_var.get())) self.port_var.set(cfg.get("port", self.port_var.get())) self.refresh_var.set(cfg.get("refresh_interval", self.refresh_var.get())) keys = cfg.get("variables_keys", []) self.variables_keys = [k.lower() for k in keys] new_vars = {} for k, data in (cfg.get("vars") or {}).items(): vi = self._deserialize_varinfo(k, data) new_vars[k.lower()] = vi self.vars = new_vars # replace scheduled tasks for t in [t.task_id for t in self.scheduler.list_tasks()]: self.scheduler.remove_task(t) for td in cfg.get("scheduled_tasks", []): self.scheduler.add_task(self._deserialize_task(td)) if self.poller and self.poller.is_alive(): self.poller.variables_keys = self.variables_keys[:] self.refresh_tree() self.refresh_actions_tree() messagebox.showinfo("Loaded", f"Configuration loaded from:\n{path}") except Exception as e: messagebox.showerror("Load failed", str(e)) def _serialize_varinfo(self, vi: VarInfo): t = vi.thresholds return { "display_name": vi.display_name, "thresholds": { "dead_low": t.dead_low, "low": t.low, "high": t.high, "extreme_high": t.extreme_high, "alarm_dead_low": t.alarm_dead_low, "alarm_low": t.alarm_low, "alarm_high": t.alarm_high, "alarm_extreme_high": t.alarm_extreme_high, "action_dead_low": t.action_dead_low, "value_dead_low": t.value_dead_low, "action_dead_low_interval": t.action_dead_low_interval, "action_low": t.action_low, "value_low": t.value_low, "action_low_interval": t.action_low_interval, "action_high": t.action_high, "value_high": t.value_high, "action_high_interval": t.action_high_interval, "action_extreme_high": t.action_extreme_high, "value_extreme_high": t.value_extreme_high, "action_extreme_high_interval": t.action_extreme_high_interval, "expr_dead_low": t.expr_dead_low, "expr_low": t.expr_low, "expr_high": t.expr_high, "expr_extreme_high": t.expr_extreme_high, "expr_target_dead_low": t.expr_target_dead_low, "expr_target_low": t.expr_target_low, "expr_target_high": t.expr_target_high, "expr_target_extreme_high": t.expr_target_extreme_high, }, } def _deserialize_varinfo(self, key: str, data: dict) -> VarInfo: vi = VarInfo(key=key.lower(), display_name=data.get("display_name", key)) td = data.get("thresholds") or {} vi.thresholds = Thresholds( dead_low=td.get("dead_low"), low=td.get("low"), high=td.get("high"), extreme_high=td.get("extreme_high"), alarm_dead_low=td.get("alarm_dead_low", True), alarm_low=td.get("alarm_low", False), alarm_high=td.get("alarm_high", False), alarm_extreme_high=td.get("alarm_extreme_high", True), action_dead_low=td.get("action_dead_low"), value_dead_low=td.get("value_dead_low", "1"), action_dead_low_interval=td.get("action_dead_low_interval", 1.0), action_low=td.get("action_low"), value_low=td.get("value_low", "1"), action_low_interval=td.get("action_low_interval", 1.0), action_high=td.get("action_high"), value_high=td.get("value_high", "1"), action_high_interval=td.get("action_high_interval", 1.0), action_extreme_high=td.get("action_extreme_high"), value_extreme_high=td.get("value_extreme_high", "1"), action_extreme_high_interval=td.get("action_extreme_high_interval", 1.0), expr_dead_low=td.get("expr_dead_low"), expr_low=td.get("expr_low"), expr_high=td.get("expr_high"), expr_extreme_high=td.get("expr_extreme_high"), expr_target_dead_low=td.get("expr_target_dead_low"), expr_target_low=td.get("expr_target_low"), expr_target_high=td.get("expr_target_high"), expr_target_extreme_high=td.get("expr_target_extreme_high"), ) return vi def _serialize_task(self, t: ActionTask): return {"name": t.name, "value": t.value, "interval_s": t.interval_s, "enabled": t.enabled} def _deserialize_task(self, d: dict) -> ActionTask: return ActionTask(name=d.get("name",""), value=d.get("value","1"), interval_s=float(d.get("interval_s", 1.0)), enabled=d.get("enabled", True)) 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]: if value is None: return None m = re.search(r"[-+]?\d+(?:\.\d+)?", str(value).replace(",", ".")) if m: try: return float(m.group(0)) except Exception: return None return None def eval_user_expression(expr: str, x: float, y: float | None = None): """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, "y": y} 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()