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

1.13

Browse Source
This commit is contained in:
Michal Tuszowski
2025-10-20 18:31:44 +02:00
parent 4beb01118a
commit 04923cc449
1 changed files with 131 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nucleares_monitor/control_board_monitor.py
+131 -11
View File
@@ -265,6 +265,11 @@ class Thresholds:
# 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"
@dataclass
class VarInfo:
@@ -297,6 +302,7 @@ class ActionTask:
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
@@ -353,9 +359,27 @@ class ActionScheduler(threading.Thread):
return self.get_value_cb(n) if n else None
except Exception:
return None
x = gv(task.x_src)
y = gv(task.y_src)
z = gv(task.z_src)
# 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:
@@ -486,7 +510,10 @@ class App(tk.Tk):
self.variables_keys: List[str] = [] # list of lower-case keys in selection order
self.functions_list: List[str] = [] # preserve original casing
self.scheduler = ActionScheduler(lambda: build_base_url(self.host_var.get().strip(), int(self.port_var.get())), self.get_current_value)
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)
@@ -653,6 +680,36 @@ class App(tk.Tk):
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)
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:
@@ -1297,6 +1354,7 @@ class App(tk.Tk):
def open_thresholds_dialog(self):
sel = self.tree.selection()
expr_xsrc_vars = {}
if not sel:
messagebox.showinfo("No selection", "Select a variable first.")
return
@@ -1418,6 +1476,20 @@ class App(tk.Tk):
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)
@@ -1458,6 +1530,11 @@ class App(tk.Tk):
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
@@ -1596,9 +1673,23 @@ class App(tk.Tk):
val = parse_first_float(bd)
if label == 'y': y_val = val
else: z_val = val
if expr and x is not None:
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, y_val, z_val)
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)}
@@ -1661,8 +1752,12 @@ class App(tk.Tk):
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)
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
@@ -1920,6 +2015,11 @@ class App(tk.Tk):
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)
@@ -1940,7 +2040,12 @@ class App(tk.Tk):
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))
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()
@@ -2015,6 +2120,11 @@ class App(tk.Tk):
"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,
},
}
def _deserialize_varinfo(self, key: str, data: dict) -> VarInfo:
@@ -2036,14 +2146,24 @@ class App(tk.Tk):
# 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"),
)
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}
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"))
return ActionTask(name=d.get("name",""), value=d.get("value","1"),
interval_s=float(d.get("interval_s", 1.0)), enabled=d.get("enabled", True),
expr=d.get("expr"), x_src=d.get("x_src"), y_src=d.get("y_src"), z_src=d.get("z_src"),
x_mode=d.get("x_mode","raw"))
def on_close(self) -> None:
try:
if self.poller and self.poller.is_alive():