gitea/vtt_work
1
0
Fork 0
mirror of https://github.com/migatu/vtt_work.git synced 2026-07-14 13:34:42 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
23084f4cb43f9c4919c1453aeb34f5d32a63e67f
vtt_work/nucleares_monitor/control_board_monitor.py
T
gitea 23084f4cb4 0.9
2025-10-20 18:27:06 +02:00

1156 lines
59 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Control Board Monitor — v0.9
Fixes:
- Selection flicker fixed: TreeView uses stable IDs and incremental updates.
- All HTTP (GET/POST) serialized via a global lock + "Connection: close" to prevent server resets when polling+posting.
- Expressions now post their result to a *chosen function* per state (not to the variable itself), avoiding "writable variable" errors.
- Function discovery preserves original case; posts use function names exactly as discovered.
Spec:
- Functions (POST actions) discovered via WEBSERVER_LIST_VARIABLES_JSON → "POST:"
- Variables for selection & bulk polling via WEBSERVER_BATCH_GET (keys)
- Default host:port = localhost:8785
"""
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/0.9 (+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']
# 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]]:
# Ensure single outstanding request at a time
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(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, "Connection": "close"})
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_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()
if vv and vv.upper() not in {"VARNAME", "VARIABLE", "NAME"} and vv not in seen:
seen.add(vv); uniq.append(vv)
return uniq
def parse_function_names_from_html_index(html_text: str) -> List[str]:
start = html_text.find("==== POST ====")
if start == -1:
return []
post_html = html_text[start:]
names = re.findall(r"<b>([A-Z0-9_]+)</b>", post_html)
seen = set(); uniq: List[str] = []
for n in names:
if n not in seen:
seen.add(n); uniq.append(n)
return uniq
# ---- Parse WEBSERVER_LIST_VARIABLES_JSON ----
def parse_weblist_names(body: str) -> Tuple[List[str], List[str]]:
"""Return (get_list, post_list) with original casing if possible."""
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
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
expr_dead_low: Optional[str] = None
expr_low: Optional[str] = None
expr_high: Optional[str] = None
expr_extreme_high: Optional[str] = None
# NEW: 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:
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()
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 (uses BATCH_GET)
# =====================
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")))
status, body, _ = http_get(base_url, {"variable": "WEBSERVER_BATCH_GET"})
if status == 200:
values_map = parse_batch_values(body)
if values_map:
# direct case-sensitive names from server
for name in list(self.variables):
# find exact or case-insensitive match
val = None
if name in values_map:
val = values_map[name]
else:
# case-insensitive fallback
for k, v in values_map.items():
if k.lower() == name.lower():
val = v; break
if val is not None:
self.ui_queue.put(("update", name, str(val), 200))
else:
self.ui_queue.put(("error", name, 206, "Not in BATCH_GET payload"))
else:
# fallback to per-variable
for name in list(self.variables):
st, b, _h = http_get(base_url, {"variable": name})
if st == 200:
self.ui_queue.put(("update", name, b, st))
else:
self.ui_queue.put(("error", name, st, coerce_preview(b, 200)))
else:
# 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)))
# 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.9")
self.geometry("1250x840")
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, selectmode="browse")
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=520, anchor="w")
self.tree.column("value", width=460, 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("<Button-3>", 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, selectmode="browse")
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("<Button-3>", 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
# Stable IDs
self._group_ids: Dict[str, str] = {}
self._var_ids: Dict[str, str] = {}
# --- 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:
discovered_vars.extend(sorted(values_map.keys()))
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
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 = 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("<Configure>", lambda e: canvas.configure(scrollregion=canvas.bbox("all")))
canvas.create_window((0,0), window=inner, anchor="nw")
canvas.configure(yscrollcommand=vsb.set)
canvas.pack(side="left", fill="both", expand=True); vsb.pack(side="right", fill="y")
current_set = set(self.variables_list or list(self.vars.keys()))
rows: List[Tuple[str, tk.BooleanVar, ttk.Checkbutton]] = []
def rebuild():
for child in inner.winfo_children():
child.destroy()
rows.clear()
f = filt_var.get().strip().lower()
for name in discovered:
if f and f not in name.lower():
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):
# Find selected variable row (skip groups)
sel = self.tree.selection()
if not sel:
messagebox.showinfo("No selection", "Select a variable first.")
return
iid = sel[0]
name = self.tree.item(iid, "text")
# If group, try to select first child
if name and iid.startswith("grp:"):
children = self.tree.get_children(iid)
if not children:
messagebox.showinfo("Empty group", "This group has no items.")
return
iid = children[0]
name = self.tree.item(iid, "text")
if not name or name.startswith("grp:"):
messagebox.showinfo("Select variable", "Select a specific variable row (not a group).")
return
vi = self.vars.get(name) or VarInfo(name=name)
dlg = tk.Toplevel(self)
dlg.title(f"Thresholds, actions & alarms — {name}")
dlg.geometry("900x650")
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=42, 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 (+ target function)
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, key, 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[key], width=56).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=30, state="readonly")
s.set(expr_target_vars[tkey].get())
s.grid(row=13+i, column=3, sticky="w", padx=6)
# keep binding in sync
def bind_setter(varkey=tkey, combo=s):
def _():
expr_target_vars[varkey].set(combo.get())
combo.bind("<<ComboboxSelected>>", lambda *_: _())
bind_setter()
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,
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[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 (compute & optionally POST to chosen function)
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
if expr and x is not None:
try:
computed = eval_user_expression(expr, x)
if target_func:
base = build_base_url(self.host_var.get().strip(), int(self.port_var.get()))
http_post(base, {"variable": target_func, "value": str(computed)})
else:
# No target selected; inform briefly
self.status_lbl.configure(text=f"Expression computed for {name} [{state}], no target function set")
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 incremental update ---
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, name: str) -> str:
iid = self._var_ids.get(name)
if iid and self.tree.exists(iid):
return iid
iid = f"var:{name}"
self._var_ids[name] = iid
if not self.tree.exists(iid):
self.tree.insert(group_iid, "end", iid=iid, text=name, values=("", "", ""), tags=("neutral",))
return iid
def refresh_tree(self) -> None:
filt = self.filter_var.get().strip().lower()
# Build groups from current vars
groups: Dict[str, List[VarInfo]] = {}
for name, vi in self.vars.items():
if filt and filt not in name.lower() and (not vi.last_value or filt not in vi.last_value.lower()):
continue
prefix = name.split("_", 1)[0] if "_" in name else "MISC"
groups.setdefault(prefix, []).append(vi)
# Preserve current selection
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()):
gid = self._ensure_group(g)
valid_group_iids.add(gid)
for vi in sorted(groups[g], key=lambda x: x.name.lower()):
vid = self._ensure_var(gid, vi.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)
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.item(vid, text=vi.name + suffix, values=(value_preview, updated_str, vi.last_status or "—"), tags=(tag,))
# Remove stale items
for name, 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(name, 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)
# Restore selection if possible
if sel_iid and self.tree.exists(sel_iid):
self.tree.selection_set(sel_iid)
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, "<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, "<expr>", "eval")
return eval(code, safe_globals, safe_locals)
def main() -> None:
app = App()
app.mainloop()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink