#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Control Board Monitor — v1.8 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 from collections import deque import time import urllib.parse import urllib.request from dataclasses import dataclass, field from datetime import datetime from typing import Dict, List, Optional, Tuple, Deque try: import tkinter as tk from tkinter import ttk, messagebox except Exception: raise SystemExit("Tkinter is required to run this app.") # Optional plotting support try: from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg from matplotlib.figure import Figure HAS_MPL = True import matplotlib as mpl mpl.rcParams["path.simplify"] = True mpl.rcParams["agg.path.chunksize"] = 10000 PLOT_SCALE = -200 except Exception: HAS_MPL = False # ===================== # Global configuration # ===================== AVERAGE_WINDOW_N: int = 60 # default for moving average window (cycles) SERVER_HOST: str = "localhost" SERVER_PORT: int = 8785 REFRESH_INTERVAL_S: int = 1 REQUEST_TIMEOUT_S: float = 5.0 USER_AGENT: str = "ControlBoardMonitor/1.14 (+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 {} # Jeśli serwer owinął odpowiedź w {"values": {...}}, wyciągamy środek: if isinstance(data, dict) and isinstance(data.get("values"), dict): data = data["values"] 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 def detect_batch_payload_type(body: str) -> str: try: data = json.loads(body) if isinstance(data, dict) and isinstance(data.get("values"), dict): return "enveloped(values+errors)" if isinstance(data, dict): return "flat" except Exception: pass return "unknown" def eval_threshold_expr(expr: str, stats_x: Optional[dict], stats_y: Optional[dict], stats_z: Optional[dict]) -> Optional[float]: """ Eval progu z dostępem do: x, x_avg, dx, dx_avg, y, y_avg, dy, dy_avg, z, z_avg, dz, dz_avg. Zwraca float albo None przy błędzie. """ if not expr: return None env: dict = {} def push(prefix: str, d: Optional[dict]): if not d: return env[prefix] = d.get("x") env[f"{prefix}_avg"] = d.get("x_avg") env[f"d{prefix}"] = d.get("dx") env[f"d{prefix}_avg"] = d.get("dx_avg") push("x", stats_x) push("y", stats_y) push("z", stats_z) try: # jeżeli masz już safe_eval w kodzie – użyj go; inaczej zwykły eval bez builtins val = safe_eval(expr, env) if 'safe_eval' in globals() else eval(expr, {"__builtins__": {}}, env) return float(val) except Exception: return None # ===================== # Data model # ===================== @dataclass class Thresholds: dead_low: Optional[float] = None low: Optional[float] = None high: Optional[float] = None extreme_high: Optional[float] = None # 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 action_operating: Optional[str] = None value_operating: str = "1" action_operating_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 expr_operating: Optional[str] = None expr_target_operating: Optional[str] = None expr_operating_interval: float = 1.0 # sources for y and z y_source: Optional[str] = None z_source: Optional[str] = None expr_x_source_dead_low: str = "raw" # "raw" | "x_avg" | "dx" | "dx_avg" expr_x_source_low: str = "raw" expr_x_source_operating: str = "raw" expr_x_source_high: str = "raw" expr_x_source_extreme_high: str = "raw" # Optional expressions for dynamic thresholds (override numeric constants if provided) expr_thr_dead_low: Optional[str] = None expr_thr_low: Optional[str] = None expr_thr_high: Optional[str] = None expr_thr_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" # analytics delta_last: Optional[float] = None history: Deque[float] = field(default_factory=lambda: deque(maxlen=3600)) history_delta: Deque[float] = field(default_factory=lambda: deque(maxlen=3600)) @dataclass class ActionTask: name: str value: str interval_s: float # dataclass ActionTask (dodaj pole) tag: Optional[str] = None enabled: bool = True last_run: float = 0.0 next_run: float = 0.0 task_id: int = 0 expr: Optional[str] = None x_src: Optional[str] = None y_src: Optional[str] = None z_src: Optional[str] = None x_mode: str = "raw" # "raw" | "x_avg" | "dx" | "dx_avg" # ===================== # Scheduler thread # ===================== class ActionScheduler(threading.Thread): def __init__(self, get_base_url_cb, get_value_cb): super().__init__(daemon=True) self.get_base_url_cb = get_base_url_cb self.get_value_cb = get_value_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 remove_tasks_by_tag(self, tag: str) -> None: with self._lock: to_del = [tid for tid, t in self.tasks.items() if getattr(t, "tag", None) == tag] for tid in to_del: self.tasks.pop(tid, 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_task_once(self, task: "ActionTask") -> Tuple[int, str, Dict[str, str]]: # Compute value if expression is provided value_to_send = task.value if getattr(task, "expr", None): # fetch x/y/z using callback; None -> 0.0 for x def gv(n): try: return self.get_value_cb(n) if n else None except Exception: return None # stara wersja: # x = gv(task.x_src); y = gv(task.y_src); z = gv(task.z_src) def pick(stats: Optional[dict], mode: str) -> Optional[float]: if not stats: return None if mode == "x_avg": return stats.get("x_avg") if mode == "dx": return stats.get("dx") if mode == "dx_avg": return stats.get("dx_avg") return stats.get("x") xs = self.get_value_cb(task.x_src) ys = self.get_value_cb(task.y_src) zs = self.get_value_cb(task.z_src) x = pick(xs, getattr(task, "x_mode", "raw")) y = pick(ys, getattr(task, "y_mode", "raw")) z = pick(zs, getattr(task, "z_mode", "raw")) if x is None: x = 0.0 try: computed = eval_user_expression(task.expr, x, y, z) value_to_send = str(computed) except Exception as e: # Return synthetic error without posting return 0, f"Expression error: {e}", {} base = self.get_base_url_cb() params = {"variable": task.name, "value": value_to_send} return http_post_query(base, params) 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_task_once(t) 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: fmt = detect_batch_payload_type(body) self.ui_queue.put(("batch_fmt", fmt)) 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.14") 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.get_stats_for ) 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) self.avg_window_var = tk.IntVar(value=AVERAGE_WINDOW_N) ttk.Label(controls, text="Host:").grid(row=0, column=0, sticky="w") ttk.Entry(controls, textvariable=self.host_var, width=16).grid(row=0, column=1, padx=(0, 10)) ttk.Label(controls, text="Port:").grid(row=0, column=2, sticky="w") ttk.Entry(controls, textvariable=self.port_var, width=8).grid(row=0, column=3, padx=(0, 10)) ttk.Label(controls, text="Refresh (s):").grid(row=0, column=4, sticky="w") ttk.Entry(controls, textvariable=self.refresh_var, width=8).grid(row=0, column=5, padx=(0, 10)) ttk.Label(controls, text="Avg N:").grid(row=0, column=6, sticky="w") ttk.Entry(controls, textvariable=self.avg_window_var, width=6).grid(row=0, column=7, padx=(0, 10)) self.status_lbl = ttk.Label(controls, text="Idle", foreground="#666") self.status_lbl.grid(row=0, column=8, padx=(10, 0)) btns = ttk.Frame(controls) btns.grid(row=0, column=9, padx=10, sticky="e") ttk.Button(btns, text="Select Vars…", command=self.open_selector).grid(row=0, column=0, padx=4) ttk.Button(btns, text="Reload (disc.)", command=self.reload_discovery).grid(row=0, column=1, padx=4) self.start_btn = ttk.Button(btns, text="Start", command=self.start_polling) self.start_btn.grid(row=0, column=2, padx=4) self.pause_btn = ttk.Button(btns, text="Pause", command=self.toggle_pause, state="disabled") self.pause_btn.grid(row=0, column=3, padx=4) 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", "delta", "davg", "avg", "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.avg_window_var.trace_add("write", lambda *_: (self.tree.heading("avg", text=f"Avg ({self.avg_window_var.get()})"), self.tree.heading("davg", text=f"Δ avg ({self.avg_window_var.get()})"))) self.tree.heading("#0", text="Variable") self.tree.heading("value", text="Value") self.tree.heading("delta", text="Δ last") self.tree.heading("davg", text=f"Δ avg ({self.avg_window_var.get()})") self.tree.heading("avg", text=f"Avg ({self.avg_window_var.get()})") self.tree.heading("updated", text="Updated") self.tree.heading("status", text="HTTP") self.tree.column("#0", width=500, anchor="w") self.tree.column("value", width=340, anchor="w") self.tree.column("delta", width=110, anchor="e") self.tree.column("davg", width=120, anchor="e") self.tree.column("avg", width=120, anchor="e") self.tree.column("updated", width=110, 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_menu.add_command(label="Open plot window…", command=self.open_plot_window) self.tree.bind("", self.on_tree_right_click) self.tree.bind("", lambda _e: self.open_plot_window()) 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) self.func_combo.pack(side=tk.LEFT, padx=6) self._attach_search_filter_to_combobox(self.func_combo, self.functions_list) 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="Open Plot for Selected…", command=self.open_plot_window) funmenu.add_separator() funmenu.add_command(label="Reload Discovery", command=self.reload_discovery) funmenu.add_command(label="Select Variables…", command=self.open_selector) menubar.add_cascade(label="Function", menu=funmenu) 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] = {} self.expr_state_tasks: Dict[tuple, int] = {} # Latest values from batch (lowercased keys) self.latest_values: Dict[str, str] = {} self.known_variables: List[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) def get_stats_for(self, name: Optional[str]) -> Optional[dict]: if not name: return None key = str(name).lower() vi = self.vars.get(key) def _avg(dq, n): if not dq or n <= 0 or len(dq) < n: return None lst = list(dq)[-n:] try: return sum(lst) / float(n) except Exception: return None try: n = max(1, int(self.avg_window_var.get())) except Exception: n = 60 if vi: return { "x": parse_first_float(vi.last_value), "x_avg": _avg(vi.history, n), "dx": vi.delta_last, "dx_avg": _avg(vi.history_delta, n), } # fallback dla zmiennych nie-monitorowanych val = self.get_current_value(name) return {"x": val, "x_avg": None, "dx": None, "dx_avg": None} # --- 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]) # Remember known variables for y/z source selectors self.known_variables = list(discovered_vars) 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 _reschedule_state_tasks_for(self, vi: VarInfo) -> None: """ Usuwa wszystkie 'while-in-state' taski dla tej zmiennej i tworzy na nowo tylko te, które powinny działać w bieżącym stanie według aktualnych progów. Interwał 0 => run once; >0 => cykliczne. """ if not hasattr(self, "scheduler") or self.scheduler is None: return t = vi.thresholds # 1) Usuń istniejące zadania powiązane z tą zmienną (po tagu): for state in ("DEAD_LOW", "LOW", "OPERATING", "HIGH", "EXTREME_HIGH"): tag = f"state::{vi.key}::{state}" self.scheduler.remove_tasks_by_tag(tag) # 2) Zmapuj ustawienia dla stanów: mapping = { "DEAD_LOW": ( t.action_dead_low, t.value_dead_low, t.action_dead_low_interval, getattr(t, "expr_dead_low", None), getattr(t, "expr_target_dead_low", None) ), "LOW": ( t.action_low, t.value_low, t.action_low_interval, getattr(t, "expr_low", None), getattr(t, "expr_target_low", None) ), "OPERATING": ( getattr(t, "action_operating", None), getattr(t, "value_operating", "1"), getattr(t, "action_operating_interval", 1.0), getattr(t, "expr_operating", None), getattr(t, "expr_target_operating", None) ), "HIGH": ( t.action_high, t.value_high, t.action_high_interval, getattr(t, "expr_high", None), getattr(t, "expr_target_high", None) ), "EXTREME_HIGH": ( t.action_extreme_high, t.value_extreme_high, t.action_extreme_high_interval, getattr(t, "expr_extreme_high", None), getattr(t, "expr_target_extreme_high", None) ), } curr_state = vi.last_state # aktualny stan (ustawiany w evaluate_thresholds / on_state_change) :contentReference[oaicite:4]{index=4} if curr_state not in mapping: self.refresh_actions_tree() return name, value, interval_s, expr, expr_target = mapping[curr_state] # Jeśli użytkownik ustawił 0 => 'run once' przy wejściu w stan: # Samo przełączenie na 0 ma natychmiast skasować ew. cykliczne zadania – zrobiliśmy to wyżej. # Nie dodajemy nowego zadania, jeśli nie ma co wykonywać. if not any([name, expr, expr_target]): self.refresh_actions_tree() return tag = f"state::{vi.key}::{curr_state}" if expr or expr_target: # zadanie 'obliczeniowe' – liczymy x/y/z (x=monitorowana) task = ActionTask( name=(name or expr_target or ""), # nazwa do tabeli/POST (gdy expr_target) value=(value if name else "0"), # value nie jest używane gdy expr_target interval_s=float(interval_s), expr=expr, x_src=vi.display_name, y_src=getattr(t, "y_source", None), z_src=getattr(t, "z_source", None), tag=tag, ) else: # zwykły POST do 'name' z 'value' task = ActionTask( name=name or "", value=str(value), interval_s=float(interval_s), tag=tag, ) self.scheduler.add_task(task) self.refresh_actions_tree() 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_task_once(t) 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 get_current_value(self, name: Optional[str]) -> Optional[float]: if not name: return None key = str(name).lower() if hasattr(self, "latest_values") and key in self.latest_values: return parse_first_float(self.latest_values.get(key)) base = build_base_url(self.host_var.get().strip(), int(self.port_var.get())) st, bd, _h = http_get(base, {"variable": key}) if st == 200: return parse_first_float(bd) return None def refresh_actions_tree(self): # zapamiętaj zaznaczenie/fokus sel_before = self.actions_tree.selection() focus_before = self.actions_tree.focus() # przebuduj 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") + (" + Expr" if getattr(t, 'expr', None) else "") 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") ) # odtwórz zaznaczenie/fokus (o ile elementy wciąż istnieją) if sel_before: kept = [iid for iid in sel_before if self.actions_tree.exists(iid)] if kept: self.actions_tree.selection_set(kept) if focus_before and self.actions_tree.exists(focus_before): self.actions_tree.focus(focus_before) def _ensure_plot_timer(self): if self._plot_timer is None: # odświeżanie wykresów zsynchronizowane z głównym intervalem try: ms = int(max(100, float(self.refresh_var.get()) * 1000)) except Exception: ms = 500 self._plot_timer = self.after(ms, self._plot_tick) def _plot_tick(self): # jeśli nie ma żadnych okien – zatrzymaj timer windows = [w for w in getattr(self, "_plot_windows", {}).values() if w.winfo_exists()] if not windows: self._plot_timer = None return for win in windows: key = getattr(win, "_key", None) vi = self.vars.get(key) if not vi: continue # ostatnie 100 próbek vals = list(vi.history)[PLOT_SCALE:] if hasattr(vi, "history") else [] dels = list(vi.history_delta)[PLOT_SCALE:] if hasattr(vi, "history_delta") else [] # --- GÓRNY wykres: linia --- xs_v = list(range(len(vals))) win._line.set_data(xs_v, vals) if vals: vmin, vmax = min(vals), max(vals) span = (vmax - vmin) or 1.0 m = 0.08 * span win._ax_val.set_xlim(0, max(1, len(xs_v) - 1)) win._ax_val.set_ylim(vmin - m, vmax + m) win._ax_val.set_ylabel("value") # --- DOLNY wykres: słupki (bez ponownego tworzenia barów jeśli długość ta sama) --- if len(dels) != win._bars_len: # przebuduj tylko gdy zmienia się liczba słupków for b in win._bars: b.remove() win._bars = win._ax_delta.bar(range(len(dels)), dels) win._bars_len = len(dels) else: # szybka ścieżka – zaktualizuj wysokości for b, h in zip(win._bars, dels): b.set_height(h) if dels: dmin, dmax = min(dels), max(dels) span = (dmax - dmin) or 1.0 m = 0.08 * span win._ax_delta.set_xlim(0, max(1, len(dels) - 1)) win._ax_delta.set_ylim(dmin - m, dmax + m) win._ax_delta.set_ylabel("Δ") win._ax_delta.set_xlabel("last 100 samples") # tight_layout tylko po resize (ustawiane flagą) if getattr(win, "_layout_dirty", False): try: win._fig.tight_layout() except Exception: pass win._layout_dirty = False # draw_idle = koalescencja try: win._canvas.draw_idle() except Exception: pass # zaplanuj kolejny tick try: ms = int(max(100, float(self.refresh_var.get()) * 1000)) except Exception: ms = 500 self._plot_timer = self.after(ms, self._plot_tick) # --- 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 _bind_mousewheel(self, widget): # Windows / macOS def _mw(event): if event.delta: widget.yview_scroll(int(-1 * (event.delta / 120)), "units") return "break" # Linux (X11) 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 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) prev_num = parse_first_float(vi.last_value) curr_num = parse_first_float(value) vi.last_value = value vi.last_updated = time.time() vi.last_status = status vi.error = None # analytics update if prev_num is not None and curr_num is not None: vi.delta_last = curr_num - prev_num try: vi.history_delta.append(vi.delta_last) except Exception: pass else: vi.delta_last = None if curr_num is not None: try: vi.history.append(curr_num) except Exception: pass 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) # refresh known_variables list from batch keys (preserve case with existing where possible) keys = list(mapping.keys()) # prefer existing display names from self.vars, else use given keys disp = [self.vars.get(k.lower()).display_name if self.vars.get(k.lower()) else k for k in keys] self.known_variables = sorted(set(list(self.known_variables) + disp), key=str.lower) elif kind == "stopped": self.status_lbl.configure(text="Stopped") elif kind == "batch_fmt": _, fmt = item # nie nadpisuj info o błędach/paused – tylko dołóż sufiks cur = self.status_lbl.cget("text") or "" base = cur.split(" | ")[0] # zachowaj, co było przedtem self.status_lbl.configure(text=f"{base} | batch: {fmt}") 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_plot_window(self): """Open (or focus) a per-variable plot window with last 100 values and last 100 deltas. Top: values line + horizontal mean; Bottom: delta bars + horizontal mean. Each subplot has its own normalized y-scale. Auto-refreshes every ~1s. """ 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 yet.") return if not HAS_MPL: messagebox.showwarning("Plotting not available", "Matplotlib is not installed. Install 'matplotlib' to enable plots.") return if not hasattr(self, "_plot_windows"): self._plot_windows = {} self._plot_timer = None # globalny timer do wykresów w = self._plot_windows.get(key) if w and w.winfo_exists(): try: w.lift(); w.focus_force() return except Exception: pass win = tk.Toplevel(self) win.title(f"Plot — {vi.display_name}") win.geometry("820x520") win.transient(self) fig = Figure(figsize=(8, 5), dpi=100) ax_val = fig.add_subplot(211) ax_delta = fig.add_subplot(212, sharex=ax_val) canvas = FigureCanvasTkAgg(fig, master=win) canvas.draw() # ... masz już fig, ax_val, ax_delta, canvas ... canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) self._plot_windows[key] = win win._fig = fig win._ax_val = ax_val win._ax_delta = ax_delta win._canvas = canvas win._key = key win._layout_dirty = True # prealokuj „artystów” – jedna linia i zestaw słupków (win._line,) = ax_val.plot([], [], linewidth=1.2) win._bars = ax_delta.bar([], []) win._bars_len = 0 # ile aktualnie słupków # tight_layout tylko po resize def _on_resize(_evt=None): win._layout_dirty = True canvas.mpl_connect("resize_event", _on_resize) # uruchom globalny tick jeśli nie działa self._ensure_plot_timer() def open_thresholds_dialog(self): sel = self.tree.selection() expr_xsrc_vars = {} 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("980x700") dlg.transient(self); dlg.grab_set() t = vi.thresholds 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)} # Scrollable body outer = ttk.Frame(dlg); outer.pack(fill=tk.BOTH, expand=True) canvas = tk.Canvas(outer, highlightthickness=0) vsb = ttk.Scrollbar(outer, orient="vertical", command=canvas.yview) canvas.configure(yscrollcommand=vsb.set) vsb.pack(side=tk.RIGHT, fill=tk.Y) canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) grid = ttk.Frame(canvas) win = canvas.create_window((0,0), window=grid, anchor="nw") def _on_cfg(event=None): canvas.configure(scrollregion=canvas.bbox("all")) canvas.itemconfigure(win, width=canvas.winfo_width()) grid.bind("", _on_cfg) canvas.bind("", _on_cfg) # thresholds ttk.Label(grid, text="DEAD LOW").grid(row=0, column=0, sticky="w", pady=2) ttk.Label(grid, text="LOW").grid(row=1, column=0, sticky="w", pady=2) ttk.Label(grid, text="HIGH").grid(row=2, column=0, sticky="w", pady=2) ttk.Label(grid, text="EXTREME HIGH").grid(row=3, column=0, sticky="w", pady=2) def mk_val(default, keyname, row): dv = auto_defaults.get(keyname) if default is None else default sv = tk.StringVar(value="" if dv is None else str(dv)) ttk.Entry(grid, textvariable=sv, width=12).grid(row=row, column=1, sticky="w", padx=6) return sv dead_low_var = mk_val(t.dead_low, "dead_low", 0) low_var = mk_val(t.low, "low", 1) high_var = mk_val(t.high, "high", 2) extreme_high_var = mk_val(t.extreme_high, "extreme_high", 3) ttk.Label(grid, text="Alarm on enter:").grid(row=0, column=3, sticky="w") alarm_dead_low = tk.BooleanVar(value=t.alarm_dead_low) alarm_low = tk.BooleanVar(value=t.alarm_low) alarm_high = tk.BooleanVar(value=t.alarm_high) alarm_ext = tk.BooleanVar(value=t.alarm_extreme_high) ttk.Checkbutton(grid, text="Dead Low", variable=alarm_dead_low).grid(row=1, column=3, sticky="w") ttk.Checkbutton(grid, text="Low", variable=alarm_low).grid(row=2, column=3, sticky="w") ttk.Checkbutton(grid, text="High", variable=alarm_high).grid(row=3, column=3, sticky="w") ttk.Checkbutton(grid, text="Extreme High", variable=alarm_ext).grid(row=4, column=3, sticky="w") # actions ttk.Label(grid, text="On enter: call function (optional)").grid(row=6, column=0, sticky="w", pady=(12,4)) ttk.Label(grid, text="Interval(s) while in state (0 = once)").grid(row=6, column=3, sticky="w") rows = [("Dead Low", "action_dead_low", "value_dead_low", "action_dead_low_interval"), ("Low", "action_low", "value_low", "action_low_interval"), ("Operating", "action_operating", "value_operating", "action_operating_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 = {} for i, (label, action_key, value_key, interval_key) in enumerate(rows): r = 7+i ttk.Label(grid, text=label).grid(row=r, column=0, sticky="w", pady=2) s = ttk.Combobox(grid, values=self.functions_list or DEFAULT_FUNCTIONS, width=44) s.set(getattr(t, action_key) or "") s.grid(row=r, column=1, sticky="w", padx=6) self._attach_search_filter_to_combobox(s, self.functions_list) action_vars[action_key] = s sv = tk.StringVar(value=getattr(t, value_key)) ttk.Entry(grid, textvariable=sv, width=12).grid(row=r, 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=r, column=3, sticky="w", padx=6) interval_vars[interval_key] = iv # y/z sources ttk.Label(grid, text="y source (optional)").grid(row=13, column=0, sticky="w", pady=(14,4)) y_source_var = tk.StringVar(value=t.y_source or "") y_combo = ttk.Combobox(grid, values=(self.known_variables or [vi.display_name]), textvariable=y_source_var, width=44) y_combo.grid(row=13, column=1, sticky="w", padx=6) self._attach_search_filter_to_combobox(y_combo, self.known_variables) ttk.Label(grid, text="z source (optional)").grid(row=13, column=2, sticky="e") z_source_var = tk.StringVar(value=t.z_source or "") z_combo = ttk.Combobox(grid, values=(self.known_variables or [vi.display_name]), textvariable=z_source_var, width=32) z_combo.grid(row=13, column=3, sticky="w", padx=6) self._attach_search_filter_to_combobox(z_combo, self.known_variables) # expressions ttk.Label(grid, text="On enter: evaluate expression/snippet (x, y, z available) and optionally POST to a function").grid(row=15, column=0, sticky="w", pady=(14,4), columnspan=3) expr_rows = [("Dead Low expr", "expr_dead_low", "expr_target_dead_low"), ("Low expr", "expr_low", "expr_target_low"), ("Operating expr", "expr_operating", "expr_target_operating"), ("High expr", "expr_high", "expr_target_high"), ("Extreme High expr", "expr_extreme_high", "expr_target_extreme_high")] expr_vars = {}; expr_target_vars = {} for i, (label, keyname, tkey) in enumerate(expr_rows): r = 16+i ttk.Label(grid, text=label).grid(row=r, column=0, sticky="w", pady=2) expr_vars[keyname] = tk.StringVar(value=getattr(t, keyname) or "") ttk.Entry(grid, textvariable=expr_vars[keyname], width=64).grid(row=r, column=1, sticky="w", padx=6) ttk.Label(grid, text="→ POST to function").grid(row=r, column=2, sticky="e") s = ttk.Combobox(grid, values=self.functions_list or DEFAULT_FUNCTIONS, width=32) s.set(getattr(t, tkey) or "") s.grid(row=r, column=3, sticky="w", padx=6) self._attach_search_filter_to_combobox(s, self.functions_list) expr_target_vars[tkey] = s ttk.Label(grid, text="x from").grid(row=r, column=4, sticky="e") src_combo = ttk.Combobox(grid, values=["raw", "x_avg", "dx", "dx_avg"], width=8) # mapowanie nazwy wiersza na pole w Thresholds: statekey = { "Dead Low expr": "dead_low", "Low expr": "low", "Operating expr": "operating", "High expr": "high", "Extreme High expr": "extreme_high", }[label] field_name = f"expr_x_source_{statekey}" src_combo.set(getattr(t, field_name, "raw")) src_combo.grid(row=r, column=5, sticky="w") expr_xsrc_vars[field_name] = src_combo # Bottom bar btns = ttk.Frame(dlg); btns.pack(fill=tk.X, padx=10, pady=10) def do_save(): def _parse_threshold_field(var_or_entry) -> tuple[Optional[float], Optional[str]]: # Przyjmujemy tk.StringVar / tk.Entry / str if hasattr(var_or_entry, "get"): s = var_or_entry.get().strip() else: s = str(var_or_entry).strip() if s == "": return None, None try: return float(s), None except Exception: # potraktuj jako wyrażenie (np. "x_avg - 5") return None, s # UŻYJ POPRAWNYCH ZMIENNYCH: dead_val, dead_expr = _parse_threshold_field(dead_low_var) low_val, low_expr = _parse_threshold_field(low_var) high_val, high_expr = _parse_threshold_field(high_var) ext_val, ext_expr = _parse_threshold_field(extreme_high_var) nt = Thresholds( dead_low=dead_val, low=low_val, high=high_val, extreme_high=ext_val, expr_thr_dead_low=dead_expr, expr_thr_low=low_expr, expr_thr_high=high_expr, expr_thr_extreme_high=ext_expr, alarm_dead_low=bool(alarm_dead_low.get()), alarm_low=bool(alarm_low.get()), alarm_high=bool(alarm_high.get()), alarm_extreme_high=bool(alarm_ext.get()), action_dead_low=action_vars["action_dead_low"].get() or None, value_dead_low=value_vars["value_dead_low"].get(), action_dead_low_interval=float(interval_vars["action_dead_low_interval"].get() or 1.0), action_low=action_vars["action_low"].get() or None, value_low=value_vars["value_low"].get(), action_low_interval=float(interval_vars["action_low_interval"].get() or 1.0), action_operating=action_vars["action_operating"].get() or None, value_operating=value_vars["value_operating"].get(), action_operating_interval=float(interval_vars["action_operating_interval"].get() or 1.0), action_high=action_vars["action_high"].get() or None, value_high=value_vars["value_high"].get(), action_high_interval=float(interval_vars["action_high_interval"].get() or 1.0), action_extreme_high=action_vars["action_extreme_high"].get() or None, value_extreme_high=value_vars["value_extreme_high"].get(), 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_operating=(expr_vars["expr_operating"].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_operating=(expr_target_vars["expr_target_operating"].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), y_source=(y_source_var.get().strip() or None), z_source=(z_source_var.get().strip() or None), expr_x_source_dead_low = expr_xsrc_vars["expr_x_source_dead_low"].get(), expr_x_source_low = expr_xsrc_vars["expr_x_source_low"].get(), expr_x_source_operating = expr_xsrc_vars["expr_x_source_operating"].get(), expr_x_source_high = expr_xsrc_vars["expr_x_source_high"].get(), expr_x_source_extreme_high = expr_xsrc_vars["expr_x_source_extreme_high"].get(), ) vi.thresholds = nt # Re-enter current state to refresh scheduled actions/intervals try: cur_x = parse_first_float(vi.last_value) self.on_state_change(vi, vi.last_state, vi.last_state, cur_x) except Exception: pass dlg.destroy() self._reschedule_state_tasks_for(vi) ttk.Button(btns, text="Save", command=do_save).pack(side=tk.RIGHT, padx=6) ttk.Button(btns, text="Cancel", command=dlg.destroy).pack(side=tk.RIGHT) def evaluate_thresholds(self, vi: VarInfo, value: str) -> None: x = parse_first_float(value) t = vi.thresholds # Dynamic thresholds stats_x = self.get_stats_for(vi.key) stats_y = self.get_stats_for(getattr(t, "y_source", None)) stats_z = self.get_stats_for(getattr(t, "z_source", None)) thr_dead = t.dead_low thr_low = t.low thr_high = t.high thr_ext = t.extreme_high if t.expr_thr_dead_low: v = eval_threshold_expr(t.expr_thr_dead_low, stats_x, stats_y, stats_z) if v is not None: thr_dead = v if t.expr_thr_low: v = eval_threshold_expr(t.expr_thr_low, stats_x, stats_y, stats_z) if v is not None: thr_low = v if t.expr_thr_high: v = eval_threshold_expr(t.expr_thr_high, stats_x, stats_y, stats_z) if v is not None: thr_high = v if t.expr_thr_extreme_high: v = eval_threshold_expr(t.expr_thr_extreme_high, stats_x, stats_y, stats_z) if v is not None: thr_ext = v # 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" no_thr = all(v is None for v in (thr_dead, thr_low, thr_high, thr_ext)) if x is None or no_thr: state = "OPERATING" else: dl = thr_dead if thr_dead is not None else -float("inf") lo = thr_low if thr_low is not None else -float("inf") hi = thr_high if thr_high is not None else float("inf") ex = thr_ext if thr_ext 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 # stop any repeating expr task for previous state etid = self.expr_state_tasks.pop((vi.key, prev_state), None) if etid is not None: try: self.scheduler.remove_task(etid) 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/z current values (prefer explicit sources) y_val = None; z_val = None y_source = (t.y_source or '').strip() or None z_source = (t.z_source or '').strip() or None if not y_source: if target_func: y_source = target_func else: action_map = {"DEAD_LOW": t.action_dead_low, "LOW": t.action_low, "HIGH": t.action_high, "EXTREME_HIGH": t.action_extreme_high} y_source = action_map.get(state) for label, src_name in (('y', y_source), ('z', z_source)): if not src_name: continue key = src_name.lower() val = None if hasattr(self, 'latest_values') and key in self.latest_values: val = parse_first_float(self.latest_values.get(key)) else: base = build_base_url(self.host_var.get().strip(), int(self.port_var.get())) st, bd, _h = http_get(base, {"variable": key}) if st == 200: val = parse_first_float(bd) if label == 'y': y_val = val else: z_val = val x_variants = self.get_stats_for(vi.key) or {} mode_map = { "DEAD_LOW": t.expr_x_source_dead_low, "LOW": t.expr_x_source_low, "OPERATING": t.expr_x_source_operating, "HIGH": t.expr_x_source_high, "EXTREME_HIGH": t.expr_x_source_extreme_high, } x_for_expr = { "raw": x_variants.get("x"), "x_avg": x_variants.get("x_avg"), "dx": x_variants.get("dx"), "dx_avg": x_variants.get("dx_avg"), }.get(mode_map.get(state, "raw"), x_variants.get("x")) if expr and x_for_expr is not None: try: computed = eval_user_expression(expr, x_for_expr, y_val, z_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), "OPERATING": (t.action_operating, t.value_operating), "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, "OPERATING": t.action_operating_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 # Expression scheduling while in state expr_key_map = { "DEAD_LOW": ("expr_dead_low", "expr_target_dead_low"), "LOW": ("expr_low", "expr_target_low"), "OPERATING": ("expr_operating", "expr_target_operating"), "HIGH": ("expr_high", "expr_target_high"), "EXTREME_HIGH": ("expr_extreme_high", "expr_target_extreme_high"), } ekey, tkey = expr_key_map.get(state, (None, None)) expr_str = getattr(t, ekey) if ekey else None expr_target = getattr(t, tkey) if tkey else None if expr_str and expr_target: # Use action intervals as default pacing for expressions expr_interval = { "DEAD_LOW": t.action_dead_low_interval, "LOW": t.action_low_interval, "OPERATING": t.action_operating_interval, "HIGH": t.action_high_interval, "EXTREME_HIGH": t.action_extreme_high_interval, }.get(state, 1.0) or 1.0 if expr_interval <= 0.0: # compute once on enter (already computed above in expressions block), # but ensure it posts now via scheduler so behavior is consistent task = ActionTask(name=expr_target, value="", interval_s=0.0, expr=expr_str, x_src=vi.key, y_src=t.y_source, z_src=t.z_source) self.scheduler.add_task(task) else: # schedule repeating expression task task = ActionTask( name=expr_target, value="", interval_s=max(0.0, float(expr_interval)), expr=expr_str, x_src=vi.key, y_src=t.y_source, z_src=t.z_source, x_mode=mode_map.get(state, "raw") ) tid = self.scheduler.add_task(task) self.expr_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, x_value=None) -> str: """Kolor tła w hex w oparciu o progi (także dynamiczne). Bezpiecznie obsługuje brak wartości x (None).""" # --- efektywne progi (uwzględniają expr_thr_*) --- t = vi.thresholds stats_x = self.get_stats_for(vi.key) stats_y = self.get_stats_for(getattr(t, "y_source", None)) stats_z = self.get_stats_for(getattr(t, "z_source", None)) def _eff(thr_num, thr_expr): if thr_expr: v = eval_threshold_expr(thr_expr, stats_x, stats_y, stats_z) if v is not None: return v return thr_num thr_dead = _eff(getattr(t, "dead_low", None), getattr(t, "expr_thr_dead_low", None)) thr_low = _eff(getattr(t, "low", None), getattr(t, "expr_thr_low", None)) thr_high = _eff(getattr(t, "high", None), getattr(t, "expr_thr_high", None)) thr_ext = _eff(getattr(t, "extreme_high", None), getattr(t, "expr_thr_extreme_high", None)) # --- x może być None (np. świeży monitor / brak liczby w payloadzie) --- try: x = float(x_value) if x_value is not None else parse_first_float(vi.last_value) except Exception: x = None if x is None: # neutralny szary – brak danych return "#E9ECEF" # --- fallback progów kiedy część jest pusta --- if thr_dead is None: thr_dead = -float("inf") if thr_low is None and thr_high is None: # brak LOW/HIGH -> otocz x wąskim korytarzem, by dać 'zielony' thr_low, thr_high = x - 1e-6, x + 1e-6 if thr_low is None: thr_low = thr_dead if thr_dead != -float("inf") else (x - 1e-6) if thr_high is None: thr_high = max(thr_low, x + 1e-6) if thr_ext is None: thr_ext = float("inf") # --- prosty lerp koloru w RGB --- def _hex(c): return "#{:02X}{:02X}{:02X}".format(*c) def _lerp(c1, c2, t): t = 0.0 if t < 0 else 1.0 if t > 1 else t return (int(c1[0] + (c2[0]-c1[0])*t), int(c1[1] + (c2[1]-c1[1])*t), int(c2[2] + (c2[2]-c1[2])*t)) # (uwaga: to tylko komentarz – kod niżej bez zmian) Y_LO = (255, 244, 178) Y_HI = (255, 149, 0) G_OK = (46, 204, 113) R_LO = (255, 138, 128) R_HI = (213, 0, 0) if x <= thr_low: denom = max(1e-9, (thr_low - thr_dead)) if thr_low > thr_dead else 1.0 t = (thr_low - x) / denom return _hex(_lerp(Y_LO, Y_HI, t)) if x <= thr_high: return _hex(G_OK) denom = max(1e-9, (thr_ext - thr_high)) if thr_ext > thr_high else 1.0 t = (x - thr_high) / denom return _hex(_lerp(R_LO, R_HI, t)) 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) # Δ last delta_str = "N/A" if vi.delta_last is not None: try: delta_str = f"{vi.delta_last:+.6g}" except Exception: delta_str = str(vi.delta_last) # Δ avg(N) davg_str = "N/A" try: n = max(1, int(self.avg_window_var.get())) except Exception: n = AVERAGE_WINDOW_N if hasattr(vi, "history_delta") and vi.history_delta: dvals = list(vi.history_delta)[-n:] if len(dvals) >= n: try: davg_val = sum(dvals) / float(n) davg_str = f"{davg_val:.6g}" except Exception: davg_str = "N/A" # value avg(N) avg_str = "N/A" if hasattr(vi, "history") and vi.history: vals = list(vi.history)[-n:] if len(vals) >= n: try: avg_val = sum(vals) / float(n) avg_str = f"{avg_val:.6g}" except Exception: avg_str = "N/A" 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, delta_str, davg_str, avg_str, 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 _attach_search_filter_to_combobox(self, combo: ttk.Combobox, source: List[str] | None): if source is None: source = [] original = list(source) combo.configure(state="normal") # allow typing to filter def on_key(_event=None): text = combo.get() vals = [v for v in original if text.lower() in v.lower()] combo["values"] = vals if vals else original combo.bind("", on_key) 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) self._attach_search_filter_to_combobox(cmb, self.functions_list) 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("520x360"); 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) self._attach_search_filter_to_combobox(cmb, self.functions_list) cmb.pack() # Mode: Fixed vs Expression mode = tk.StringVar(value="expr") frm = ttk.Frame(top); frm.pack(pady=(8,2)) ttk.Radiobutton(frm, text="Fixed value", variable=mode, value="fixed").pack(side=tk.LEFT, padx=6) ttk.Radiobutton(frm, text="Expression", variable=mode, value="expr").pack(side=tk.LEFT, padx=6) # Fixed value input val = tk.StringVar(value="1") fixed_row = ttk.Frame(top); fixed_row.pack(fill=tk.X, padx=10, pady=(4,2)) ttk.Label(fixed_row, text="Value:").pack(side=tk.LEFT) val_entry = ttk.Entry(fixed_row, textvariable=val, width=16) val_entry.pack(side=tk.LEFT, padx=6) # Expression + sources expr = tk.StringVar(value="x") expr_row = ttk.Frame(top); expr_row.pack(fill=tk.X, padx=10, pady=(4,2)) ttk.Label(expr_row, text="Expression (x,y,z):").pack(side=tk.LEFT) expr_entry = ttk.Entry(expr_row, textvariable=expr, width=46) expr_entry.pack(side=tk.LEFT, padx=6) # Sources row src_row = ttk.Frame(top); src_row.pack(fill=tk.X, padx=10, pady=(4,2)) ttk.Label(src_row, text="x:").pack(side=tk.LEFT) x_src = tk.StringVar(value="") x_combo = ttk.Combobox(src_row, values=self.known_variables, textvariable=x_src, width=18) self._attach_search_filter_to_combobox(x_combo, self.known_variables) x_combo.pack(side=tk.LEFT, padx=4) ttk.Label(src_row, text="y:").pack(side=tk.LEFT) y_src = tk.StringVar(value="") y_combo = ttk.Combobox(src_row, values=self.known_variables, textvariable=y_src, width=18) self._attach_search_filter_to_combobox(y_combo, self.known_variables) y_combo.pack(side=tk.LEFT, padx=4) ttk.Label(src_row, text="z:").pack(side=tk.LEFT) z_src = tk.StringVar(value="") z_combo = ttk.Combobox(src_row, values=self.known_variables, textvariable=z_src, width=18) self._attach_search_filter_to_combobox(z_combo, self.known_variables) z_combo.pack(side=tk.LEFT, padx=4) xmode_row = ttk.Frame(top); xmode_row.pack(fill=tk.X, padx=10, pady=(4,2)) ttk.Label(xmode_row, text="x from:").pack(side=tk.LEFT) x_mode = tk.StringVar(value="raw") xmode_combo = ttk.Combobox(xmode_row, values=["raw", "x_avg", "dx", "dx_avg"], textvariable=x_mode, width=10) xmode_combo.pack(side=tk.LEFT, padx=6) # Interval input 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 update_mode(*_): if mode.get() == "fixed": val_entry.configure(state="normal") expr_entry.configure(state="disabled") x_combo.configure(state="disabled"); y_combo.configure(state="disabled"); z_combo.configure(state="disabled") else: val_entry.configure(state="disabled") expr_entry.configure(state="normal") x_combo.configure(state="normal"); y_combo.configure(state="normal"); z_combo.configure(state="normal") mode.trace_add("write", update_mode); update_mode() def go(): fname = name.get().strip() if not fname: return if mode.get() == "fixed": task = ActionTask(name=fname, value=val.get(), interval_s=max(0.0, float(interval.get()))) else: task = ActionTask( name=fname, value="", interval_s=max(0.0, float(interval.get())), expr=expr.get().strip() or "x", x_src=(x_src.get().strip() or None), y_src=(y_src.get().strip() or None), z_src=(z_src.get().strip() or None), x_mode=x_mode.get() ) tid = self.scheduler.add_task(task) 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, "action_operating": t.action_operating, "value_operating": t.value_operating, "action_operating_interval": t.action_operating_interval, "expr_dead_low": t.expr_dead_low, "expr_low": t.expr_low, "expr_operating": t.expr_operating, "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_operating": t.expr_target_operating, "expr_target_high": t.expr_target_high, "expr_target_extreme_high": t.expr_target_extreme_high, # nowości: "expr_operating_interval": getattr(t, "expr_operating_interval", 1.0), "y_source": getattr(t, "y_source", None), "z_source": getattr(t, "z_source", None), "expr_x_source_dead_low": t.expr_x_source_dead_low, "expr_x_source_low": t.expr_x_source_low, "expr_x_source_operating": t.expr_x_source_operating, "expr_x_source_high": t.expr_x_source_high, "expr_x_source_extreme_high": t.expr_x_source_extreme_high, "expr_thr_dead_low": t.expr_thr_dead_low, "expr_thr_low": t.expr_thr_low, "expr_thr_high": t.expr_thr_high, "expr_thr_extreme_high": t.expr_thr_extreme_high, }, } def _deserialize_varinfo(self, key: str, data: dict) -> VarInfo: name = data.get("display_name", key) td = (data or {}).get("thresholds", {}) or {} t = Thresholds( dead_low=td.get("dead_low"), low=td.get("low"), high=td.get("high"), extreme_high=td.get("extreme_high"), alarm_dead_low=bool(td.get("alarm_dead_low", True)), alarm_low=bool(td.get("alarm_low", False)), alarm_high=bool(td.get("alarm_high", False)), alarm_extreme_high=bool(td.get("alarm_extreme_high", True)), action_dead_low=td.get("action_dead_low"), value_dead_low=str(td.get("value_dead_low", "1")), action_dead_low_interval=float(td.get("action_dead_low_interval", 1.0)), action_low=td.get("action_low"), value_low=str(td.get("value_low", "1")), action_low_interval=float(td.get("action_low_interval", 1.0)), action_high=td.get("action_high"), value_high=str(td.get("value_high", "1")), action_high_interval=float(td.get("action_high_interval", 1.0)), action_extreme_high=td.get("action_extreme_high"), value_extreme_high=str(td.get("value_extreme_high", "1")), action_extreme_high_interval=float(td.get("action_extreme_high_interval", 1.0)), action_operating=td.get("action_operating"), value_operating=str(td.get("value_operating", "1")), action_operating_interval=float(td.get("action_operating_interval", 1.0)), expr_dead_low=td.get("expr_dead_low"), expr_low=td.get("expr_low"), expr_operating=td.get("expr_operating"), 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_operating=td.get("expr_target_operating"), expr_target_high=td.get("expr_target_high"), expr_target_extreme_high=td.get("expr_target_extreme_high"), # nowości (wstecznie opcjonalne): expr_operating_interval=float(td.get("expr_operating_interval", 1.0)), y_source=td.get("y_source"), z_source=td.get("z_source"), expr_x_source_dead_low=td.get("expr_x_source_dead_low","raw"), expr_x_source_low=td.get("expr_x_source_low","raw"), expr_x_source_operating=td.get("expr_x_source_operating","raw"), expr_x_source_high=td.get("expr_x_source_high","raw"), expr_x_source_extreme_high=td.get("expr_x_source_extreme_high","raw"), expr_thr_dead_low=td.get("expr_thr_dead_low"), expr_thr_low=td.get("expr_thr_low"), expr_thr_high=td.get("expr_thr_high"), expr_thr_extreme_high=td.get("expr_thr_extreme_high"), ) return VarInfo(key=key.lower(), display_name=name, thresholds=t) def _serialize_task(self, t: ActionTask): return {"name": t.name, "value": t.value, "interval_s": t.interval_s, "enabled": t.enabled, "expr": t.expr, "x_src": t.x_src, "y_src": t.y_src, "z_src": t.z_src, "x_mode": getattr(t, "x_mode", "raw")} 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), expr=d.get("expr"), x_src=d.get("x_src"), y_src=d.get("y_src"), z_src=d.get("z_src"), x_mode=d.get("x_mode","raw")) 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 try: if self._plot_timer is not None: self.after_cancel(self._plot_timer) self._plot_timer = None 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 safe_eval(expression: str, env: dict): """ Bezpieczna ewaluacja krótkich wyrażeń progowych. Dostępne: math, min, max, abs, round, int, float. Zmiennych szukamy w `env` (np. x, x_avg, dx, dx_avg, y, y_avg, ...). """ allowed_globals = { "__builtins__": {}, "math": math, "min": min, "max": max, "abs": abs, "round": round, "int": int, "float": float, } # tylko 'eval' – bez snippetów wieloliniowych: code = compile(expression, "", "eval") return eval(code, allowed_globals, env) def eval_user_expression(expr: str, x: float, y: float | None = None, z: 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, "z": z} 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()