gitea/vtt_work
1
0
Fork 0
mirror of https://github.com/migatu/vtt_work.git synced 2026-07-14 21:38:39 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
692bcb3b4bd156b7d8013e4775476010759153ab
vtt_work/nucleares_monitor/control_board_monitor.py
T
gitea 692bcb3b4b 1.16-interim
2025-10-21 09:18:28 +02:00

3274 lines
155 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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://<host>:<port>/?variable=<NAME>&value=<ARG> (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
import math # NOWE: do decymacji i obliczeń
import traceback
# --- Qt log & DPI: uspokojenie Qt na multi-monitor (opcjonalne, ale pomaga) ---
import os
os.environ.setdefault("QT_LOGGING_RULES", "qt.core.qobject.connect=false")
os.environ.setdefault("QT_ENABLE_HIGHDPI_SCALING", "0")
os.environ.setdefault("QT_AUTO_SCREEN_SCALE_FACTOR", "0")
DEBUG_LOG = True # możesz wyłączyć na False gdy już będzie stabilnie
def _dbg(msg: str):
if not DEBUG_LOG:
return
try:
from datetime import datetime
with open("cbm_debug.log", "a", encoding="utf-8") as f:
f.write(f"{datetime.now().strftime('%H:%M:%S.%f')} {msg}\n")
except Exception:
pass
def _dbg_exc(where: str):
_dbg(f"[EXC] {where}\n{traceback.format_exc()}")
SAFE_GLOBALS = {
"__builtins__": {},
"abs": abs, "min": min, "max": max, "round": round,
"int": int, "float": float, "pow": pow, "math": math,
}
def safe_eval(expr: str, symbols: dict):
"""Bardzo ograniczony eval – tylko dopuszczone funkcje i przekazane symbole."""
clean = {k: v for k, v in (symbols or {}).items() if v is not None}
return eval(expr, SAFE_GLOBALS, clean)
def eval_user_expression(expr: str, x, y, z) -> float:
"""Eval wyrażenia użytkownika (np. w schedulerze) na (x,y,z)."""
val = safe_eval(expr, {"x": x, "y": y, "z": z})
return float(val)
# Spójna warstwa Qt z pyqtgraph (nie mieszamy bezpośrednio PyQt5/PySide6)
_pyqtgraph_available = False
_pyqtgraph_unavailable_reason = ""
try:
import pyqtgraph as pg
from pyqtgraph.Qt import QtCore, QtWidgets
_pyqtgraph_available = True
except Exception as e:
_pyqtgraph_unavailable_reason = str(e)
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
# --- Daemon thread spawner ----------------------------------------------------
def spawn_daemon(name: str, target, *args, **kwargs):
"""
Start background thread as daemon=True. Returns the Thread object.
"""
t = threading.Thread(target=target, args=args, kwargs=kwargs, daemon=True, name=name)
t.start()
return t
# =====================
# 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.16 (+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"<b>([A-Z0-9_]+)</b>", post_html)
seen = set(); uniq: List[str] = []
for n in names:
if n not in seen:
seen.add(n); uniq.append(n)
return uniq
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))
hist_ver: int = 0 # rośnie przy każdym append do history (optymalizacja pod ploty)
@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 stop(self) -> None:
self._stop.set()
def run(self) -> None:
# Pętla bez time.sleep; tylko _stop.wait(...)
while not self._stop.is_set():
now = time.time()
to_run: List[ActionTask] = []
with self._lock:
# wybierz zadania do uruchomienia
for t in list(self.tasks.values()):
if not t.enabled:
continue
if t.interval_s == 0:
# jednorazowe – uruchom dokładnie raz (gdy nie było jeszcze run)
if t.last_run <= 0:
to_run.append(t)
else:
if now >= t.next_run:
to_run.append(t)
# wykonaj poza lockiem
for t in to_run:
try:
code, msg, _hdrs = self.run_task_once(t)
t.last_run = time.time()
if t.interval_s == 0:
# jednorazowe – wyłącz po wykonaniu
t.enabled = False
else:
t.next_run = t.last_run + max(0.0, float(t.interval_s))
except Exception:
# nie blokujemy pętli scheduler’a na wyjątkach z pojedynczego taska
_dbg_exc("Scheduler.run->run_task_once")
# krótka, przerywalna drzemka
# (nie używamy time.sleep; dzięki temu zamknięcie jest natychmiastowe)
self._stop.wait(0.01)
# =====================
# 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()
try:
self.ui_queue.put(("cycle_start", datetime.now().strftime("%H:%M:%S")))
# główny odczyt (BATCH lub fallback na pojedyncze)
status, body, _ = http_get(base_url, {"variable": "WEBSERVER_BATCH_GET"})
if status == 200:
values_map = parse_batch_values(body)
if values_map:
lower_map = {k.lower(): v for k, v in values_map.items()}
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)))
except Exception:
# nigdy nie wywalamy wątku na zewnątrz
_dbg_exc("Poller.run(main loop)")
# OBSŁUGA PAUZY – aktywnie czekamy, ale przerywalnie
while self.paused_event.is_set() and not self.stop_event.is_set():
# czekaj 50ms lub do przerwania
self.stop_event.wait(0.05)
# Odmierz pozostały czas do końca interwału – też przerywalnie
cycle_dt = time.time() - cycle_start
remaining = float(self.refresh_interval) - cycle_dt
if remaining > 0:
waited = 0.0
# czekamy w małych porcjach, by reagować na pause/stop
while waited < remaining and not self.stop_event.is_set():
if self.paused_event.is_set():
break
chunk = min(0.1, remaining - waited)
self.stop_event.wait(chunk)
waited += chunk
# sygnalizacja zakończenia
try:
self.ui_queue.put(("stopped", ""))
except Exception:
_dbg_exc("Poller.run(stop)")
# =====================
# GUI
# =====================
class App(tk.Tk):
def __init__(self) -> None:
super().__init__()
self.title("Control Board Monitor — v1.16")
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._closing = False
self._stop_event = threading.Event()
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)
self.display_show_thresholds_var = tk.BooleanVar(value=True)
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("<Button-3>", self.on_tree_right_click)
self.tree.bind("<Double-1>", 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)
# DOMYŚLNE USTAWIENIA WYŚWIETLANIA (trzymane w tk.Variable, będą też zapisane w konfigu)
self.display_backend_var = tk.StringVar(value="matplotlib") # "matplotlib" | "canvas" | "pyqtgraph"
self.default_samples_var = tk.IntVar(value=200) # domyślna liczba próbek na wykres
self.default_maxpts_var = tk.IntVar(value=400) # ile maks. punktów realnie rysować
# Menu „Opcje” -> „Display”
menubar = tk.Menu(self) # <-- najpierw twórz menubar
opmenu = tk.Menu(menubar, tearoff=0)
display_menu = tk.Menu(opmenu, tearoff=0)
display_menu.add_radiobutton(label="Backend: Matplotlib",
variable=self.display_backend_var, value="matplotlib")
display_menu.add_radiobutton(label="Backend: Canvas (lite)",
variable=self.display_backend_var, value="canvas")
display_menu.add_radiobutton(label="Backend: PyQtGraph",
variable=self.display_backend_var, value="pyqtgraph",
state=("normal" if _pyqtgraph_available else "disabled"))
display_menu.add_separator()
display_menu.add_command(label="Display Defaults…",
command=self.open_display_defaults_dialog)
# Sklej podmenu Display pod „Opcje”
opmenu.add_cascade(label="Display", menu=display_menu)
opmenu.add_separator()
opmenu.add_command(label="Arrange plot windows", command=self.arrange_plot_windows)
opmenu.add_command(label="Count plot windows", command=lambda: tk.messagebox.showinfo("Plots", f"Open plots: {self.count_plot_windows()}"))
# Reszta menubar
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)
# Dodaj „Opcje” do menubar na końcu
menubar.add_cascade(label="Opcje", menu=opmenu)
# Podłącz menubar do okna
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("<Button-3>", self.on_actions_right_click)
if _pyqtgraph_available:
self._qt_ensure_app()
self.after(16, self._pump_qt_events)
def _qt_ensure_app(self):
"""Utwórz (raz) QApplication – wspólną dla wszystkich okien Qt."""
try:
from pyqtgraph.Qt import QtWidgets
except Exception:
self._qt_app = None
return
if getattr(self, "_qt_app", None) is None:
self._qt_app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
def _pump_qt_events(self):
try:
if not getattr(self, "_plot_windows", None):
return # nic do pompowania
if not any(getattr(w, "_backend", "") == "pyqtgraph" for w in self._plot_windows.values()):
return
from pyqtgraph.Qt import QtWidgets
app = getattr(self, "_qt_app", None) or QtWidgets.QApplication.instance()
if app is not None:
app.processEvents()
except Exception as e:
_dbg_exc("_pump_qt_events")
if not getattr(self, "_closing", False):
self.after(16, self._pump_qt_events)
def _safe_bool(self, tkvar, default=False):
try:
return bool(tkvar.get())
except Exception:
return default
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("<Configure>", lambda e: canvas.configure(scrollregion=canvas.bbox("all")))
canvas.create_window((0,0), window=inner, anchor="nw")
canvas.configure(yscrollcommand=vsb.set)
canvas.pack(side="left", fill="both", expand=True); vsb.pack(side="right", fill="y")
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("<MouseWheel>", _mw)
inner.bind("<MouseWheel>", _mw)
canvas.bind("<Button-4>", _mw_up)
canvas.bind("<Button-5>", _mw_down)
inner.bind("<Button-4>", _mw_up)
inner.bind("<Button-5>", _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 getattr(self, "_plot_timer", None) is None:
try:
ms = int(max(150, float(self.refresh_var.get()) * 1000)) # >= ~6 FPS
except Exception:
ms = 300
if self._closing:
return
self._plot_timer = self.after(ms, self._plot_tick) if not self._closing else None
def _plot_tick(self):
windows = [w for w in getattr(self, "_plot_windows", {}).values()
if w.winfo_exists() and getattr(w, "_canvas", None)]
if not windows:
self._plot_timer = None
return
for win in windows:
try:
key = getattr(win, "_key", None)
vi = self.vars.get(key)
if not vi:
continue
# ustaw tytuł okna: nazwa + aktualna wartość (bez nadmiernego odświeżania)
val_num = parse_first_float(vi.last_value)
new_title = f"{vi.display_name} — {val_num:.6g}" if val_num is not None else f"{vi.display_name}"
if getattr(win, "_last_title", None) != new_title:
win.title(new_title)
win._last_title = new_title
# pomiń jeśli brak nowych danych od ostatniego rysowania
last_ver = getattr(win, "_last_seen_ver", -1)
if getattr(vi, "hist_ver", 0) == last_ver:
continue
# pobierz dane (per-okno długość próbek)
try:
n = int(win._sample_len_var.get())
except Exception:
n = 200
vals = list(vi.history)[-n:] if hasattr(vi, "history") else []
dels = list(vi.history_delta)[-n:] if hasattr(vi, "history_delta") else []
# decymacja (ogranicz rysowane punkty do _max_draw_pts)
try:
max_pts = int(getattr(win, "_max_draw_pts", int(win._max_draw_var.get())))
except Exception:
max_pts = 400
if len(vals) > max_pts:
step = max(1, math.ceil(len(vals) / max_pts))
vals_draw = vals[::step]
xs_v = list(range(0, len(vals), step))[:len(vals_draw)]
else:
vals_draw = vals
xs_v = list(range(len(vals)))
win._line.set_data(xs_v, vals_draw)
if vals_draw:
vmin, vmax = min(vals_draw), max(vals_draw)
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)
# --- progi i wskaźnik najbliższego progu (Matplotlib) ---
if getattr(win, "_backend", "") == "matplotlib":
show_thr = bool(getattr(win, "_show_thr_var", tk.BooleanVar(value=True)).get())
for ln in win._thr_lines.values():
ln.set_visible(False)
win._tri_val.set_visible(False)
win._tri_delta.set_visible(False)
if show_thr and vals_draw:
thr = self._effective_thresholds(vi)
vmin, vmax = win._ax_val.get_ylim()
def within(y): return (y is not None) and (vmin <= y <= vmax)
for thr_name in ("dead_low","low","mid","high","extreme"):
y = thr.get(thr_name)
if within(y):
win._thr_lines[thr_name].set_ydata([y, y])
win._thr_lines[thr_name].set_visible(True)
# wskaźnik: najbliższy próg do aktualnej wartości
cur = parse_first_float(vi.last_value)
candidates = [(k, thr[k]) for k in ("dead_low","low","mid","high","extreme") if thr.get(k) is not None]
if cur is not None and candidates:
k_best, y_best = min(candidates, key=lambda kv: abs(kv[1]-cur))
# kolor zgodny z linią
color_map = {
"dead_low": "#FBC02D", "low": "#FFF59D", "mid": "#2ECC71",
"high": "#FF8A80", "extreme": "#D50000",
}
col = color_map.get(k_best, "#2ECC71")
# trójkąt w dół, jeśli próg < aktualna; w górę, jeśli próg > aktualna
up = (y_best > cur)
char = "▲" if up else "▼"
win._tri_val.set_text(char); win._tri_val.set_color(col); win._tri_val.set_visible(True)
win._tri_delta.set_text(char); win._tri_delta.set_color(col); win._tri_delta.set_visible(True)
# słupki delty z taką samą decymacją
if len(dels) > max_pts:
step_b = max(1, math.ceil(len(dels) / max_pts))
dels_draw = dels[::step_b]
xs_b = list(range(0, len(dels), step_b))[:len(dels_draw)]
else:
dels_draw = dels
xs_b = list(range(len(dels)))
if len(dels_draw) != win._bars_len:
for b in win._bars:
b.remove()
win._bars = win._ax_delta.bar(xs_b, dels_draw)
win._bars_len = len(dels_draw)
else:
for b, h in zip(win._bars, dels_draw):
b.set_height(h)
if dels_draw:
dmin, dmax = min(dels_draw), max(dels_draw)
span = (dmax - dmin) or 1.0
m = 0.08 * span
win._ax_delta.set_xlim(0, max(1, len(dels_draw) - 1))
win._ax_delta.set_ylim(dmin - m, dmax + m)
win._ax_delta.set_xlabel(f"last {len(vals)} samples")
# tight_layout tylko po resize
if getattr(win, "_layout_dirty", False):
try:
win._fig.tight_layout()
except Exception:
pass
win._layout_dirty = False
win._canvas.draw_idle()
win._last_seen_ver = getattr(vi, "hist_ver", win._last_seen_ver)
except Exception:
_dbg_exc("_plot_tick(one window)")
continue
# kolejny tick
try:
ms = int(max(150, float(self.refresh_var.get()) * 1000))
except Exception:
ms = 300
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("<MouseWheel>", _mw)
widget.bind("<Button-4>", _mw_up)
widget.bind("<Button-5>", _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)
vi.hist_ver += 1 # NOWE: sygnał dla wykresów, że są nowe dane
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, var_key: str | None = None):
# Ustal monitored key po IID 'var:<key>'
if var_key is None:
sel = self.tree.selection()
if not sel:
tk.messagebox.showwarning("Plot", "Najpierw zaznacz zmienną w tabeli.")
return
iid = sel[0]
if not iid.startswith("var:"):
tk.messagebox.showwarning("Plot", "Wybierz wiersz ze zmienną (nie grupę).")
return
key = iid.split("var:", 1)[1]
else:
key = str(var_key).lower()
if key not in self.vars:
for k, vi in self.vars.items():
if vi.display_name.lower() == str(var_key).lower():
key = k
break
if key not in self.vars:
tk.messagebox.showwarning("Plot", f"Zmienna '{var_key or key}' nie jest monitorowana.")
return
backend = self.display_backend_var.get().lower()
if backend == "canvas":
self._open_canvas_window(key); return
if backend == "pyqtgraph":
if _pyqtgraph_available:
self._open_pyqtgraph_window(key); return
tk.messagebox.showwarning("PyQtGraph", "PyQtGraph nie jest dostępny – użyj Matplotlib lub Canvas.")
return
# === Matplotlib (domyślnie) ===
import matplotlib
matplotlib.use("Agg")
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from matplotlib.figure import Figure
vi = self.vars[key]
win = tk.Toplevel(self)
# pierwszy tytuł bez wartości; aktualna wartość będzie dopisywana w _plot_tick
win.title(f"{vi.display_name} — plot")
win.geometry("760x420")
win._key = key
win._backend = "matplotlib"
tools = ttk.Frame(win); tools.pack(side=tk.TOP, fill=tk.X)
win._show_thr_var = tk.BooleanVar(value=bool(self.display_show_thresholds_var.get()))
ttk.Checkbutton(tools, text="Show thresholds", variable=win._show_thr_var).pack(side=tk.LEFT, padx=(8,6))
ttk.Label(tools, text="Samples:").pack(side=tk.LEFT, padx=(8,2))
win._sample_len_var = tk.IntVar(value=int(self.default_samples_var.get()))
ttk.Spinbox(tools, from_=20, to=5000, increment=10,
textvariable=win._sample_len_var, width=6).pack(side=tk.LEFT)
ttk.Label(tools, text="Max draw pts:").pack(side=tk.LEFT, padx=(10,2))
win._max_draw_var = tk.IntVar(value=int(self.default_maxpts_var.get()))
sp = ttk.Spinbox(tools, from_=100, to=5000, increment=50,
textvariable=win._max_draw_var, width=6)
sp.pack(side=tk.LEFT)
win._max_draw_pts = int(win._max_draw_var.get())
sp.configure(command=lambda w=win: setattr(w, "_max_draw_pts", int(w._max_draw_var.get())))
fig = Figure(figsize=(8,4), dpi=100)
ax_val = fig.add_subplot(2,1,1)
ax_delta = fig.add_subplot(2,1,2, sharex=ax_val)
ax_val.set_ylabel("value"); ax_delta.set_ylabel("Δ"); ax_delta.set_xlabel("samples")
# threshold lines (na osi wartości)
win._thr_lines = {
"dead_low": ax_val.axhline(0, color="#FBC02D", linewidth=1.2, linestyle="--", visible=False), # ciemny żółty
"low": ax_val.axhline(0, color="#FFF59D", linewidth=1.2, linestyle="--", visible=False), # jasny żółty
"mid": ax_val.axhline(0, color="#2ECC71", linewidth=1.0, linestyle=":", visible=False), # zielony (środek)
"high": ax_val.axhline(0, color="#FF8A80", linewidth=1.2, linestyle="--", visible=False), # jasny czerwony
"extreme": ax_val.axhline(0, color="#D50000", linewidth=1.2, linestyle="--", visible=False), # ciemny czerwony
}
canvas = FigureCanvasTkAgg(fig, master=win)
canvas.draw(); canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)
# wskazniki 'najbliższego progu' (po lewej; poza obszarem osi)
win._tri_val = ax_val.text(-0.03, -0.10, "▲", transform=ax_val.transAxes,
ha="left", va="top", color="#2ECC71", fontsize=12,
clip_on=False, visible=False)
win._tri_delta = ax_delta.text(-0.03, 1.10, "▲", transform=ax_delta.transAxes,
ha="left", va="bottom", color="#2ECC71", fontsize=12,
clip_on=False, visible=False)
(line,) = ax_val.plot([], [], linewidth=1.2, antialiased=False)
bars = ax_delta.bar([], [], linewidth=0, antialiased=False)
win._fig = fig; win._ax_val = ax_val; win._ax_delta = ax_delta
win._canvas = canvas; win._line = line; win._bars = bars; win._bars_len = 0
win._last_seen_ver = -1
win._layout_dirty = True
canvas.mpl_connect("resize_event", lambda _evt=None: setattr(win, "_layout_dirty", True))
if not hasattr(self, "_plot_windows"):
self._plot_windows = {}
self._plot_windows[key] = win
self._ensure_plot_timer()
def open_display_defaults_dialog(self):
dlg = tk.Toplevel(self)
dlg.title("Display defaults")
dlg.geometry("360x220")
dlg.transient(self)
dlg.grab_set()
# Backend
ttk.Label(dlg, text="Default backend:").pack(anchor="w", padx=10, pady=(10,2))
back = tk.StringVar(value=self.display_backend_var.get())
frm = ttk.Frame(dlg); frm.pack(anchor="w", padx=10)
ttk.Radiobutton(frm, text="Matplotlib", variable=back, value="matplotlib").pack(side=tk.LEFT, padx=(0,10))
ttk.Radiobutton(frm, text="Canvas (lite)", variable=back, value="canvas").pack(side=tk.LEFT, padx=(0,10))
state = "normal" if _pyqtgraph_available else "disabled"
ttk.Radiobutton(frm, text="PyQtGraph", variable=back, value="pyqtgraph", state=state).pack(side=tk.LEFT)
# Samples / max points
box = ttk.Frame(dlg); box.pack(fill=tk.X, padx=10, pady=(12,2))
ttk.Label(box, text="Default samples:").grid(row=0, column=0, sticky="w")
ds = tk.IntVar(value=int(self.default_samples_var.get()))
ttk.Spinbox(box, from_=20, to=5000, increment=10, textvariable=ds, width=6).grid(row=0, column=1, padx=6)
ttk.Label(box, text="Default max draw pts:").grid(row=1, column=0, sticky="w", pady=(6,0))
md = tk.IntVar(value=int(self.default_maxpts_var.get()))
ttk.Spinbox(box, from_=100, to=5000, increment=50, textvariable=md, width=6).grid(row=1, column=1, padx=6, pady=(6,0))
# Show thresholds on plots
chk_box = ttk.Frame(dlg); chk_box.pack(fill=tk.X, padx=10, pady=(10,0))
show_thr_local = tk.BooleanVar(value=bool(self.display_show_thresholds_var.get()))
ttk.Checkbutton(chk_box, text="Show thresholds (plots)", variable=show_thr_local).pack(anchor="w")
# Buttons
bar = ttk.Frame(dlg); bar.pack(fill=tk.X, pady=12, padx=10)
def apply():
self.display_backend_var.set(back.get())
self.default_samples_var.set(int(ds.get()))
self.default_maxpts_var.set(int(md.get()))
self.display_show_thresholds_var.set(bool(show_thr_local.get()))
dlg.destroy()
tk.messagebox.showinfo("Display defaults",
"Zapisz te ustawienia na stałe przez File → Save Configuration.\n"
"Nowe wykresy będą używać tych domyślnych wartości.")
ttk.Button(bar, text="OK", command=apply).pack(side=tk.RIGHT, padx=6)
ttk.Button(bar, text="Cancel", command=dlg.destroy).pack(side=tk.RIGHT)
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("<Configure>", _on_cfg)
canvas.bind("<Configure>", _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 _effective_thresholds(self, vi:"WarInfo") -> dict:
return self._compute_effective_thresholds(vi)
def _compute_effective_thresholds(self, vi: "VarInfo") -> dict[str, float | None]:
"""
Liczy progi efektywne: bierzemy liczby z konfiguracji + nadpisujemy je, jeśli
zdefiniowano formuły expr_thr_*. Zwracamy też 'mid'=(low+high)/2 jeżeli możliwe.
"""
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(num, expr):
if expr:
try:
v = eval_threshold_expr(expr, stats_x, stats_y, stats_z)
if v is not None:
return float(v)
except Exception:
_dbg_exc("eval_threshold_expr")
return float(num) if num is not None else None
dead = eff(getattr(t, "dead_low", None), getattr(t, "expr_thr_dead_low", None))
low = eff(getattr(t, "low", None), getattr(t, "expr_thr_low", None))
high = eff(getattr(t, "high", None), getattr(t, "expr_thr_high", None))
ext = eff(getattr(t, "extreme_high", None), getattr(t, "expr_thr_extreme_high", None))
mid = (low + high) / 2.0 if (low is not None and high is not None) else None
return {"dead_low": dead, "low": low, "high": high, "extreme": ext, "mid": mid}
def evaluate_thresholds(self, vi: VarInfo, value: str) -> None:
try:
x = parse_first_float(value)
thr = self._compute_effective_thresholds(vi)
dead, low, high, ext = thr["dead_low"], thr["low"], thr["high"], thr["extreme"]
prev = vi.last_state
state = "OPERATING"
if x is None:
state = "OPERATING"
else:
dl = dead if dead is not None else -float("inf")
lo = low if low is not None else -float("inf")
hi = high if high is not None else float("inf")
ex = ext if 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
except Exception:
_dbg_exc("evaluate_thresholds")
def _state_color(self, vi, x_value=None) -> str:
"""Kolor tła w hex na bazie progów efektywnych; odporne na None."""
try:
thr = self._compute_effective_thresholds(vi)
thr_dead = thr["dead_low"]
thr_low = thr["low"]
thr_high = thr["high"]
thr_ext = thr["extreme"]
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:
return "#E9ECEF" # brak danych
if thr_dead is None:
thr_dead = -float("inf")
if thr_low is None and thr_high is None:
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")
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(c1[2] + (c2[2]-c1[2])*t),
)
Y_LO, Y_HI = (255, 244, 178), (255, 149, 0)
G_OK = (46, 204, 113)
R_LO, R_HI = (255, 138, 128), (213, 0, 0)
if x <= thr_low:
denom = (thr_low - thr_dead) if (thr_low > thr_dead) else 1.0
t = (thr_low - x) / max(denom, 1e-9)
return _hex(_lerp(Y_LO, Y_HI, t))
if x <= thr_high:
return _hex(G_OK)
denom = (thr_ext - thr_high) if (thr_ext > thr_high) else 1.0
t = (x - thr_high) / max(denom, 1e-9)
return _hex(_lerp(R_LO, R_HI, t))
except Exception:
_dbg_exc("_state_color")
return "#E9ECEF"
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("<KeyRelease>", 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()],
}
cfg["display_defaults"] = {
"backend": self.display_backend_var.get(),
"samples": int(self.default_samples_var.get()),
"max_draw_pts": int(self.default_maxpts_var.get()),
"show_thresholds": bool(self.display_show_thresholds_var.get()),
}
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
dd = cfg.get("display_defaults", {})
try:
self.display_backend_var.set(dd.get("backend", "matplotlib"))
except Exception:
pass
try:
self.default_samples_var.set(int(dd.get("samples", 200)))
except Exception:
pass
try:
self.default_maxpts_var.set(int(dd.get("max_draw_pts", 400)))
except Exception:
pass
try:
self.display_show_thresholds_var.set(bool(dd.get("show_thresholds", True)))
except Exception:
pass
# 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:
# zatrzymaj Pollera
self._closing = True
# zatrzymaj okna/timery Qt (jeśli są)
try:
for _k, tup in list(getattr(self, "_qt_windows", {}).items()):
try:
win, timer = tup
except Exception:
continue
try:
timer.stop(); timer.deleteLater()
except Exception:
pass
try:
win.close()
except Exception:
pass
self._qt_windows.clear()
except Exception:
pass
try:
if getattr(self, "poller", None) and self.poller.is_alive():
self.stop_event.set() # natychmiast przerwie wait() w pętli
self.paused_event.clear()
self.poller.join(timeout=1.5)
except Exception:
pass
# zatrzymaj scheduler
try:
if getattr(self, "scheduler", None) and self.scheduler.is_alive():
self.scheduler.stop() # ustawia wewnętrzny _stop Event
self.scheduler.join(timeout=1.5)
except Exception:
pass
# zatrzymaj timer do plotów, jeśli działa
try:
if getattr(self, "_plot_timer", None):
self.after_cancel(self._plot_timer)
except Exception:
pass
# zamknij okna wykresów
try:
for win in list(self._plot_windows.values()):
try:
win.destroy()
except Exception:
pass
except Exception:
pass
self.destroy()
def _open_canvas_window(self, key: str):
vi = self.vars[key]
win = tk.Toplevel(self)
win.title(f"{vi.display_name} — canvas")
win.geometry("760x420")
win._backend = "canvas"
win._key = key
tools = ttk.Frame(win); tools.pack(side=tk.TOP, fill=tk.X)
ttk.Label(tools, text="Samples:").pack(side=tk.LEFT, padx=(8,2))
win._sample_len_var = tk.IntVar(value=int(self.default_samples_var.get()))
ttk.Spinbox(tools, from_=20, to=5000, increment=10,
textvariable=win._sample_len_var, width=6).pack(side=tk.LEFT)
ttk.Label(tools, text="Max draw pts:").pack(side=tk.LEFT, padx=(10,2))
win._max_draw_var = tk.IntVar(value=int(self.default_maxpts_var.get()))
ttk.Spinbox(tools, from_=100, to=5000, increment=50,
textvariable=win._max_draw_var, width=6).pack(side=tk.LEFT)
# Dwie części: górna (linia wartości), dolna (słupki Δ)
win._canvas = tk.Canvas(win, bg="#ffffff", highlightthickness=0)
win._canvas.pack(fill=tk.BOTH, expand=True)
# dane do rysowania
win._line_items = [] # lista segmentów linii
win._bar_items = [] # lista prostokątów
def _tick():
if not win.winfo_exists():
return
key_local = getattr(win, "_key", None)
vi_local = self.vars.get(key_local)
if not vi_local:
win.after(400, _tick); return
# Title z wartością
try:
val_num = parse_first_float(vi_local.last_value)
win.title(f"{vi_local.display_name} — {val_num:.6g}" if val_num is not None else vi_local.display_name)
except Exception:
pass
# …po utworzeniu Toplevel win…
if not hasattr(self, "_plot_windows_canvas"):
self._plot_windows_canvas = []
self._plot_windows_canvas.append(win)
def _on_close_canvas():
try:
self._plot_windows_canvas.remove(win)
except Exception:
pass
win.destroy()
win.protocol("WM_DELETE_WINDOW", _on_close_canvas)
# dane
try: n = int(win._sample_len_var.get())
except Exception: n = 200
try: max_pts = int(win._max_draw_var.get())
except Exception: max_pts = 400
vals = list(vi_local.history)[-n:]
dels = list(vi_local.history_delta)[-n:]
# decymacja
def decimate(arr, m):
if len(arr) <= m:
xs = list(range(len(arr))); return xs, arr
step = max(1, int(math.ceil(len(arr) / m)))
arr2 = arr[::step]; xs = list(range(0, len(arr), step))[:len(arr2)]
return xs, arr2
xs_v, vals_d = decimate(vals, max_pts)
xs_b, dels_d = decimate(dels, max_pts)
# geometra płótna
cw = max(10, win._canvas.winfo_width())
ch = max(10, win._canvas.winfo_height())
mid = ch // 2
top_h = int(ch * 0.62)
bot_y0 = top_h + 1
win._canvas.delete("all")
# oś X mapping
def map_x(i, npts):
return int((i / max(1, npts-1)) * (cw-20)) + 10
# linia wartości
if vals_d:
vmin, vmax = min(vals_d), max(vals_d)
span = (vmax - vmin) or 1.0
def map_y(v):
return int((1 - (v - vmin) / span) * (top_h-20)) + 10
last = None
for i, v in enumerate(vals_d):
x = map_x(i, len(vals_d)); y = map_y(v)
if last is not None:
win._canvas.create_line(last[0], last[1], x, y)
last = (x, y)
# słupki Δ
if dels_d:
dmin, dmax = min(dels_d), max(dels_d)
span = (dmax - dmin) or 1.0
def map_yb(v):
# dolny panel
y = int((1 - (v - dmin)/span) * (ch - bot_y0 - 20)) + bot_y0 + 10
return y
bw = max(1, int((cw-20) / max(1, len(dels_d))))
for i, v in enumerate(dels_d):
x = map_x(i, len(dels_d)); y = map_yb(v)
win._canvas.create_rectangle(x, y, x+bw, ch-8, outline="", fill="#8888ff")
# kolejny tick
if self._closing or not win.winfo_exists():
return
try:
ms = int(max(120, float(self.refresh_var.get())*1000))
except Exception:
ms = 250
win.after(ms, _tick)
_tick()
def _open_pyqtgraph_window(self, key: str):
if not _pyqtgraph_available:
tk.messagebox.showwarning("PyQtGraph", "PyQtGraph nie jest dostępny – użyj Matplotlib lub Canvas.")
return
self._qt_ensure_app()
if key not in self.vars:
tk.messagebox.showwarning("Plot", f"Zmienna '{key}' nie jest monitorowana.")
return
vi = self.vars[key]
# ... wcześniej: vi = self.vars[key]
glw = pg.GraphicsLayoutWidget(title=f"{vi.display_name}")
p1 = glw.addPlot(row=0, col=0); p1.showGrid(x=True, y=True)
p2 = glw.addPlot(row=1, col=0); p2.showGrid(x=True, y=True)
curve = p1.plot([], [], pen=pg.mkPen(width=1))
bars = p2.plot([], [], pen=None, symbol=None, fillLevel=0,
brush=(100, 100, 255, 160), stepMode=False)
# linie progów – tworzone raz
glw._thr_lines = {
"dead_low": pg.InfiniteLine(pos=0, angle=0, pen=pg.mkPen("#FBC02D", width=1), movable=False),
"low": pg.InfiniteLine(pos=0, angle=0, pen=pg.mkPen("#FFF59D", width=1), movable=False),
"mid": pg.InfiniteLine(pos=0, angle=0, pen=pg.mkPen("#2ECC71", width=1, style=QtCore.Qt.DotLine), movable=False),
"high": pg.InfiniteLine(pos=0, angle=0, pen=pg.mkPen("#FF8A80", width=1), movable=False),
"extreme": pg.InfiniteLine(pos=0, angle=0, pen=pg.mkPen("#D50000", width=1), movable=False),
}
for _ln in glw._thr_lines.values():
_ln.setVisible(False); p1.addItem(_ln)
glw._tri = pg.TextItem("", color=(46,204,113)); glw._tri.setVisible(False); p1.addItem(glw._tri)
# okno
glw.resize(820, 520)
glw.setWindowTitle(f"{vi.display_name} — pyqtgraph")
glw.show()
glw._key = key
glw._display_name = vi.display_name
# rejestr okien
if not hasattr(self, "_plot_windows_qt"):
self._plot_windows_qt = []
self._plot_windows_qt.append(glw)
# sprzątanie
def _on_close(evt):
try:
t = getattr(glw, "_timer", None)
if t: t.stop(); t.deleteLater()
except Exception:
pass
try:
self._plot_windows_qt.remove(glw)
except Exception:
pass
evt.accept()
glw.closeEvent = _on_close
# parametry rysowania
try:
samples = int(self.default_samples_var.get())
except Exception:
samples = 100
try:
maxpts = int(self.default_maxpts_var.get())
except Exception:
maxpts = 500
def decimate(arr, m):
if len(arr) <= m:
return list(range(len(arr))), arr
step = max(1, int(math.ceil(len(arr) / m)))
arr2 = arr[::step]
xs = list(range(0, len(arr), step))[:len(arr2)]
return xs, arr2
# reentrancy guard + log
glw._in_update = False
def _log_qt(msg):
try:
with open("qt_plot_err.log", "a", encoding="utf-8") as f:
f.write(msg + "\n")
except Exception:
pass
def update():
if glw._in_update:
return
glw._in_update = True
try:
vi_local = self.vars.get(key)
if not vi_local:
return
vals = list(vi_local.history)[-samples:]
dels = list(vi_local.history_delta)[-samples:]
xs_v, vals_d = decimate(vals, maxpts)
xs_b, dels_d = decimate(dels, maxpts)
curve.setData(xs_v, vals_d)
bars.setData(xs_b, dels_d)
# progi i wskaźnik
for _ln in glw._thr_lines.values():
_ln.setVisible(False)
glw._tri.setVisible(False)
show_thr = self._safe_bool(self.display_show_thresholds_var, False)
if show_thr and vals_d:
thr = self._effective_thresholds(vi_local)
# UWAGA: bierz zakres z danych, nie z viewRange (viewRange potrafi być zmienne w trakcie resize)
ymin, ymax = min(vals_d), max(vals_d)
def within(y): return (y is not None) and (ymin <= y <= ymax)
for thr_name in ("dead_low","low","mid","high","extreme"):
y = thr.get(thr_name)
if within(y):
glw._thr_lines[thr_name].setPos(y)
glw._thr_lines[thr_name].setVisible(True)
cur = parse_first_float(vi_local.last_value)
candidates = [(n, thr[n]) for n in ("dead_low","low","mid","high","extreme") if thr.get(n) is not None]
if cur is not None and candidates:
k_best, y_best = min(candidates, key=lambda kv: abs(kv[1]-cur))
color_map = {"dead_low": (251,192,45), "low": (255,245,157), "mid": (46,204,113),
"high": (255,138,128), "extreme": (213,0,0)}
col = color_map.get(k_best, (46,204,113))
up = (y_best > cur); char = "▲" if up else "▼"
glw._tri.setColor(col); glw._tri.setText(char)
# ustaw w bezpiecznej pozycji przy lewej krawędzi, w obrębie danych
x_left = xs_v[0] if xs_v else 0
y_pos = ymin if up else ymax
glw._tri.setPos(x_left, y_pos)
glw._tri.setVisible(True)
# tytuł z aktualną wartością
try:
val_num = parse_first_float(vi_local.last_value)
glw.setWindowTitle(f"{vi_local.display_name} — {val_num:.6g}" if val_num is not None else vi_local.display_name)
except Exception:
pass
except Exception as e:
_log_qt(f"[update] {type(e).__name__}: {e}")
finally:
glw._in_update = False
# jeden QTimer – jako child okna
glw._timer = QtCore.QTimer(glw)
try:
ms = max(80, int(float(self.refresh_var.get()) * 1000))
except Exception:
ms = 160
glw._timer.timeout.connect(update)
glw._timer.start(ms)
update()
def count_plot_windows(self) -> int:
n_tk = len([w for w in getattr(self, "_plot_windows", {}).values() if getattr(w, "winfo_exists", lambda: False)()])
n_qt = len([w for w in getattr(self, "_plot_windows_qt", []) if getattr(w, "isVisible", lambda: False)()])
return n_tk + n_qt
# --- helpers: monitory (Windows) + fallback ---
def _enum_monitors(self):
monitors = []
try:
import ctypes
from ctypes import wintypes
class RECT(ctypes.Structure):
_fields_ = [("left", ctypes.c_long), ("top", ctypes.c_long),
("right", ctypes.c_long), ("bottom", ctypes.c_long)]
MONITORENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_int, ctypes.c_ulong, ctypes.c_ulong, ctypes.POINTER(RECT), ctypes.c_double)
user32 = ctypes.windll.user32
def _cb(hMon, hDC, lprc, dwData):
r = lprc.contents
monitors.append((int(r.left), int(r.top), int(r.right), int(r.bottom)))
return 1
user32.EnumDisplayMonitors(0, 0, MONITORENUMPROC(_cb), 0)
if not monitors:
raise RuntimeError("No monitors from EnumDisplayMonitors")
except Exception:
# Fallback: pojedynczy ekran wg Tk
w = self.winfo_screenwidth()
h = self.winfo_screenheight()
monitors = [(0, 0, w, h)]
return monitors
def _win_center(self, x, y, w, h):
return (x + w // 2, y + h // 2)
def _which_monitor(self, monitors, x, y, w, h):
cx, cy = self._win_center(x, y, w, h)
best = 0; best_d = float("inf")
for i, (L, T, R, B) in enumerate(monitors):
if L <= cx <= R and T <= cy <= B:
return i
# odległość do środka monitora
mcx, mcy = (L + R) // 2, (T + B) // 2
d = (mcx - cx) ** 2 + (mcy - cy) ** 2
if d < best_d:
best, best_d = i, d
return best
def _best_grid(self, n):
if n <= 0:
return (1, 1)
c = int(math.ceil(math.sqrt(n)))
r = int(math.ceil(n / c))
return (r, c)
def arrange_plot_windows(self):
"""
Układa wszystkie otwarte okna wykresów (Tk-Matplotlib, Tk-Canvas, Qt-PyQtGraph)
w siatki na poszczególnych monitorach.
Zasady:
- grupujemy okna wg monitora, NA KTÓRYM AKTUALNIE SĄ otwarte,
- w każdej grupie sortujemy alfabetycznie wg nazwy zmiennej (display_name),
- wypełniamy siatkę od lewej do prawej, potem kolejny wiersz,
- jeśli n>=9 na danym monitorze → siatka pełnoekranowa,
jeśli n<9 → okna max ~połowa domyślnego rozmiaru (dodatkowo skalujemy w dół, gdy trzeba),
- zostawiamy marginesy na krawędziach (TOP_MARGIN, SIDE/BOTTOM), żeby tytuły nie „wychodziły” poza ekran.
"""
monitors = self._enum_monitors()
# Zbierz okna (Tk MPL)
tk_wins = []
for w in list(getattr(self, "_plot_windows", {}).values()):
if getattr(w, "winfo_exists", lambda: False)():
try:
x = w.winfo_rootx(); y = w.winfo_rooty()
ww = w.winfo_width(); hh = w.winfo_height()
if ww <= 1 or hh <= 1:
geo = w.geometry() # "WxH+X+Y"
parts = geo.replace("x", "+").split("+")
ww = int(parts[0]); hh = int(parts[1]); x = int(parts[2]); y = int(parts[3])
key = getattr(w, "_key", None)
name = None
if key and key in self.vars:
name = self.vars[key].display_name
if not name:
# fallback: tytuł okna
name = str(w.title() or "")
tk_wins.append(("tk", w, x, y, ww, hh, name))
except Exception:
pass
# Zbierz okna (Tk Canvas)
for w in list(getattr(self, "_plot_windows_canvas", [])):
if getattr(w, "winfo_exists", lambda: False)():
try:
x = w.winfo_rootx(); y = w.winfo_rooty()
ww = w.winfo_width(); hh = w.winfo_height()
if ww <= 1 or hh <= 1:
geo = w.geometry()
parts = geo.replace("x", "+").split("+")
ww = int(parts[0]); hh = int(parts[1]); x = int(parts[2]); y = int(parts[3])
key = getattr(w, "_key", None)
name = None
if key and key in self.vars:
name = self.vars[key].display_name
if not name:
name = str(w.title() or "")
tk_wins.append(("tk", w, x, y, ww, hh, name))
except Exception:
pass
# Zbierz okna (Qt pyqtgraph)
qt_wins = []
for qw in list(getattr(self, "_plot_windows_qt", [])):
if getattr(qw, "isVisible", lambda: False)():
try:
fg = qw.frameGeometry()
x = fg.x(); y = fg.y(); ww = fg.width(); hh = fg.height()
key = getattr(qw, "_key", None)
name = getattr(qw, "_display_name", None)
if not name and key and key in self.vars:
name = self.vars[key].display_name
if not name:
name = str(qw.windowTitle() or "")
qt_wins.append(("qt", qw, x, y, ww, hh, name))
except Exception:
pass
all_wins = tk_wins + qt_wins
if not all_wins:
tk.messagebox.showinfo("Arrange", "Brak otwartych okien wykresów.")
return
# Grupowanie po monitorach (wg aktualnej pozycji)
groups = {i: [] for i in range(len(monitors))}
for kind, win, x, y, ww, hh, name in all_wins:
mid = self._which_monitor(monitors, x, y, ww, hh)
groups[mid].append((kind, win, name))
# USTAWIENIA rozmiaru/marginesów
DEF_W, DEF_H = 760, 420
MAX_W, MAX_H = DEF_W // 2, DEF_H // 2
SIDE_MARGIN = 12
TOP_MARGIN = 40 # większy top, żeby tytuły Qt nie wychodziły
BOTTOM_MARGIN = 12
CELL_PAD = 10 # odstęp między kratkami
# Helper: najlepsza siatka (r,c)
def best_grid(n):
if n <= 0: return (1, 1)
c = int(math.ceil(math.sqrt(n)))
r = int(math.ceil(n / c))
return (r, c)
for midx, items in groups.items():
if not items:
continue
# Sortowanie alfabetycznie wg nazwy zmiennej (display_name)
items.sort(key=lambda it: (str(it[2]).lower(), str(it[2])))
L, T, R, B = monitors[midx]
mon_w = R - L
mon_h = B - T
n = len(items)
rows, cols = best_grid(n)
full = (n >= 9)
# wstępny wymiar kratki
if full:
cell_w = max(1, (mon_w - (cols + 1) * CELL_PAD - 2 * SIDE_MARGIN) // cols)
cell_h = max(1, (mon_h - (rows + 1) * CELL_PAD - TOP_MARGIN - BOTTOM_MARGIN) // rows)
else:
cell_w = min(MAX_W, max(1, (mon_w - (cols + 1) * CELL_PAD - 2 * SIDE_MARGIN) // cols))
cell_h = min(MAX_H, max(1, (mon_h - (rows + 1) * CELL_PAD - TOP_MARGIN - BOTTOM_MARGIN) // rows))
# dopasowanie, jeśli dalej nie mieści
while (cell_w * cols + (cols + 1) * CELL_PAD + 2 * SIDE_MARGIN) > mon_w or \
(cell_h * rows + (rows + 1) * CELL_PAD + TOP_MARGIN + BOTTOM_MARGIN) > mon_h:
cell_w = max(120, int(cell_w * 0.9))
cell_h = max(100, int(cell_h * 0.9))
if cell_w <= 120 and cell_h <= 100:
break
# rozmiar całej siatki + pozycja startowa (centrowanie)
grid_w = cell_w * cols + (cols + 1) * CELL_PAD
grid_h = cell_h * rows + (rows + 1) * CELL_PAD
origin_x = L + SIDE_MARGIN + max(0, (mon_w - grid_w - 2 * SIDE_MARGIN) // 2)
origin_y = T + TOP_MARGIN + max(0, (mon_h - grid_h - TOP_MARGIN - BOTTOM_MARGIN) // 2)
# „bezpieczny” rozmiar (obcięty o mały bufor, różne dekoracje)
cell_w_adj = max(50, cell_w - 8)
cell_h_adj = max(50, cell_h - 12)
# Rozmieszczaj od lewej do prawej, potem kolejny wiersz
i = 0
for c in range(cols):
for r in range(rows):
if i >= n:
break
kind, win, _nm = items[i]
x = origin_x + CELL_PAD + c * (cell_w + CELL_PAD)
y = origin_y + CELL_PAD + r * (cell_h + CELL_PAD)
# „clamp” do granic ekranu (z marginesami)
x = max(L + SIDE_MARGIN, min(x, R - SIDE_MARGIN - cell_w_adj))
y = max(T + TOP_MARGIN, min(y, B - BOTTOM_MARGIN - cell_h_adj))
try:
if kind == "tk":
win.geometry(f"{cell_w_adj}x{cell_h_adj}+{x}+{y}")
else:
win.setGeometry(x, y, cell_w_adj, cell_h_adj)
except Exception:
pass
i += 1
# --- 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, "<thr>", "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, "<expr>", "exec")
exec(code, safe_globals, safe_locals)
if "result" not in safe_locals:
raise ValueError("Snippet must assign to 'result', e.g., result = x * 1.1")
return safe_locals["result"]
else:
code = compile(expr, "<expr>", "eval")
return eval(code, safe_globals, safe_locals)
def main() -> None:
app = App()
import atexit
def _graceful_shutdown():
try:
# jeżeli App jeszcze żyje – wywołaj on_close
# (dopasuj, jeśli trzymasz referencję gdzie indziej)
app.on_close()
except Exception:
pass
atexit.register(_graceful_shutdown)
app.mainloop()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink