louis/flow_control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完善流量控制 GUI 的开环扫点与手动操作流程

Browse Source 
新增可配置开环扫点、多 CSV 前馈表拟合及结果文件生成；补充闭环停止、手动开度安全互锁和 PID 未启用状态显示，并同步更新自动化测试、使用文档、打包配置与 Windows 可执行程序。
This commit is contained in:
louis
2026-09-04 16:47:16 +08:00
parent ead625bf2d
commit c1a4a2a229
23 changed files with 36300 additions and 7959 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gui_main.py
+556 -16
View File
@@ -5,8 +5,11 @@ from __future__ import annotations
import argparse
import ctypes
from datetime import datetime
from queue import Empty
from pathlib import Path
from queue import Empty, Queue
import sys
from tempfile import TemporaryDirectory
import threading
import tkinter as tk
from tkinter import filedialog, messagebox, ttk
@@ -14,16 +17,202 @@ import config
from control_runtime import (
BUILTIN_MODEL_NAME,
ControlWorker,
OpenLoopSettings,
application_directory,
build_open_loop_points,
build_runtime_logger,
load_model,
validate_device_settings,
validate_opening,
validate_open_loop_settings,
validate_target,
)
from identify_valve import DEFAULT_TAIL, create_diagnostic_plot, identify
WINDOW_TITLE = "气体流量控制"
class OpenLoopDialog:
"""开环扫点参数弹窗。"""
def __init__(self, parent, on_start):
self.parent = parent
self.on_start = on_start
defaults = OpenLoopSettings()
self.window = tk.Toplevel(parent)
self.window.title("开环扫点参数")
self.window.transient(parent)
self.window.resizable(False, False)
self.window.protocol("WM_DELETE_WINDOW", self._cancel)
self.values = {
"min_opening_pct": tk.StringVar(value=f"{defaults.min_opening_pct:g}"),
"max_opening_pct": tk.StringVar(value=f"{defaults.max_opening_pct:g}"),
"opening_step_pct": tk.StringVar(value=f"{defaults.opening_step_pct:g}"),
"random_seed": tk.StringVar(value=str(defaults.random_seed or "")),
"sample_period_s": tk.StringVar(value=f"{defaults.sample_period_s:g}"),
"steady_window_s": tk.StringVar(value=f"{defaults.steady_window_s:g}"),
"steady_std_pct": tk.StringVar(value=f"{defaults.steady_std_pct:g}"),
"max_wait_s": tk.StringVar(value=f"{defaults.max_wait_s:g}"),
"min_abs_std_slm": tk.StringVar(value=f"{defaults.min_abs_std_slm:g}"),
"fine_stroke_start": tk.StringVar(value=f"{defaults.fine_stroke_start:g}"),
"fine_stroke_step": tk.StringVar(value=f"{defaults.fine_stroke_step:g}"),
}
self.include_fine_scan = tk.BooleanVar(value=defaults.include_fine_scan)
self.summary_var = tk.StringVar()
self.fine_entries = []
self._build_ui()
for variable in self.values.values():
variable.trace_add("write", self._refresh_summary)
self.include_fine_scan.trace_add("write", self._fine_scan_changed)
self._fine_scan_changed()
self.window.update_idletasks()
x = parent.winfo_rootx() + (parent.winfo_width() - self.window.winfo_width()) // 2
y = parent.winfo_rooty() + (parent.winfo_height() - self.window.winfo_height()) // 2
self.window.geometry(f"+{max(0, x)}+{max(0, y)}")
self.window.grab_set()
self.window.focus_set()
def _build_ui(self):
outer = ttk.Frame(self.window, padding=14)
outer.pack(fill="both", expand=True)
ttk.Label(
outer,
text="开环扫点会停止闭环，并直接按扫描序列下发阀门开度。",
foreground="#8a4b00",
wraplength=560,
).pack(anchor="w", pady=(0, 9))
basic = ttk.LabelFrame(outer, text="扫描范围", padding=9)
basic.pack(fill="x")
self._add_field(basic, 0, 0, "起始开度", "min_opening_pct", "%")
self._add_field(basic, 0, 3, "结束开度", "max_opening_pct", "%")
self._add_field(basic, 1, 0, "开度间隔", "opening_step_pct", "%")
self._add_field(basic, 1, 3, "随机种子", "random_seed", "（留空=随机）")
steady = ttk.LabelFrame(outer, text="采样与稳态判定", padding=9)
steady.pack(fill="x", pady=(9, 0))
self._add_field(steady, 0, 0, "采样周期", "sample_period_s", "s")
self._add_field(steady, 0, 3, "稳态窗口", "steady_window_s", "s")
self._add_field(steady, 1, 0, "相对标准差", "steady_std_pct", "%")
self._add_field(steady, 1, 3, "绝对标准差下限", "min_abs_std_slm", "SLM")
self._add_field(steady, 2, 0, "单点最长等待", "max_wait_s", "s")
fine = ttk.LabelFrame(outer, text="闭阀端精细扫描", padding=9)
fine.pack(fill="x", pady=(9, 0))
ttk.Checkbutton(
fine,
text="追加精细行程点",
variable=self.include_fine_scan,
).grid(row=0, column=0, columnspan=3, sticky="w")
start_entry = self._add_field(
fine, 1, 0, "起始行程", "fine_stroke_start", "（终点 1000）"
)
step_entry = self._add_field(
fine, 1, 3, "行程间隔", "fine_stroke_step", ""
)
self.fine_entries.extend((start_entry, step_entry))
summary = tk.Label(
outer,
textvariable=self.summary_var,
bg="#edf5ff",
fg="#174f85",
anchor="w",
justify="left",
padx=9,
pady=7,
)
summary.pack(fill="x", pady=(9, 0))
output_dir = application_directory() / "open_loop_data"
ttk.Label(
outer,
text=f"输出目录：{output_dir}（每次生成 CSV，存在采样时同时生成 PNG）",
foreground="#5f6368",
wraplength=560,
).pack(anchor="w", pady=(7, 0))
buttons = ttk.Frame(outer)
buttons.pack(fill="x", pady=(12, 0))
ttk.Button(buttons, text="取消", command=self._cancel, width=12).pack(
side="right"
)
ttk.Button(
buttons,
text="开始扫点",
command=self._start,
style="Primary.TButton",
width=14,
).pack(side="right", padx=(0, 8))
def _add_field(self, parent, row, column, label, key, unit):
ttk.Label(parent, text=label).grid(
row=row, column=column, sticky="w", pady=(5 if row else 0, 0)
)
entry = ttk.Entry(parent, textvariable=self.values[key], width=10)
entry.grid(
row=row,
column=column + 1,
sticky="w",
padx=(6, 4),
pady=(5 if row else 0, 0),
)
ttk.Label(parent, text=unit).grid(
row=row,
column=column + 2,
sticky="w",
padx=(0, 16),
pady=(5 if row else 0, 0),
)
return entry
def _settings(self):
return validate_open_loop_settings(
**{key: variable.get() for key, variable in self.values.items()},
include_fine_scan=self.include_fine_scan.get(),
)
def _refresh_summary(self, *_args):
try:
settings = self._settings()
point_count = len(build_open_loop_points(settings))
maximum_minutes = point_count * settings.max_wait_s / 60.0
self.summary_var.set(
f"共 {point_count} 个扫描点；按单点最长等待估算，上限约 "
f"{maximum_minutes:.1f} 分钟（稳定后会提前切换）。"
)
except ValueError as exc:
self.summary_var.set(f"参数待修正：{exc}")
def _fine_scan_changed(self, *_args):
state = "normal" if self.include_fine_scan.get() else "disabled"
for entry in self.fine_entries:
entry.configure(state=state)
self._refresh_summary()
def _start(self):
try:
settings = self._settings()
except ValueError as exc:
messagebox.showerror("扫点参数无效", str(exc), parent=self.window)
return
if self.on_start(settings):
self.window.grab_release()
self.window.destroy()
def _cancel(self):
try:
self.window.grab_release()
except tk.TclError:
pass
self.window.destroy()
class FlowControlApp:
"""单窗口操作界面。"""
@@ -32,6 +221,8 @@ class FlowControlApp:
"CONNECTING": "正在连接",
"MONITORING": "已连接 / 监测中",
"CONTROLLING": "闭环运行中",
"OPEN_LOOP": "开环扫点中",
"FINALIZING": "正在生成结果",
"DISCONNECTING": "正在断开",
"STOPPING": "正在退出",
"FAULT": "故障 / 已断开",
@@ -42,6 +233,8 @@ class FlowControlApp:
"CONNECTING": ("#875a00", "#fff3cd"),
"MONITORING": ("#0b6b3a", "#dff4e8"),
"CONTROLLING": ("#075cad", "#deefff"),
"OPEN_LOOP": ("#7a3e00", "#ffe9cc"),
"FINALIZING": ("#875a00", "#fff3cd"),
"DISCONNECTING": ("#875a00", "#fff3cd"),
"STOPPING": ("#875a00", "#fff3cd"),
"FAULT": ("#a61b1b", "#fde5e5"),
@@ -50,8 +243,8 @@ class FlowControlApp:
def __init__(self, root):
self.root = root
self.root.title(WINDOW_TITLE)
self.root.geometry("860x720")
self.root.minsize(780, 640)
self.root.geometry("860x760")
self.root.minsize(780, 700)
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
self.logger = build_runtime_logger()
@@ -60,12 +253,15 @@ class FlowControlApp:
self.closing = False
self.selected_model_path = None
self.model_info = load_model()
self.fit_events = Queue()
self.fit_in_progress = False
self.host_var = tk.StringVar(value=config.PLC_HOST)
self.port_var = tk.StringVar(value=str(config.PLC_PORT))
self.slave_var = tk.StringVar(value=str(config.PLC_SLAVE_ID))
self.model_var = tk.StringVar(value=self.model_info.display_name)
self.target_var = tk.StringVar(value="")
self.manual_opening_var = tk.StringVar(value="100")
self.active_target_var = tk.StringVar(value="--")
self.status_var = tk.StringVar(value="未连接")
self.detail_var = tk.StringVar(value="请选择模型并连接设备")
@@ -156,8 +352,12 @@ class FlowControlApp:
foreground="#174f85",
)
self.model_label.grid(
row=1, column=1, columnspan=3, sticky="w", padx=(7, 10), pady=(10, 0)
row=1, column=1, columnspan=2, sticky="w", padx=(7, 10), pady=(10, 0)
)
self.open_loop_button = ttk.Button(
connection, text="开环扫点", command=self._open_open_loop_dialog
)
self.open_loop_button.grid(row=1, column=3, pady=(10, 0), padx=(0, 7))
self.model_browse_button = ttk.Button(
connection, text="选择 JSON…", command=self._choose_model
)
@@ -185,6 +385,13 @@ class FlowControlApp:
width=16,
)
self.disconnect_button.pack(side="left", padx=(10, 0))
self.fit_model_button = ttk.Button(
button_row,
text="拟合前馈表",
command=self._fit_feedforward_model,
width=16,
)
self.fit_model_button.pack(side="right")
target_frame = ttk.LabelFrame(
outer,
@@ -195,25 +402,34 @@ class FlowControlApp:
target_frame.pack(fill="x", pady=(10, 0))
target_frame.columnconfigure(1, weight=1)
ttk.Label(target_frame, text="目标值").grid(row=0, column=0, sticky="w")
target_input = ttk.Frame(target_frame)
target_input.grid(row=0, column=1, sticky="w", padx=(8, 5))
self.target_entry = ttk.Entry(
target_frame,
target_input,
textvariable=self.target_var,
width=18,
width=9,
font=("Microsoft YaHei UI", 12),
)
self.target_entry.grid(row=0, column=1, sticky="w", padx=(8, 5))
self.target_entry.pack(side="left")
self.target_entry.bind("<Return>", lambda _event: self._set_target())
ttk.Label(target_frame, text="SLM").grid(row=0, column=2, sticky="w")
ttk.Label(target_input, text="SLM").pack(side="left", padx=(5, 0))
self.target_button = ttk.Button(
target_frame,
text="设置目标流量",
command=self._set_target,
style="Primary.TButton",
width=17,
width=15,
)
self.target_button.grid(row=0, column=3, padx=(18, 0))
self.target_button.grid(row=0, column=2, padx=(12, 0))
self.stop_control_button = ttk.Button(
target_frame,
text="停止流量控制",
command=self._stop_control,
width=15,
)
self.stop_control_button.grid(row=0, column=3, padx=(8, 0))
ttk.Label(target_frame, text="当前目标：").grid(
row=0, column=4, padx=(24, 0)
row=0, column=4, padx=(18, 0)
)
ttk.Label(
target_frame,
@@ -221,6 +437,39 @@ class FlowControlApp:
font=("Microsoft YaHei UI", 11, "bold"),
foreground="#075cad",
).grid(row=0, column=5, sticky="w")
ttk.Label(
target_frame,
text="手动开度",
).grid(row=1, column=0, sticky="w", pady=(9, 0))
manual_opening_input = ttk.Frame(target_frame)
manual_opening_input.grid(
row=1, column=1, sticky="w", padx=(8, 5), pady=(9, 0)
)
self.manual_opening_entry = ttk.Entry(
manual_opening_input,
textvariable=self.manual_opening_var,
width=9,
font=("Microsoft YaHei UI", 12),
)
self.manual_opening_entry.pack(side="left")
self.manual_opening_entry.bind(
"<Return>", lambda _event: self._set_manual_opening()
)
ttk.Label(manual_opening_input, text="%").pack(side="left", padx=(5, 0))
self.manual_opening_button = ttk.Button(
target_frame,
text="设置开度",
command=self._set_manual_opening,
width=15,
)
self.manual_opening_button.grid(
row=1, column=2, padx=(12, 0), pady=(9, 0)
)
ttk.Label(
target_frame,
text="仅在闭环和扫点均停止后生效",
foreground="#5f6368",
).grid(row=1, column=4, columnspan=2, sticky="w", padx=(24, 0), pady=(9, 0))
ttk.Label(
target_frame,
text=(
@@ -228,7 +477,7 @@ class FlowControlApp:
f"{config.TARGET_FLOW_MAX_SLM:g} SLM；首次设置即启动闭环"
),
foreground="#5f6368",
).grid(row=1, column=0, columnspan=6, sticky="w", pady=(7, 0))
).grid(row=2, column=0, columnspan=6, sticky="w", pady=(7, 0))
telemetry = ttk.LabelFrame(
outer,
@@ -354,6 +603,161 @@ class FlowControlApp:
self.detail_var.set(f"正在设置目标流量 {value:g} SLM…")
self.worker.request_target(value)
def _stop_control(self):
if self.current_state != "CONTROLLING":
return
self.detail_var.set("正在停止流量控制并下发 100% 开度…")
self._append_log("请求停止流量控制并下发 100% 开度")
self.stop_control_button.configure(state="disabled")
self.worker.request_stop_control()
def _set_manual_opening(self):
if self.current_state == "CONTROLLING":
messagebox.showwarning(
"请先停止流量控制",
"闭环流量控制正在运行。请先点击“停止流量控制”，再设置手动开度。",
parent=self.root,
)
return
if self.current_state == "OPEN_LOOP":
messagebox.showwarning(
"请先停止开环扫点",
"开环扫点正在运行。请先点击“停止扫点”，再设置手动开度。",
parent=self.root,
)
return
if self.current_state != "MONITORING":
messagebox.showwarning(
"无法设置开度",
"请先连接设备，再设置手动开度。",
parent=self.root,
)
return
try:
opening = validate_opening(self.manual_opening_var.get())
except ValueError as exc:
messagebox.showerror("手动开度无效", str(exc), parent=self.root)
return
self.detail_var.set(f"正在下发手动开度 {opening:g}%…")
self.worker.request_manual_opening(opening)
def _open_open_loop_dialog(self):
if self.current_state == "OPEN_LOOP":
self.detail_var.set("正在停止扫点并生成结果…")
self._append_log("请求提前停止开环扫点")
self.open_loop_button.configure(state="disabled")
self.worker.request_open_loop_stop()
return
if self.current_state not in {"MONITORING", "CONTROLLING"}:
messagebox.showwarning(
"无法开始扫点",
"请先连接设备并确认传感器数据正常。",
parent=self.root,
)
return
OpenLoopDialog(self.root, self._start_open_loop)
def _start_open_loop(self, settings):
if self.current_state not in {"MONITORING", "CONTROLLING"}:
messagebox.showwarning(
"无法开始扫点", "设备当前未处于可扫点状态。", parent=self.root
)
return False
if self.current_state == "CONTROLLING" and not messagebox.askyesno(
"切换到开环扫点",
"开始扫点将停止当前闭环控制，并按扫描序列直接下发阀门开度。是否继续？",
parent=self.root,
):
return False
point_count = len(build_open_loop_points(settings))
self.active_target_var.set("--")
self.detail_var.set(f"正在启动开环扫点，共 {point_count} 点…")
self._append_log(f"请求启动开环扫点，共 {point_count} 个扫描点")
self.worker.request_open_loop_start(settings)
return True
def _fit_feedforward_model(self):
if self.fit_in_progress:
return
initial_directory = application_directory() / "open_loop_data"
if not initial_directory.is_dir():
initial_directory = application_directory()
csv_paths = filedialog.askopenfilenames(
parent=self.root,
title="选择一份或多份开环扫点 CSV",
initialdir=str(initial_directory),
filetypes=(("开环扫点 CSV", "*.csv"), ("所有文件", "*.*")),
)
if not csv_paths:
return
default_name = datetime.now().strftime("valve_model_%Y%m%d_%H%M%S.json")
output_path = filedialog.asksaveasfilename(
parent=self.root,
title="保存拟合后的前馈表",
initialdir=str(Path(csv_paths[0]).resolve().parent),
initialfile=default_name,
defaultextension=".json",
filetypes=(("阀模型 JSON", "*.json"),),
confirmoverwrite=True,
)
if not output_path:
return
self.fit_in_progress = True
self.fit_model_button.configure(text="正在拟合…", state="disabled")
self._append_log(
f"开始合并拟合 {len(csv_paths)} 份开环 CSV；输出：{output_path}"
)
threading.Thread(
target=self._run_feedforward_fit,
args=(tuple(csv_paths), output_path),
name="feedforward-model-fit",
daemon=True,
).start()
def _run_feedforward_fit(self, csv_paths, output_path):
try:
model, steady, table, dropped, stats, area_points = identify(
csv_paths,
tail=DEFAULT_TAIL,
out_path=output_path,
)
image_path = str(Path(output_path).with_suffix(".png"))
plot_warning = None
try:
create_diagnostic_plot(
steady,
table,
dropped,
area_points,
image_path,
)
except Exception as exc:
image_path = None
plot_warning = str(exc)
self.logger.exception("前馈表诊断图生成失败")
self.fit_events.put(
{
"kind": "model_fit_finished",
"csv_count": len(csv_paths),
"output_path": str(Path(output_path).resolve()),
"image_path": image_path,
"point_count": len(model.area_table),
"steady_count": len(steady),
"dropped_count": len(dropped),
"stats": stats,
"plot_warning": plot_warning,
}
)
except Exception as exc:
self.logger.exception("拟合前馈表失败")
self.fit_events.put(
{
"kind": "model_fit_failed",
"message": str(exc),
}
)
def _choose_model(self):
path = filedialog.askopenfilename(
parent=self.root,
@@ -391,6 +795,11 @@ class FlowControlApp:
self._handle_worker_event(event)
except Empty:
pass
try:
while True:
self._handle_fit_event(self.fit_events.get_nowait())
except Empty:
pass
if not self.closing:
self.root.after(80, self._poll_worker_events)
elif self.worker.is_alive():
@@ -412,8 +821,52 @@ class FlowControlApp:
target = event["target"]
self.active_target_var.set(f"{target:g} SLM")
self._append_log(f"目标流量已生效：{target:g} SLM")
elif kind == "control_stopped":
self.active_target_var.set("--")
self.value_vars["opening"].set(
self._format_number(event.get("opening_pct"), digits=2)
)
self.value_vars["motor"].set(
self._format_number(event.get("motor_position"), digits=1)
)
self.value_vars["feedforward"].set("--")
self.value_vars["correction"].set("--")
self.value_vars["mode"].set("未启用")
self._append_log("流量控制已停止；总开度已下发为 100%")
elif kind == "manual_opening_applied":
opening = event["opening_pct"]
self.manual_opening_var.set(f"{opening:g}")
self.value_vars["opening"].set(f"{opening:.2f}")
self.value_vars["motor"].set(f"{event['motor_position']:.1f}")
self.value_vars["feedforward"].set("--")
self.value_vars["correction"].set("--")
self.value_vars["mode"].set("未启用")
self._append_log(f"手动开度已生效：{opening:g}%")
elif kind == "model_active":
self._append_log(f"当前控制模型：{event['display_name']}")
elif kind == "open_loop_started":
self.active_target_var.set("--")
self._append_log(
f"开环扫点已启动，共 {event['total_steps']} 点；CSV："
f"{event['csv_path']}"
)
elif kind == "open_loop_progress":
self.detail_var.set(
f"扫点 {event['current_step']}/{event['total_steps']}："
f"开度 {event['opening_pct']:.2f}% / "
f"行程 {event['motor_position']:.1f}，"
f"本点 {event['step_elapsed_s']:.1f} s"
)
elif kind == "open_loop_finished":
result = (
f"开环扫点{event['reason']}，记录 {event['sample_count']} 条采样。\n"
f"CSV：{event['csv_path']}"
)
if event.get("image_path"):
result += f"\nPNG：{event['image_path']}"
self._append_log(result.replace("\n", "；"))
if event.get("completed") and not self.closing:
messagebox.showinfo("开环扫点完成", result, parent=self.root)
elif kind == "warning":
message = event.get("message", "未知警告")
self.detail_var.set(message)
@@ -429,6 +882,36 @@ class FlowControlApp:
if errors:
self._append_log("退出警告：" + "；".join(errors))
def _handle_fit_event(self, event):
self.fit_in_progress = False
if hasattr(self, "fit_model_button"):
fit_enabled = self.current_state in {"DISCONNECTED", "FAULT", "MONITORING"}
self.fit_model_button.configure(
text="拟合前馈表",
state="normal" if fit_enabled and not self.closing else "disabled",
)
if event["kind"] == "model_fit_failed":
message = event.get("message", "未知错误")
self._append_log("拟合前馈表失败：" + message)
if not self.closing:
messagebox.showerror("拟合前馈表失败", message, parent=self.root)
return
message = (
f"已合并 {event['csv_count']} 份 CSV，读取 {event['steady_count']} 个稳态点，"
f"生成 {event['point_count']} 个 A_eff 标定点。\n"
f"JSON：{event['output_path']}"
)
if event.get("image_path"):
message += f"\n诊断图：{event['image_path']}"
if event.get("dropped_count"):
message += f"\n非单调检查丢弃 {event['dropped_count']} 个点。"
if event.get("plot_warning"):
message += f"\n诊断图生成失败，但 JSON 已保留：{event['plot_warning']}"
self._append_log(message.replace("\n", "；"))
if not self.closing:
messagebox.showinfo("拟合前馈表完成", message, parent=self.root)
def _update_telemetry(self, event):
self.value_vars["flow"].set(self._format_number(event.get("measured_flow_slm")))
self.value_vars["pressure_before"].set(
@@ -453,7 +936,7 @@ class FlowControlApp:
self.value_vars["mode"].set(str(event.get("pid_mode") or "--"))
status = event.get("status", "")
if status and status not in {"OK", "MONITORING"}:
if status and status not in {"OK", "MONITORING", "OPEN_LOOP"}:
self.detail_var.set(f"采样状态：{status}")
@staticmethod
@@ -473,7 +956,17 @@ class FlowControlApp:
self.status_badge.configure(fg=foreground, bg=background)
editable = state in {"DISCONNECTED", "FAULT"}
connected = state in {"MONITORING", "CONTROLLING"}
connected = state in {"MONITORING", "CONTROLLING", "OPEN_LOOP"}
target_button_enabled = state in {"MONITORING", "CONTROLLING"}
target_entry_enabled = state in {"MONITORING", "CONTROLLING"}
stop_control_enabled = state == "CONTROLLING"
manual_button_enabled = state in {"MONITORING", "CONTROLLING", "OPEN_LOOP"}
manual_entry_enabled = state == "MONITORING"
open_loop_enabled = state in {"MONITORING", "CONTROLLING", "OPEN_LOOP"}
fit_enabled = state in {"DISCONNECTED", "FAULT", "MONITORING"}
self.open_loop_button.configure(
text="停止扫点" if state == "OPEN_LOOP" else "开环扫点"
)
self._set_state(self.host_entry, "normal" if editable else "disabled")
self._set_state(self.port_entry, "normal" if editable else "disabled")
self._set_state(self.slave_entry, "normal" if editable else "disabled")
@@ -483,12 +976,39 @@ class FlowControlApp:
self._set_state(
self.model_builtin_button, "normal" if editable else "disabled"
)
self._set_state(
self.open_loop_button, "normal" if open_loop_enabled else "disabled"
)
self._set_state(self.connect_button, "normal" if editable else "disabled")
self._set_state(
self.disconnect_button, "normal" if connected else "disabled"
)
self._set_state(self.target_entry, "normal" if connected else "disabled")
self._set_state(self.target_button, "normal" if connected else "disabled")
self._set_state(
self.target_entry, "normal" if target_entry_enabled else "disabled"
)
self._set_state(
self.target_button, "normal" if target_button_enabled else "disabled"
)
self._set_state(
self.stop_control_button,
"normal" if stop_control_enabled else "disabled",
)
self._set_state(
self.manual_opening_entry,
"normal" if manual_entry_enabled else "disabled",
)
self._set_state(
self.manual_opening_button,
"normal" if manual_button_enabled else "disabled",
)
if hasattr(self, "fit_model_button"):
self.fit_model_button.configure(
state=(
"normal"
if fit_enabled and not self.fit_in_progress and not self.closing
else "disabled"
)
)
if state in {"DISCONNECTED", "FAULT"}:
self.active_target_var.set("--")
@@ -506,6 +1026,13 @@ class FlowControlApp:
def _on_close(self):
if self.closing:
return
if self.fit_in_progress:
messagebox.showinfo(
"正在拟合前馈表",
"请等待拟合完成后再关闭程序，以免输出文件不完整。",
parent=self.root,
)
return
self.closing = True
self._apply_state("STOPPING", "正在安全退出，请稍候…")
self._append_log("正在停止控制、恢复 100% 开度并关闭连接")
@@ -532,6 +1059,19 @@ def smoke_test():
if info.point_count < 2 or info.path.name != BUILTIN_MODEL_NAME:
raise RuntimeError("内置阀模型校验失败")
from PcControl import Easy521ModbusClient # noqa: F401
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
with TemporaryDirectory(prefix="flow_control_smoke_") as temp_directory:
image_path = Path(temp_directory) / "matplotlib_smoke.png"
figure, axis = plt.subplots(figsize=(2, 1))
axis.plot([0, 1], [0, 1])
figure.savefig(image_path, dpi=50)
plt.close(figure)
if not image_path.is_file() or image_path.stat().st_size == 0:
raise RuntimeError("开环曲线图依赖校验失败")
return 0