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

新增气体流量控制 GUI 与 Windows 可执行程序

Browse Source 
新增 Tkinter 单窗口操作界面和后台设备状态机，支持连接与断开 PLC、在线设置目标流量、实时显示传感器数据与下发总开度，并在监测及闭环阶段统一执行安全保护。

默认将 valve_model_004.json 内置到单文件 EXE，同时支持校验并选择外部阀模型；整理标定 JSON/PNG 到 valve_model 目录并同步配置路径。补充运行层离线测试、PyInstaller 构建配置、交付文件和完整的 README 使用及重新打包说明。
This commit is contained in:
louis
2026-09-03 17:35:40 +08:00
parent d7ed4181df
commit ead625bf2d
51 changed files with 44578 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gui_main.py
+568
View File
@@ -0,0 +1,568 @@
"""气体流量控制桌面程序。"""
from __future__ import annotations
import argparse
import ctypes
from datetime import datetime
from queue import Empty
import sys
import tkinter as tk
from tkinter import filedialog, messagebox, ttk
import config
from control_runtime import (
BUILTIN_MODEL_NAME,
ControlWorker,
build_runtime_logger,
load_model,
validate_device_settings,
validate_target,
)
WINDOW_TITLE = "气体流量控制"
class FlowControlApp:
"""单窗口操作界面。"""
STATE_TEXT = {
"DISCONNECTED": "未连接",
"CONNECTING": "正在连接",
"MONITORING": "已连接 / 监测中",
"CONTROLLING": "闭环运行中",
"DISCONNECTING": "正在断开",
"STOPPING": "正在退出",
"FAULT": "故障 / 已断开",
}
STATE_COLORS = {
"DISCONNECTED": ("#5f6368", "#eef0f2"),
"CONNECTING": ("#875a00", "#fff3cd"),
"MONITORING": ("#0b6b3a", "#dff4e8"),
"CONTROLLING": ("#075cad", "#deefff"),
"DISCONNECTING": ("#875a00", "#fff3cd"),
"STOPPING": ("#875a00", "#fff3cd"),
"FAULT": ("#a61b1b", "#fde5e5"),
}
def __init__(self, root):
self.root = root
self.root.title(WINDOW_TITLE)
self.root.geometry("860x720")
self.root.minsize(780, 640)
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
self.logger = build_runtime_logger()
self.worker = ControlWorker(self.logger)
self.current_state = "DISCONNECTED"
self.closing = False
self.selected_model_path = None
self.model_info = load_model()
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.active_target_var = tk.StringVar(value="--")
self.status_var = tk.StringVar(value="未连接")
self.detail_var = tk.StringVar(value="请选择模型并连接设备")
self.value_vars = {
"flow": tk.StringVar(value="--"),
"pressure_before": tk.StringVar(value="--"),
"pressure_after": tk.StringVar(value="--"),
"opening": tk.StringVar(value="--"),
"motor": tk.StringVar(value="--"),
"feedforward": tk.StringVar(value="--"),
"correction": tk.StringVar(value="--"),
"mode": tk.StringVar(value="--"),
}
self._configure_style()
self._build_ui()
self._apply_state("DISCONNECTED", "未连接")
self._append_log(
f"已加载 {self.model_info.display_name}；双击程序不会自动连接设备"
)
self.worker.start()
self.root.after(80, self._poll_worker_events)
def _configure_style(self):
self.root.option_add("*Font", ("Microsoft YaHei UI", 10))
style = ttk.Style(self.root)
try:
style.theme_use("vista")
except tk.TclError:
pass
style.configure("Title.TLabel", font=("Microsoft YaHei UI", 20, "bold"))
style.configure("Section.TLabelframe.Label", font=("Microsoft YaHei UI", 11, "bold"))
style.configure("Primary.TButton", font=("Microsoft YaHei UI", 10, "bold"))
style.configure("Danger.TButton", font=("Microsoft YaHei UI", 10, "bold"))
def _build_ui(self):
outer = ttk.Frame(self.root, padding=(18, 14, 18, 14))
outer.pack(fill="both", expand=True)
ttk.Label(outer, text=WINDOW_TITLE, style="Title.TLabel").pack(anchor="w")
ttk.Label(
outer,
text="Easy521 PLC · Modbus TCP · 10 Hz 闭环控制",
foreground="#5f6368",
).pack(anchor="w", pady=(0, 10))
warning = tk.Label(
outer,
text=(
"安全提示：连接成功后及断开设备前，程序会命令阀门至 100% 开度"
f"（行程 {config.MOTOR_OPEN_POSITION:g}）。"
),
bg="#fff4e5",
fg="#8a4b00",
padx=10,
pady=7,
anchor="w",
)
warning.pack(fill="x", pady=(0, 10))
connection = ttk.LabelFrame(
outer,
text="设备与模型",
padding=10,
style="Section.TLabelframe",
)
connection.pack(fill="x")
connection.columnconfigure(1, weight=1)
connection.columnconfigure(3, weight=1)
ttk.Label(connection, text="PLC 地址").grid(row=0, column=0, sticky="w")
self.host_entry = ttk.Entry(connection, textvariable=self.host_var, width=20)
self.host_entry.grid(row=0, column=1, sticky="ew", padx=(7, 14))
ttk.Label(connection, text="端口").grid(row=0, column=2, sticky="w")
self.port_entry = ttk.Entry(connection, textvariable=self.port_var, width=9)
self.port_entry.grid(row=0, column=3, sticky="w", padx=(7, 14))
ttk.Label(connection, text="站号").grid(row=0, column=4, sticky="w")
self.slave_entry = ttk.Entry(connection, textvariable=self.slave_var, width=7)
self.slave_entry.grid(row=0, column=5, sticky="w", padx=(7, 0))
ttk.Label(connection, text="A_eff 模型").grid(
row=1, column=0, sticky="w", pady=(10, 0)
)
self.model_label = ttk.Label(
connection,
textvariable=self.model_var,
foreground="#174f85",
)
self.model_label.grid(
row=1, column=1, columnspan=3, sticky="w", padx=(7, 10), pady=(10, 0)
)
self.model_browse_button = ttk.Button(
connection, text="选择 JSON…", command=self._choose_model
)
self.model_browse_button.grid(row=1, column=4, pady=(10, 0), padx=(0, 7))
self.model_builtin_button = ttk.Button(
connection, text="使用内置", command=self._use_builtin_model
)
self.model_builtin_button.grid(row=1, column=5, pady=(10, 0))
button_row = ttk.Frame(connection)
button_row.grid(row=2, column=0, columnspan=6, sticky="ew", pady=(12, 0))
self.connect_button = ttk.Button(
button_row,
text="连接设备",
command=self._connect,
style="Primary.TButton",
width=16,
)
self.connect_button.pack(side="left")
self.disconnect_button = ttk.Button(
button_row,
text="断开设备",
command=self._disconnect,
style="Danger.TButton",
width=16,
)
self.disconnect_button.pack(side="left", padx=(10, 0))
target_frame = ttk.LabelFrame(
outer,
text="目标流量",
padding=10,
style="Section.TLabelframe",
)
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")
self.target_entry = ttk.Entry(
target_frame,
textvariable=self.target_var,
width=18,
font=("Microsoft YaHei UI", 12),
)
self.target_entry.grid(row=0, column=1, sticky="w", padx=(8, 5))
self.target_entry.bind("<Return>", lambda _event: self._set_target())
ttk.Label(target_frame, text="SLM").grid(row=0, column=2, sticky="w")
self.target_button = ttk.Button(
target_frame,
text="设置目标流量",
command=self._set_target,
style="Primary.TButton",
width=17,
)
self.target_button.grid(row=0, column=3, padx=(18, 0))
ttk.Label(target_frame, text="当前目标：").grid(
row=0, column=4, padx=(24, 0)
)
ttk.Label(
target_frame,
textvariable=self.active_target_var,
font=("Microsoft YaHei UI", 11, "bold"),
foreground="#075cad",
).grid(row=0, column=5, sticky="w")
ttk.Label(
target_frame,
text=(
f"允许范围 {config.TARGET_FLOW_MIN_SLM:g}～"
f"{config.TARGET_FLOW_MAX_SLM:g} SLM；首次设置即启动闭环"
),
foreground="#5f6368",
).grid(row=1, column=0, columnspan=6, sticky="w", pady=(7, 0))
telemetry = ttk.LabelFrame(
outer,
text="实时监视",
padding=10,
style="Section.TLabelframe",
)
telemetry.pack(fill="x", pady=(10, 0))
for column in range(4):
telemetry.columnconfigure(column, weight=1, uniform="card")
self._add_value_card(telemetry, 0, 0, "当前流量", "flow", "SLM")
self._add_value_card(telemetry, 0, 1, "阀前压力", "pressure_before", "kPa")
self._add_value_card(telemetry, 0, 2, "阀后压力", "pressure_after", "kPa")
self._add_value_card(
telemetry, 0, 3, "下发总开度", "opening", "%", prominent=True
)
self._add_value_card(telemetry, 1, 0, "电机行程", "motor", "")
self._add_value_card(telemetry, 1, 1, "前馈开度", "feedforward", "%")
self._add_value_card(telemetry, 1, 2, "PI 修正", "correction", "%")
self._add_value_card(telemetry, 1, 3, "PID 模式", "mode", "")
status_frame = ttk.LabelFrame(
outer,
text="运行状态",
padding=10,
style="Section.TLabelframe",
)
status_frame.pack(fill="both", expand=True, pady=(10, 0))
status_header = ttk.Frame(status_frame)
status_header.pack(fill="x")
self.status_badge = tk.Label(
status_header,
textvariable=self.status_var,
padx=10,
pady=4,
font=("Microsoft YaHei UI", 10, "bold"),
)
self.status_badge.pack(side="left")
ttk.Label(status_header, textvariable=self.detail_var).pack(
side="left", padx=(10, 0)
)
self.log_text = tk.Text(
status_frame,
height=3,
wrap="word",
state="disabled",
bg="#fafafa",
relief="solid",
borderwidth=1,
font=("Microsoft YaHei UI", 9),
)
self.log_text.pack(fill="both", expand=True, pady=(8, 0))
def _add_value_card(self, parent, row, column, title, key, unit, prominent=False):
frame = tk.Frame(
parent,
bg="#edf5ff" if prominent else "#f6f7f8",
highlightbackground="#b8c6d5" if prominent else "#d9dde1",
highlightthickness=1,
padx=10,
pady=8,
)
frame.grid(
row=row,
column=column,
sticky="nsew",
padx=(0 if column == 0 else 5, 0),
pady=(0 if row == 0 else 7, 0),
)
tk.Label(
frame,
text=title,
bg=frame["bg"],
fg="#4c545c",
font=("Microsoft YaHei UI", 9),
).pack(anchor="w")
value_row = tk.Frame(frame, bg=frame["bg"])
value_row.pack(anchor="w", pady=(3, 0))
tk.Label(
value_row,
textvariable=self.value_vars[key],
bg=frame["bg"],
fg="#075cad" if prominent else "#202124",
font=("Microsoft YaHei UI", 16 if prominent else 14, "bold"),
).pack(side="left")
if unit:
tk.Label(
value_row,
text=f" {unit}",
bg=frame["bg"],
fg="#5f6368",
font=("Microsoft YaHei UI", 9),
).pack(side="left", anchor="s", pady=(0, 2))
def _connect(self):
try:
settings = validate_device_settings(
self.host_var.get(), self.port_var.get(), self.slave_var.get()
)
# 在主线程先校验一次，避免无效模型进入连接流程。
self.model_info = load_model(self.selected_model_path)
except ValueError as exc:
messagebox.showerror("无法连接", str(exc), parent=self.root)
return
self._apply_state("CONNECTING", f"正在连接 {settings.host}:{settings.port}…")
self._append_log(f"请求连接 {settings.host}:{settings.port}，站号 {settings.slave_id}")
self.worker.request_connect(settings, self.selected_model_path)
def _disconnect(self):
self._apply_state("DISCONNECTING", "正在安全断开设备…")
self._append_log("请求停止控制并安全断开设备")
self.worker.request_disconnect()
def _set_target(self):
try:
value = validate_target(self.target_var.get())
except ValueError as exc:
messagebox.showerror("目标流量无效", str(exc), parent=self.root)
return
self.detail_var.set(f"正在设置目标流量 {value:g} SLM…")
self.worker.request_target(value)
def _choose_model(self):
path = filedialog.askopenfilename(
parent=self.root,
title="选择 A_eff 阀模型",
filetypes=(("JSON 阀模型", "*.json"), ("所有文件", "*.*")),
)
if not path:
return
try:
info = load_model(path)
except ValueError as exc:
messagebox.showerror("模型无效", str(exc), parent=self.root)
return
self.selected_model_path = str(info.path)
self.model_info = info
self.model_var.set(info.display_name)
monotonic_text = "单调" if info.is_monotonic else "非单调（按关闭端优先反解）"
self._append_log(f"已选择 {info.display_name}，{monotonic_text}")
def _use_builtin_model(self):
try:
info = load_model()
except ValueError as exc:
messagebox.showerror("内置模型不可用", str(exc), parent=self.root)
return
self.selected_model_path = None
self.model_info = info
self.model_var.set(info.display_name)
self._append_log(f"已切换为 {info.display_name}")
def _poll_worker_events(self):
try:
while True:
event = self.worker.events.get_nowait()
self._handle_worker_event(event)
except Empty:
pass
if not self.closing:
self.root.after(80, self._poll_worker_events)
elif self.worker.is_alive():
self.root.after(80, self._poll_worker_events)
else:
self.root.destroy()
def _handle_worker_event(self, event):
kind = event["kind"]
if kind == "state":
state = event["state"]
message = event.get("message", "")
self._apply_state(state, message)
if state != "DISCONNECTED" or message != "未连接":
self._append_log(message)
elif kind == "telemetry":
self._update_telemetry(event)
elif kind == "target_applied":
target = event["target"]
self.active_target_var.set(f"{target:g} SLM")
self._append_log(f"目标流量已生效：{target:g} SLM")
elif kind == "model_active":
self._append_log(f"当前控制模型：{event['display_name']}")
elif kind == "warning":
message = event.get("message", "未知警告")
self.detail_var.set(message)
self._append_log("警告：" + message)
elif kind == "fault":
message = event.get("message", "未知故障")
self._apply_state("FAULT", message)
self._append_log("故障：" + message)
if not self.closing:
messagebox.showerror("设备或控制故障", message, parent=self.root)
elif kind == "shutdown_complete" and self.closing:
errors = event.get("errors") or []
if errors:
self._append_log("退出警告：" + "；".join(errors))
def _update_telemetry(self, event):
self.value_vars["flow"].set(self._format_number(event.get("measured_flow_slm")))
self.value_vars["pressure_before"].set(
self._format_number(event.get("pressure_before_kpa"))
)
self.value_vars["pressure_after"].set(
self._format_number(event.get("pressure_kpa"))
)
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._format_number(event.get("feedforward_pct"), digits=2)
)
correction = event.get("correction_pct")
self.value_vars["correction"].set(
"--" if correction is None else f"{float(correction):+.2f}"
)
self.value_vars["mode"].set(str(event.get("pid_mode") or "--"))
status = event.get("status", "")
if status and status not in {"OK", "MONITORING"}:
self.detail_var.set(f"采样状态：{status}")
@staticmethod
def _format_number(value, digits=2):
if value is None:
return "--"
return f"{float(value):.{digits}f}"
def _apply_state(self, state, detail):
self.current_state = state
self.status_var.set(self.STATE_TEXT.get(state, state))
self.detail_var.set(detail)
foreground, background = self.STATE_COLORS.get(
state, ("#202124", "#eef0f2")
)
if hasattr(self, "status_badge"):
self.status_badge.configure(fg=foreground, bg=background)
editable = state in {"DISCONNECTED", "FAULT"}
connected = state in {"MONITORING", "CONTROLLING"}
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")
self._set_state(
self.model_browse_button, "normal" if editable else "disabled"
)
self._set_state(
self.model_builtin_button, "normal" if editable 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")
if state in {"DISCONNECTED", "FAULT"}:
self.active_target_var.set("--")
@staticmethod
def _set_state(widget, state):
widget.configure(state=state)
def _append_log(self, message):
timestamp = datetime.now().strftime("%H:%M:%S")
self.log_text.configure(state="normal")
self.log_text.insert("end", f"[{timestamp}] {message}\n")
self.log_text.see("end-1c")
self.log_text.configure(state="disabled")
def _on_close(self):
if self.closing:
return
self.closing = True
self._apply_state("STOPPING", "正在安全退出，请稍候…")
self._append_log("正在停止控制、恢复 100% 开度并关闭连接")
self.worker.request_shutdown()
def enable_dpi_awareness():
"""让 Windows 高 DPI 显示更清晰；失败时安全忽略。"""
if sys.platform != "win32":
return
try:
ctypes.windll.shcore.SetProcessDpiAwareness(1)
except (AttributeError, OSError):
try:
ctypes.windll.user32.SetProcessDPIAware()
except (AttributeError, OSError):
pass
def smoke_test():
"""供打包后自动检查资源和关键依赖，不连接设备。"""
config.validate_config()
info = load_model()
if info.point_count < 2 or info.path.name != BUILTIN_MODEL_NAME:
raise RuntimeError("内置阀模型校验失败")
from PcControl import Easy521ModbusClient # noqa: F401
return 0
def parse_args(argv=None):
parser = argparse.ArgumentParser(add_help=True)
parser.add_argument(
"--smoke-test",
action="store_true",
help="仅检查内置资源和依赖，不打开界面、不连接设备",
)
return parser.parse_args(argv)
def main(argv=None):
args = parse_args(argv)
if args.smoke_test:
return smoke_test()
enable_dpi_awareness()
root = tk.Tk()
try:
FlowControlApp(root)
except Exception as exc:
root.withdraw()
messagebox.showerror("程序启动失败", str(exc), parent=root)
root.destroy()
return 1
root.mainloop()
return 0
if __name__ == "__main__":
raise SystemExit(main())