# gui.py import base64 import matplotlib # from prompt_toolkit.key_binding.bindings.named_commands import self_insert import requests matplotlib.use('TkAgg') import matplotlib.pyplot as plt from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg, NavigationToolbar2Tk import warnings import tkinter as tk from tkinter import ttk # from tkinter import scrolledtext import threading import time import os import pickle import datetime import json import threading import sys import numpy as np # import pandas as pd import serial.tools.list_ports from stable_baselines3 import SAC import torch # from PressureEnv import CustomPressureEnv from PcControl import Easy521ModbusClient, MotorModbusRTUClient # from zzp import SECRET_KEY # from PcControl import PressureModbusRTUClient, MotorModbusRTUClient from controllers import IncrementalPID from styles import apply_app_style from get_V import measure_volume from ind_collector import collect_data_with_prbs import logging sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from core.simulated_device import SimulatedDevice from api import base_url, data_record_url, the_folder warnings.filterwarnings('ignore') logging.getLogger("pymodbus").setLevel(logging.ERROR) # 优化matplotlib设置 matplotlib.rcParams['figure.max_open_warning'] = 20 matplotlib.rcParams['axes.linewidth'] = 0.5 matplotlib.rcParams['lines.linewidth'] = 1.0 plt.rcParams['font.sans-serif'] = [ 'Microsoft YaHei', # Windows 优先 (微软雅黑) 'SimHei', # Windows 备选 (黑体) 'PingFang SC', # macOS 优先 (苹方) 'Heiti TC', # macOS 备选 (黑体) 'sans-serif' # 最终兜底 ] plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题 def get_base_path(): """获取程序运行时的当前真实根目录""" if getattr(sys, 'frozen', False): return os.path.dirname(sys.executable) # exe所在目录 else: return os.path.dirname(os.path.abspath(__file__)) # py脚本所在目录 class ControlGUI: def __init__(self, root): self.root = root self.root.title("ReinLoop-V1.0 - 收敛有界") self.root.geometry("900x700") # 初始化Modbus客户端和NMPC控制器 self.modbus_client = None self.IncrementalPID = IncrementalPID(kp=1.0, ki=0.4, kd=0, dt=0.1, out_min=0, out_max=100) # 定义 (容积, 流量) 组合与模型的映射关系 # self.update_selectors_from_config() # 更新下拉菜单 self.condition_to_model_map = {} # 先初始化为空 # self.load_config_from_json() # 调用外部读取方法 # 控制标志 self.running = False self.control_thread = None # 数据记录 self.pressure_data = [] self.target_data = [] self.valve_data = [] self.time_data = [] self.cycle_count = 0 self.start_time = None # --- 数据记录与状态显示 --- # --- 数据收集与系统辨识相关 --- self.collect_data_var = tk.BooleanVar(value=False) # 数据收集开关 self.episode_data_raw = [] # 存放所有 Episode 的列表 self.current_episode = None # 当前正在记录的 Episode self.steady_count = 0 # 稳态计数器 self.last_target_rl = None # 记录上一个目标值，用于RL模型 self.last_target_record = None # 记录上一个目标值，用于切分 Episode # 绘图相关 self.current_fig = None self.current_ax1 = None self.current_ax2 = None self.current_canvas = None self.x_min_var = None self.x_max_var = None self.is_plotting = False self.setup_gui() self.scan_models_folder() self.confirmed_target_pressure = float(self.target_entry.get()) def safe_log(self, message): if hasattr(self, "log_text"): self.log_message(message) else: print(message) def update_selectors_from_config(self): """根据加载到的配置更新界面下拉框内容""" if not self.condition_to_model_map: return vols = sorted(list(set([k[0] for k in self.condition_to_model_map.keys()])), key=float) flows = sorted(list(set([k[1] for k in self.condition_to_model_map.keys()])), key=float) if hasattr(self, 'volume_selector'): self.volume_selector['values'] = vols self.flow_selector['values'] = flows def _update_pid_ui(self, kp, ki, kd=None): """实时更新界面上的 PID 参数显示""" # 必须先解除禁用状态才能修改文字 current_state = self.Kp_entry['state'] self.Kp_entry.config(state=tk.NORMAL) self.Ki_entry.config(state=tk.NORMAL) self.Kd_entry.config(state=tk.NORMAL) self.Kp_entry.delete(0, tk.END) self.Kp_entry.insert(0, f"{kp:.3f}") self.Ki_entry.delete(0, tk.END) self.Ki_entry.insert(0, f"{ki:.3f}") if kd is not None: self.Kd_entry.delete(0, tk.END) self.Kd_entry.insert(0, f"{kd:.3f}") # 恢复之前的状态 (如果是在RL模式下，它应该变回灰色的 DISABLED) self.Kp_entry.config(state=current_state) self.Ki_entry.config(state=current_state) self.Kd_entry.config(state=current_state) def setup_gui(self): """严格划分双标签页（页1：连接设置，页2：控制设置）""" # ========================================== # 1. 应用统一配色与全局 ttk 样式（定义见 styles.py） # ========================================== colors = apply_app_style(self.root) BG_COLOR = colors["BG_COLOR"] CARD_BG = colors["CARD_BG"] BORDER_COLOR = colors["BORDER_COLOR"] TEXT_MAIN = colors["TEXT_MAIN"] TEXT_MUTED = colors["TEXT_MUTED"] ACCENT_LIGHT = colors["ACCENT_LIGHT"] ACCENT_DARK = colors["ACCENT_DARK"] HOVER_BLUE = colors["HOVER_BLUE"] ACCENT_BLUE = colors["ACCENT_BLUE"] SUCCESS_GREEN = colors["SUCCESS_GREEN"] # ========================================== # 2. 创建顶部导航栏（深色横贯条：第一行系统名，第二行标签页） # 系统名与标签页同处一个深色容器内，浑然一体，无分界线 # ========================================== nav_frame = tk.Frame(self.root, bg=ACCENT_DARK) nav_frame.pack(fill=tk.X, side=tk.TOP) # --- 第一行：系统名 --- title_row = tk.Frame(nav_frame, bg=ACCENT_DARK) title_row.pack(fill=tk.X) title_label = tk.Label( title_row, text="ReinLoop", bg=ACCENT_DARK, fg="white", font=("Microsoft YaHei", 20, "bold") ) title_label.pack(side=tk.LEFT, padx=20, pady=(10, 4)) # --- 第二行：标签页（自定义按钮，仅颜色变化，尺寸恒定）--- tab_row = tk.Frame(nav_frame, bg=ACCENT_DARK) tab_row.pack(fill=tk.X) # ========================================== # 3. 创建核心双标签页容器（隐藏自带标签栏，由上方导航栏切换） # ========================================== self.notebook = ttk.Notebook(self.root) self.notebook.pack(expand=True, fill=tk.BOTH, padx=10, pady=10) # 构建三个独立的标签页 Frame self.tab1 = ttk.Frame(self.notebook, padding="15") self.tab2 = ttk.Frame(self.notebook, padding="15") self.tab3 = ttk.Frame(self.notebook, padding="15") self.notebook.add(self.tab1) self.notebook.add(self.tab2) self.notebook.add(self.tab3) # --- 自定义导航标签按钮 --- # 颜色：未选中=深色主题色(与导航栏融为一体)，选中=浅色主题色，悬停=过渡色 self._nav_tab_buttons = [] nav_tabs = [("连接设置", self.tab1), ("控制设置", self.tab2), ("模型调试", self.tab3)] def _select_nav_tab(index): self.notebook.select(index) for i, btn in enumerate(self._nav_tab_buttons): if i == index: btn.config(bg=ACCENT_LIGHT, fg="white") else: btn.config(bg=ACCENT_DARK, fg="white") # 立即刷新空闲任务队列，强制新页面马上重绘 # （否则单击时事件队列为空，页面重绘会被延迟到下一个事件到来时才显示） self.notebook.update_idletasks() for idx, (label_text, _tab) in enumerate(nav_tabs): btn = tk.Label( tab_row, text=label_text, bg=ACCENT_DARK, fg="white", font=("Microsoft YaHei", 16, "bold"), padx=24, pady=8, cursor="hand2" ) btn.pack(side=tk.LEFT, padx=(20 if idx == 0 else 4, 0), pady=(0, 4)) btn.bind("", lambda e, i=idx: _select_nav_tab(i)) def _on_enter(e, b=btn, i=idx): if self.notebook.index(self.notebook.select()) != i: b.config(bg=HOVER_BLUE) def _on_leave(e, b=btn, i=idx): if self.notebook.index(self.notebook.select()) != i: b.config(bg=ACCENT_DARK) btn.bind("", _on_enter) btn.bind("", _on_leave) self._nav_tab_buttons.append(btn) # 默认选中第一个标签页 _select_nav_tab(0) # ========================================== # 3. 布局【页面 1：连接设置】 # ========================================== # 列配置：标签列固定宽度，控件列自适应 self.tab1.columnconfigure(0, minsize=140) self.tab1.columnconfigure(1, weight=1) _r = 0 # 行计数器 # ---------------- Modbus TCP 区 ---------------- ttk.Label(self.tab1, text="Modbus TCP", style="Section.TLabel").grid( row=_r, column=0, columnspan=2, sticky=tk.W, padx=12, pady=(8, 2)) _r += 1 # PLC 地址 ttk.Label(self.tab1, text="PLC地址:").grid(row=_r, column=0, sticky=tk.W, padx=(12, 2), pady=8) self.tcp_ip_entry = ttk.Entry(self.tab1, width=20) self.tcp_ip_entry.insert(0, "192.168.1.88") self.tcp_ip_entry.grid(row=_r, column=1, sticky=tk.W, padx=(2, 10), pady=8) _r += 1 # 端口 ttk.Label(self.tab1, text="端口:").grid(row=_r, column=0, sticky=tk.W, padx=(12, 2), pady=8) self.tcp_port_entry = ttk.Entry(self.tab1, width=20) self.tcp_port_entry.insert(0, "502") self.tcp_port_entry.grid(row=_r, column=1, sticky=tk.W, padx=(2, 10), pady=8) _r += 1 # 读取压力寄存器地址 ttk.Label(self.tab1, text="读取压力寄存器地址:").grid(row=_r, column=0, sticky=tk.W, padx=(12, 2), pady=8) self.pressure_addr_entry = ttk.Entry(self.tab1, width=20) self.pressure_addr_entry.insert(0, "504") self.pressure_addr_entry.grid(row=_r, column=1, sticky=tk.W, padx=(2, 10), pady=8) _r += 1 # ---------------- Modbus RTU 区 ---------------- ttk.Label(self.tab1, text="Modbus RTU", style="Section.TLabel").grid( row=_r, column=0, columnspan=2, sticky=tk.W, padx=12, pady=(18, 2)) _r += 1 # 端口号（下拉） ttk.Label(self.tab1, text="端口号:").grid(row=_r, column=0, sticky=tk.W, padx=(12, 2), pady=8) self.serial_port_var = tk.StringVar() self.serial_port_cb = ttk.Combobox(self.tab1, textvariable=self.serial_port_var, width=18, state="readonly") self.serial_port_cb.grid(row=_r, column=1, sticky=tk.W, padx=(2, 10), pady=8) _r += 1 # 波特率（下拉） ttk.Label(self.tab1, text="波特率:").grid(row=_r, column=0, sticky=tk.W, padx=(12, 2), pady=8) self.baudrate_var = tk.StringVar(value="115200") self.baudrate_cb = ttk.Combobox(self.tab1, textvariable=self.baudrate_var, width=18, state="readonly", values=["9600", "19200", "38400", "57600", "115200"]) self.baudrate_cb.grid(row=_r, column=1, sticky=tk.W, padx=(2, 10), pady=8) _r += 1 # 站号 / 数据位 / 停止位（同一行） ttk.Label(self.tab1, text="站号:").grid(row=_r, column=0, sticky=tk.W, padx=(12, 2), pady=8) rtu_line = tk.Frame(self.tab1, bg=CARD_BG) rtu_line.grid(row=_r, column=1, sticky=tk.W, padx=(2, 10), pady=8) self.rtu_slave_entry = ttk.Entry(rtu_line, width=6) self.rtu_slave_entry.insert(0, "4") self.rtu_slave_entry.pack(side=tk.LEFT) ttk.Label(rtu_line, text="数据位:").pack(side=tk.LEFT, padx=(20, 5)) self.databits_entry = ttk.Entry(rtu_line, width=6) self.databits_entry.insert(0, "8") self.databits_entry.pack(side=tk.LEFT) ttk.Label(rtu_line, text="停止位:").pack(side=tk.LEFT, padx=(20, 5)) self.stopbits_entry = ttk.Entry(rtu_line, width=6) self.stopbits_entry.insert(0, "1") self.stopbits_entry.pack(side=tk.LEFT) _r += 1 # 校验位（下拉） ttk.Label(self.tab1, text="校验位:").grid(row=_r, column=0, sticky=tk.W, padx=(12, 2), pady=8) self.parity_var = tk.StringVar(value="None") self.parity_cb = ttk.Combobox(self.tab1, textvariable=self.parity_var, width=18, state="readonly", values=["None", "Odd", "Even"]) self.parity_cb.grid(row=_r, column=1, sticky=tk.W, padx=(2, 10), pady=8) _r += 1 # 操作动作按钮组 btn_group = tk.Frame(self.tab1, bg=CARD_BG) btn_group.grid(row=_r, column=0, columnspan=2, sticky=tk.W, padx=10, pady=20) self.refresh_port_btn = ttk.Button(btn_group, text="🔄 刷新", command=self.refresh_serial_ports) self.refresh_port_btn.pack(side=tk.LEFT, padx=(0, 15)) self.connect_btn = ttk.Button(btn_group, text="连接设备", style="Action.TButton", command=self.toggle_connection) self.connect_btn.pack(side=tk.LEFT, padx=5) # 初始化调用一次串口刷新 self.refresh_serial_ports() # ========================================== # 4. 布局【页面 2：控制设置】 # ========================================== self.tab2.columnconfigure(0, weight=1) self.tab2.rowconfigure(3, weight=1) # 允许底部的数据图表与日志终端拉伸 # --- [A. 实时监控大字号仪表看板] --- status_frame = ttk.LabelFrame(self.tab2, text=" 系统状态 ", padding="10") status_frame.grid(row=0, column=0, sticky=(tk.W, tk.E), pady=(0, 10)) status_frame.columnconfigure((0, 1, 2), weight=1, uniform="status_cards") # 当前压力 p_card = tk.Frame(status_frame, bg=CARD_BG, bd=1, relief="solid") p_card.grid(row=0, column=0, padx=6, pady=6, sticky="nsew") tk.Label(p_card, text="当前系统压力", bg=CARD_BG, fg=TEXT_MUTED, font=("Microsoft YaHei", 18)).pack(anchor="w", padx=10, pady=(8, 2)) self.current_pressure_var = tk.StringVar(value="0.0 kPa") tk.Label(p_card, textvariable=self.current_pressure_var, bg=CARD_BG, fg="#059669", font=("Century Gothic", 28, "bold")).pack(anchor="w", padx=10, pady=(0, 8)) # 目标压力 t_card = tk.Frame(status_frame, bg=CARD_BG, bd=1, relief="solid") t_card.grid(row=0, column=1, padx=6, pady=6, sticky="nsew") tk.Label(t_card, text="设定目标压力", bg=CARD_BG, fg=TEXT_MUTED, font=("Microsoft YaHei", 18)).pack(anchor="w", padx=10, pady=(8, 2)) self.target_pressure_var = tk.StringVar(value="0.0 kPa") tk.Label(t_card, textvariable=self.target_pressure_var, bg=CARD_BG, fg=ACCENT_BLUE, font=("Century Gothic", 28, "bold")).pack(anchor="w", padx=10, pady=(0, 8)) # 阀门开度 v_card = tk.Frame(status_frame, bg=CARD_BG, bd=1, relief="solid") v_card.grid(row=0, column=2, padx=6, pady=6, sticky="nsew") tk.Label(v_card, text="控制阀门开度", bg=CARD_BG, fg=TEXT_MUTED, font=("Microsoft YaHei", 18)).pack(anchor="w", padx=10, pady=(8, 2)) self.valve_opening_var = tk.StringVar(value="0.0 %") tk.Label(v_card, textvariable=self.valve_opening_var, bg=CARD_BG, fg="#D97706", font=("Century Gothic", 28, "bold")).pack(anchor="w", padx=10, pady=(0, 8)) # --- [B. 参数运行模态配置区] --- control_frame = ttk.LabelFrame(self.tab2, text=" 控制设置 ", padding="10") control_frame.grid(row=1, column=0, sticky=(tk.W, tk.E), pady=(10, 0)) # 物理工况：环境容积 + 稳态流量 row0 = tk.Frame(control_frame, bg=CARD_BG) row0.pack(fill=tk.X, pady=5) ttk.Label(row0, text="物理工况:").pack(side=tk.LEFT, padx=(0, 15)) ttk.Label(row0, text="容积:").pack(side=tk.LEFT) self.volume_var = tk.StringVar(value="2") self.volume_entry = ttk.Entry(row0, textvariable=self.volume_var, width=6) self.volume_entry.pack(side=tk.LEFT, padx=5) ttk.Label(row0, text="L").pack(side=tk.LEFT, padx=(2, 20)) ttk.Label(row0, text="流量:").pack(side=tk.LEFT) self.flow_var = tk.StringVar(value="100") self.flow_entry = ttk.Entry(row0, textvariable=self.flow_var, width=6) self.flow_entry.pack(side=tk.LEFT, padx=5) ttk.Label(row0, text="L/min").pack(side=tk.LEFT, padx=2) # 目标压力设定 + 控制启停 row1 = tk.Frame(control_frame, bg=CARD_BG) row1.pack(fill=tk.X, pady=5) ttk.Label(row1, text="目标压力:").pack(side=tk.LEFT, padx=(0, 15)) self.target_entry = ttk.Entry(row1, width=10) self.target_entry.insert(0, "80.0") self.target_entry.pack(side=tk.LEFT, padx=5) ttk.Label(row1, text="kPa").pack(side=tk.LEFT, padx=(2, 10)) self.set_target_btn = ttk.Button(row1, text="设置目标", command=self.set_target_pressure) self.set_target_btn.pack(side=tk.LEFT, padx=5) # 控制模式单选组 row2 = tk.Frame(control_frame, bg=CARD_BG) row2.pack(fill=tk.X, pady=5) ttk.Label(row2, text="控制方式:").pack(side=tk.LEFT, padx=(0, 15)) self.control_mode_var = tk.StringVar(value="RL") self.radio_rl = ttk.Radiobutton(row2, text="智能自动", variable=self.control_mode_var, value="RL", command=self._on_mode_change) self.radio_rl.pack(side=tk.LEFT, padx=10) self.radio_pid = ttk.Radiobutton(row2, text="手动PID", variable=self.control_mode_var, value="PID", command=self._on_mode_change) self.radio_pid.pack(side=tk.LEFT, padx=10) self.radio_manual = ttk.Radiobutton(row2, text="设置开度", variable=self.control_mode_var, value="MANUAL", command=self._on_mode_change) self.radio_manual.pack(side=tk.LEFT, padx=10) # 强化学习决策模型加载组 self.rl_frame = tk.Frame(control_frame, bg=CARD_BG) self.rl_frame.pack(fill=tk.X, pady=5) ttk.Label(self.rl_frame, text="决策模型:").pack(side=tk.LEFT, padx=(0, 15)) self.model_combobox = ttk.Combobox(self.rl_frame, width=25, state="readonly") self.model_combobox.pack(side=tk.LEFT, padx=5) self.load_model_btn = ttk.Button(self.rl_frame, text="加载模型", command=self.load_rl_model) self.load_model_btn.pack(side=tk.LEFT, padx=5) self.refresh_models_btn = ttk.Button(self.rl_frame, text="🔄 刷新", width=6, command=self.scan_models_folder) self.refresh_models_btn.pack(side=tk.LEFT, padx=5) # 经典PID调节面板 self.pid_frame = tk.Frame(control_frame, bg=CARD_BG) self.pid_frame.pack(fill=tk.X, pady=5) ttk.Label(self.pid_frame, text="PID 调节:").pack(side=tk.LEFT, padx=(0, 15)) ttk.Label(self.pid_frame, text="Kp:").pack(side=tk.LEFT) self.Kp_entry = ttk.Entry(self.pid_frame, width=6) self.Kp_entry.pack(side=tk.LEFT, padx=5) ttk.Label(self.pid_frame, text="Ki:").pack(side=tk.LEFT) self.Ki_entry = ttk.Entry(self.pid_frame, width=6) self.Ki_entry.pack(side=tk.LEFT, padx=5) ttk.Label(self.pid_frame, text="Kd:").pack(side=tk.LEFT) self.Kd_entry = ttk.Entry(self.pid_frame, width=6) self.Kd_entry.pack(side=tk.LEFT, padx=5) self.update_pid_btn = ttk.Button(self.pid_frame, text="更新PID参数", command=self.update_pid_parameters) self.update_pid_btn.pack(side=tk.LEFT, padx=(10, 0)) # 设置开度 self.manual_frame = tk.Frame(control_frame, bg=CARD_BG) self.manual_frame.pack(fill=tk.X, pady=5) ttk.Label(self.manual_frame, text="设置开度:").pack(side=tk.LEFT, padx=(0, 15)) self.valve_entry = ttk.Entry(self.manual_frame, width=10) self.valve_entry.insert(0, " ") self.valve_entry.pack(side=tk.LEFT, padx=5) ttk.Label(self.manual_frame, text="%").pack(side=tk.LEFT, padx=(2, 10)) self.set_valve_btn = ttk.Button(self.manual_frame, text="设置", command=self.set_valve) self.set_valve_btn.pack(side=tk.LEFT, padx=5) # 控制启停行（最后一行，始终在底部） control_row = tk.Frame(control_frame, bg=CARD_BG) control_row.pack(fill=tk.X, pady=5, side=tk.BOTTOM) self.start_btn = ttk.Button(control_row, text="开始控制", style="Action.TButton", command=self.toggle_control) self.start_btn.pack(side=tk.LEFT, padx=5) self.plot_btn = ttk.Button(control_row, text="绘制曲线", command=self.plot_control_data) self.plot_btn.pack(side=tk.LEFT, padx=5) # self.chk_collect_data = tk.Checkbutton(control_row, text="同步收集数据集", variable=self.collect_data_var, bg=CARD_BG) self.chk_collect_data = tk.Checkbutton( control_row, text="同步收集数据集", variable=self.collect_data_var, bg=CARD_BG, activebackground=CARD_BG, selectcolor="#0354AE", # 勾选时背景色为蓝色 fg="#1F2937", # 文字颜色 activeforeground="#1F2937" ) self.chk_collect_data.pack(side=tk.LEFT, padx=(15, 0)) # --- [D. 预留空间] --- # 日志栏已移至底部导航栏 # ========================================== # 5. 布局【页面 3：模型调试】—— 与页面1一致：grid 排列、无外框 # ========================================== self.tab3.columnconfigure(0, weight=1) # --- [A. 系统辨识] --- identify_frame = ttk.LabelFrame(self.tab3, text=" 系统辨识 ", padding="10") identify_frame.grid(row=0, column=0, sticky=(tk.W, tk.E), pady=(0, 10)) # 压力上限（对应 measure_volume 的 p_max） pmax_row = tk.Frame(identify_frame, bg=CARD_BG) pmax_row.pack(fill=tk.X, pady=5) ttk.Label(pmax_row, text="压力上限:").pack(side=tk.LEFT) self.p_max_var = tk.StringVar(value="200") self.p_max_entry = ttk.Entry(pmax_row, textvariable=self.p_max_var, width=6) self.p_max_entry.pack(side=tk.LEFT, padx=5) ttk.Label(pmax_row, text="kPa").pack(side=tk.LEFT) # 过程升温（对应 measure_volume 的 t_delta，单位 °C） tdelta_row = tk.Frame(identify_frame, bg=CARD_BG) tdelta_row.pack(fill=tk.X, pady=5) ttk.Label(tdelta_row, text="过程升温:").pack(side=tk.LEFT) self.t_delta_var = tk.StringVar(value="30") self.t_delta_entry = ttk.Entry(tdelta_row, textvariable=self.t_delta_var, width=6) self.t_delta_entry.pack(side=tk.LEFT, padx=5) ttk.Label(tdelta_row, text="°C").pack(side=tk.LEFT) # 约束上界 / 下界（对应 measure_volume 的 fit_high / fit_low） constraint_row = tk.Frame(identify_frame, bg=CARD_BG) constraint_row.pack(fill=tk.X, pady=5) ttk.Label(constraint_row, text="约束上界:").pack(side=tk.LEFT) self.fit_high_var = tk.StringVar(value="150") self.fit_high_entry = ttk.Entry(constraint_row, textvariable=self.fit_high_var, width=6) self.fit_high_entry.pack(side=tk.LEFT, padx=5) ttk.Label(constraint_row, text="下界:").pack(side=tk.LEFT, padx=(20, 0)) self.fit_low_var = tk.StringVar(value="50") self.fit_low_entry = ttk.Entry(constraint_row, textvariable=self.fit_low_var, width=6) self.fit_low_entry.pack(side=tk.LEFT, padx=5) # 容积 + 测试按钮 volumn_row = tk.Frame(identify_frame, bg=CARD_BG) volumn_row.pack(fill=tk.X, pady=5) ttk.Label(volumn_row, text="容积:").pack(side=tk.LEFT) self.volume_var = tk.StringVar(value="") self.volume_entry = ttk.Entry(volumn_row, textvariable=self.volume_var, width=6) self.volume_entry.pack(side=tk.LEFT, padx=5) ttk.Label(volumn_row, text="L").pack(side=tk.LEFT, padx=(0, 20)) self.test_btn = ttk.Button(volumn_row, text="测试", command=self.get_V) self.test_btn.pack(side=tk.LEFT, padx=5) # 周期（对应 collect_data_with_prbs 的 t_c） period_row = tk.Frame(identify_frame, bg=CARD_BG) period_row.pack(fill=tk.X, pady=5) ttk.Label(period_row, text="周期:").pack(side=tk.LEFT) self.period_var = tk.StringVar(value="2.5") self.period_entry = ttk.Entry(period_row, textvariable=self.period_var, width=6) self.period_entry.pack(side=tk.LEFT, padx=5) ttk.Label(period_row, text="s").pack(side=tk.LEFT) # 阶数（对应 n_order） order_row = tk.Frame(identify_frame, bg=CARD_BG) order_row.pack(fill=tk.X, pady=5) ttk.Label(order_row, text="阶数:").pack(side=tk.LEFT) self.order_var = tk.StringVar(value="6") self.order_entry = ttk.Entry(order_row, textvariable=self.order_var, width=6) self.order_entry.pack(side=tk.LEFT, padx=5) # 序列（对应 levels）+ 开始辨识按钮 ident_row = tk.Frame(identify_frame, bg=CARD_BG) ident_row.pack(fill=tk.X, pady=5) ttk.Label(ident_row, text="序列:").pack(side=tk.LEFT) self.levels_var = tk.StringVar() self.levels_entry = ttk.Entry(ident_row, textvariable=self.levels_var, width=15) self.levels_entry.pack(side=tk.LEFT, padx=5) self.identify_btn = ttk.Button(ident_row, text="开始辨识", command=self.start_identification) self.identify_btn.pack(side=tk.LEFT, padx=5) # --- [B. 高级设置] --- advanced_frame = ttk.LabelFrame(self.tab3, text=" 高级设置 ", padding="10") advanced_frame.grid(row=1, column=0, sticky=(tk.W, tk.E), pady=(10, 0)) # 死区（控制用 dead_area + collect_data_with_prbs 的 dead_area） dz_row = tk.Frame(advanced_frame, bg=CARD_BG) dz_row.pack(fill=tk.X, pady=5) ttk.Label(dz_row, text="死区:").pack(side=tk.LEFT) self.dz_var = tk.StringVar(value="") self.dz_entry = ttk.Entry(dz_row, textvariable=self.dz_var, width=6) self.dz_entry.pack(side=tk.LEFT, padx=5) # 单步限幅（原 motor_max -> du_max） bound_row = tk.Frame(advanced_frame, bg=CARD_BG) bound_row.pack(fill=tk.X, pady=5) ttk.Label(bound_row, text="单步限幅:").pack(side=tk.LEFT) self.motor_max_var = tk.StringVar(value="") self.motor_max_entry = ttk.Entry(bound_row, textvariable=self.motor_max_var, width=6) self.motor_max_entry.pack(side=tk.LEFT, padx=5) # 总限幅（对应 collect_data_with_prbs 的 xa_full） xa_full_row = tk.Frame(advanced_frame, bg=CARD_BG) xa_full_row.pack(fill=tk.X, pady=5) ttk.Label(xa_full_row, text="总限幅:").pack(side=tk.LEFT) self.xa_full_var = tk.StringVar(value="749") self.xa_full_entry = ttk.Entry(xa_full_row, textvariable=self.xa_full_var, width=6) self.xa_full_entry.pack(side=tk.LEFT, padx=5) # ========================================== # 6. 底部导航栏（包含日志和连接状态） # ========================================== self.bottom_frame = tk.Frame(self.root, bg=BG_COLOR) self.bottom_frame.pack(fill=tk.X, side=tk.BOTTOM, padx=5, pady=1) # 设置两列权重：第一列（日志）占 3，第二列（状态）占 1，即比例 3:1 self.bottom_frame.columnconfigure(0, weight=3) # 日志区域 self.bottom_frame.columnconfigure(1, weight=1, minsize=150) # 状态区域 # 左侧容器：日志 log_container = tk.Frame(self.bottom_frame, bg=BG_COLOR) log_container.grid(row=0, column=0, sticky="nsew", padx=10, pady=5) # self.log_text = scrolledtext.ScrolledText( # log_container, height=2, # bg=BG_COLOR, fg="#059669", # insertbackground="#1F2937", selectbackground="#93C5FD", # highlightthickness=0, # relief="flat", borderwidth=0, font=("Consolas", 15) # ) self.log_text = tk.Text( log_container, height=1, fg=TEXT_MUTED, insertbackground="#1F2937", selectbackground="#93C5FD", highlightthickness=0, relief="flat", borderwidth=0, font=("Consolas", 15) ) self.log_text.pack(fill=tk.BOTH, expand=True) # 右侧容器：连接状态 status_container = tk.Frame(self.bottom_frame, bg=BG_COLOR) status_container.grid(row=0, column=1, sticky="nsew", padx=0, pady=5) self.connection_status_var = tk.StringVar(value="未连接") self.status_lbl = tk.Label( status_container, textvariable=self.connection_status_var, bg=BG_COLOR, fg="#EF4444", font=("Microsoft YaHei", 20, "bold") ) self.status_lbl.pack(expand=True, fill=tk.BOTH) # 联动更新初始的 PID/RL 输入框置灰状态 self._on_mode_change() def refresh_serial_ports(self): """扫描当前电脑可用的所有物理/虚拟串口并更新两个下拉框（压力表和电机）""" ports = [port.device for port in serial.tools.list_ports.comports()] # 更新压力表串口下拉框 # self.pressure_serial_cb['values'] = ports # if ports: # self.pressure_serial_cb.current(0) # 默认选中第一个可用串口 # else: # self.pressure_serial_cb.set("无可用串口") # self.safe_log("警告: 未检测到任何可用串口，请检查压力表线缆连接！") # 更新电机串口下拉框 self.serial_port_cb['values'] = ports if ports: self.serial_port_cb.current(0) # 默认选中第一个可用串口 # self.log_message(f"已扫描到 {len(ports)} 个串口") # 可选：为了避免启动时日志太啰嗦，这行可以注释掉 else: self.serial_port_cb.set("无可用串口") self.safe_log("警告: 未检测到任何可用串口，请检查电机线缆连接！") def scan_models_folder(self): """从云端 model_config 文件夹扫描模型文件（不再扫描本地）""" def fetch_models(): try: payload = {"type": "listModels", "folder": f"{the_folder}/model_config"} resp = requests.post(data_record_url, json=payload, timeout=10) result = resp.json() if result.get("success"): files = result.get("files", []) file_list = result.get("fileList", []) # 建立 文件名 -> fileID 的映射（downloadModel 云函数需要 fileID） self.model_file_map = { item.get("fileName"): item.get("fileID") for item in file_list if item.get("fileName") } def _update_combobox(): if files: self.model_combobox['values'] = files self.model_combobox.current(0) # 默认选中第一个 self.root.after(0, lambda: self.log_message(f"成功刷新")) else: self.model_combobox['values'] = [] self.model_combobox.set("无模型文件") self.root.after(0, _update_combobox) else: err = result.get('errMsg') self.root.after(0, lambda err=err: self.log_message(f"获取模型列表失败: {err}")) except Exception as e: msg = str(e) self.root.after(0, lambda msg=msg: self.log_message(f"扫描模型异常: {msg}")) threading.Thread(target=fetch_models, daemon=True).start() # """扫描 model_config 文件夹下的所有模型文件（支持 .onnx, .pt, .pth）""" # base_path = get_base_path() # models_dir = os.path.join(base_path, "model_config") # if not os.path.exists(models_dir): # os.makedirs(models_dir, exist_ok=True) # self.model_combobox['values'] = [] # self.model_combobox.set("") # return # # 支持的模型文件扩展名 # extensions = ('.enc', '.sys', '.zip', '.mlr') # # extensions = ('.pt', '.pth', '.zip', 'rar', '.onnx') # model_files = [] # for f in os.listdir(models_dir): # if f.lower().endswith(extensions): # model_files.append(f) # 只保存文件名，完整路径在加载时拼接 # model_files.sort() # self.model_combobox['values'] = model_files # if model_files: # self.model_combobox.current(0) # 默认选中第一个 # else: # self.model_combobox.set("无模型文件") def _on_mode_change(self): mode = self.control_mode_var.get() if mode == "PID": # 显示PID栏，隐藏其他 self.rl_frame.pack_forget() self.pid_frame.pack(fill=tk.X, pady=5) self.manual_frame.pack_forget() self.Kp_entry.config(state=tk.NORMAL) self.Ki_entry.config(state=tk.NORMAL) self.Kd_entry.config(state=tk.NORMAL) self.update_pid_btn.config(state=tk.NORMAL) self.volume_entry.config(state=tk.NORMAL) self.flow_entry.config(state=tk.NORMAL) self.model_combobox.config(state=tk.DISABLED) self.load_model_btn.config(state=tk.DISABLED) self.refresh_models_btn.config(state=tk.DISABLED) # self.collect_data_var.set(False) self.chk_collect_data.config(state=tk.NORMAL) elif mode == "RL": # RL # 显示决策模型栏，隐藏其他 self.rl_frame.pack(fill=tk.X, pady=5) self.pid_frame.pack_forget() self.manual_frame.pack_forget() self.Kp_entry.config(state=tk.DISABLED) self.Ki_entry.config(state=tk.DISABLED) self.Kd_entry.config(state=tk.DISABLED) self.update_pid_btn.config(state=tk.DISABLED) self.volume_entry.config(state=tk.NORMAL) self.flow_entry.config(state=tk.NORMAL) self.model_combobox.config(state="readonly") self.load_model_btn.config(state=tk.NORMAL) self.refresh_models_btn.config(state=tk.NORMAL) self.chk_collect_data.config(state=tk.NORMAL) elif mode == "MANUAL": # 手动设置开度 # 显示设置开度栏，隐藏其他 self.rl_frame.pack_forget() self.pid_frame.pack_forget() self.manual_frame.pack(fill=tk.X, pady=5) self.Kp_entry.config(state=tk.DISABLED) self.Ki_entry.config(state=tk.DISABLED) self.Kd_entry.config(state=tk.DISABLED) self.update_pid_btn.config(state=tk.DISABLED) self.volume_entry.config(state=tk.NORMAL) self.flow_entry.config(state=tk.NORMAL) self.model_combobox.config(state=tk.DISABLED) self.load_model_btn.config(state=tk.DISABLED) self.refresh_models_btn.config(state=tk.DISABLED) self.collect_data_var.set(False) self.chk_collect_data.config(state=tk.DISABLED) def load_rl_model(self): """从下拉框选择的文件名加载 RL 模型""" selected = self.model_combobox.get() if not selected or selected == "无模型文件": self.log_message("错误：请先选择一个有效的模型") return def download_and_load(): try: # 1. 取出该模型对应的 fileID（downloadModel 云函数只认 fileID） file_id = getattr(self, "model_file_map", {}).get(selected) if not file_id: self.root.after(0, lambda: self.log_message("获取模型下载链接失败: 缺少 fileID，请先刷新模型列表")) return else: self.root.after(0, lambda: self.log_message(f"正在加载模型: {selected}...")) # print(f"Debug: 选中的模型文件 {selected} 对应的 fileID 是 {file_id}") # 2. 请求云函数获取模型文件的临时下载 URL payload = { "type": "downloadModel", "fileID": file_id } resp = requests.post(data_record_url, json=payload, timeout=15) result = resp.json() if not result.get("success"): err = result.get('errMsg') self.root.after(0, lambda err=err: self.log_message(f"获取模型下载链接失败: {err}")) return url = result['url'] # 3. 下载模型文件（二进制内容） model_resp = requests.get(url, timeout=30) if model_resp.status_code != 200: code = model_resp.status_code self.root.after(0, lambda code=code: self.log_message(f"下载模型文件失败: HTTP {code}")) return model_bytes = model_resp.content # 4. 直接加载到内存（无需解密） import io model_stream = io.BytesIO(model_bytes) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = SAC.load(model_stream, device=device) # 5. 保存模型实例 self.rl_model = model self.root.after(0, lambda: self.log_message(f"成功加载模型: {selected}")) except Exception as e: msg = str(e) self.root.after(0, lambda msg=msg: self.log_message(f"加载模型失败: {msg}")) threading.Thread(target=download_and_load, daemon=True).start() # base_path = get_base_path() # model_path = os.path.join(base_path, "model_config", selected) # if not os.path.exists(model_path): # self.log_message(f"错误：模型文件不存在 -> {model_path}") # return # # if not selected.lower().endswith('.sys'): # # self.log_message(f"错误：仅支持 .sys 格式的文件，当前文件为 {selected}") # # return # try: # # 获取当前界面输入的容积和流量（用于构建临时环境） # # vol_str = self.volume_var.get().strip() # # flow_str = self.flow_var.get().strip() # # if not vol_str or not flow_str: # # self.log_message("错误：请先填写容积(L)和流量(L/min)") # # return # # V = float(vol_str) # # Q_in = float(flow_str) # # temp_env = CustomPressureEnv(Q_in=Q_in, V=V, dt=self.IncrementalPID.dt) # # 加载 SAC 模型 # import io # from cryptography.fernet import Fernet # # 1. 把你刚才生成的密钥硬编码写在这里 # cipher = Fernet(SECRET_KEY) # # 2. 读取硬盘上的加密乱码文件 # with open(model_path, 'rb') as f: # encrypted_data = f.read() # # 3. 在内存中瞬间解密 # decrypted_data = cipher.decrypt(encrypted_data) # # 4. 🌟 核心技巧：将内存中的字节数组伪装成一个“文件对象” # model_stream = io.BytesIO(decrypted_data) # # 5. 直接让 SAC 从内存流中加载模型，不接触硬盘！ # device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # model = SAC.load(model_stream, device=device) # # model = SAC.load(model_path, env=temp_env, device=device) # # 将加载的模型和临时环境保存到实例变量中 # self.rl_model = model # # self.rl_env = temp_env # 保留环境引用，以便后续获取归一化参数 # self.log_message(f"成功加载模型: {selected}") # except Exception as e: # self.log_message(f"加载模型失败: {e}") def log_message(self, message): """添加消息到日志（线程安全优化版）- 一次只显示一条""" def _append_log(): # 清除之前的日志，只显示最新一条 self.log_text.delete('1.0', tk.END) self.log_text.insert(tk.END, f"{time.strftime('%H:%M:%S')} - {message}") # 🚨 绝对不要在这里使用 self.root.update() 🚨 # 将打印任务打包，丢给主线程的事件队列去安全执行，绝对不阻塞当前控制线程 self.root.after(0, _append_log) def start_identification(self): """启动辨识数据采集""" if self.running: self.log_message("错误：请先停止控制再进行辨识") return # 新增：检查是否正在辨识中 if hasattr(self, 'identifying') and self.identifying: self.log_message("辨识正在进行中，请等待完成") return if not self.modbus_client or not self.modbus_client.connect: self.log_message("错误：请先连接压力表") return if not hasattr(self, 'motor') or not self.motor.connect: self.log_message("错误：请先连接电机") return q_input = self.flow_var.get().strip() V_val = float(self.volume_var.get()) if self.volume_var.get() else 0 if not q_input: self.log_message("错误：请先在控制设置中输入流量") return try: q_in_val = float(q_input) except ValueError: self.log_message("错误：流量输入必须是有效数字") return self.identifying = True # 设置辨识中标志 self.log_message("开始辨识数据采集...") def collect_thread(): try: dt = 0.1 # 阶数 n_order（页面3 阶数输入框） try: n_order = int(self.order_var.get()) except (ValueError, AttributeError): self.log_message("警告: 阶数输入无效，使用默认值 6") n_order = 6 # 周期 t_c（页面3 周期输入框，单位 s） try: t_c = float(self.period_var.get()) except (ValueError, AttributeError): self.log_message("警告: 周期输入无效，使用默认值 2.5") t_c = 2.5 # 死区 dead_area（页面3 高级设置-死区） dz_str = self.dz_var.get().strip() try: dead_area = float(dz_str) if dz_str else 240 except ValueError: self.log_message("警告: 死区输入无效，使用默认值 240") dead_area = 240 # 总限幅 xa_full（页面3 高级设置-总限幅） try: xa_full = float(self.xa_full_var.get()) except (ValueError, AttributeError): self.log_message("警告: 总限幅输入无效，使用默认值 749") xa_full = 749 levels_str = self.levels_var.get().strip() try: levels = [int(x.strip()) for x in levels_str.split(',')] if len(levels) < 2: self.log_message("警告: 序列至少需要2个值，使用默认值") levels = [10, 20, 30, 40, 50, 60, 70, 80] except ValueError: self.log_message("警告: 序列输入格式错误，使用默认值") levels = [10, 20, 30, 40, 50, 60, 70, 80] def _on_sample(t, u_cmd, p): self.root.after(0, lambda u=u_cmd, p=p: [ self.valve_opening_var.set(f"{u:.1f} %"), self.current_pressure_var.set(f"{p:.1f} kPa") ]) result = collect_data_with_prbs( self.modbus_client, self.motor, q_in_val=q_in_val, dt=dt, n_order=n_order, t_c=t_c, levels=levels, dead_area=dead_area, xa_full=xa_full, # save_dir=os.path.join(get_base_path(), "ind_data"), V_val=V_val, should_stop=lambda: self.identifying is False, log=self.log_message, on_sample=_on_sample, ) if result['success']: # ========== 直接上传内存中的 CSV 数据 ========== csv_data = result.get('csv_data') filename = result.get('filename') if csv_data and filename: # 编码为 base64 file_base64 = base64.b64encode(csv_data).decode('utf-8') payload = { "type": "uploadDataFile", "fileName": filename, "fileBase64": file_base64, "folder": f"{the_folder}/ind_data" } try: resp = requests.post(data_record_url, json=payload, timeout=30) resp_json = resp.json() if resp_json.get("success"): self.root.after(0, lambda: self.log_message(f"辨识数据上传成功")) else: self.root.after(0, lambda: self.log_message(f"辨识数据上传失败: {resp_json.get('errMsg')}")) except Exception as e: err_msg = f"上传辨识数据异常: {e}" self.root.after(0, lambda msg=err_msg: self.log_message(msg)) self.root.after(0, lambda: self.log_message("辨识数据采集完成")) else: self.root.after(0, lambda: self.log_message("辨识未采集到数据")) except Exception as e: # 🚀 顺手加上打印完整的崩溃调用栈，以后如果再错就能一眼看出是哪行代码的问题 import traceback self.log_message(f"辨识数据采集详细错误: {traceback.format_exc()}") self.log_message(f"辨识数据采集失败: {e}") finally: self.identifying = False # 清除辨识中标志 self.log_message("辨识结束") thread = threading.Thread(target=collect_thread) thread.daemon = True thread.start() def get_V(self): """获取体积""" if self.running: self.log_message("错误：请先停止控制再进行测试") return # 新增：检查是否正在辨识中 if hasattr(self, 'identifying') and self.identifying: self.log_message("测试正在进行中，请等待完成") return if not self.modbus_client or not self.modbus_client.connect: self.log_message("错误：请先连接压力表") return if not hasattr(self, 'motor') or not self.motor.connect: self.log_message("错误：请先连接电机") return q_input = self.flow_var.get().strip() if not q_input: self.log_message("错误：请先在控制设置中输入流量") return try: q_in_val = float(q_input) except ValueError: self.log_message("错误：流量输入必须是有效数字") return # 约束上界 / 下界（页面3 约束上界/下界 -> measure_volume 的 fit_high / fit_low） try: fit_high = float(self.fit_high_var.get()) fit_low = float(self.fit_low_var.get()) except ValueError: self.log_message("错误：约束上界/下界必须是有效数字") return # 压力上限 p_max、过程升温 t_delta（页面3 系统辨识） try: p_max = float(self.p_max_var.get()) except ValueError: self.log_message("错误：压力上限必须是有效数字") return try: t_delta = float(self.t_delta_var.get()) except ValueError: self.log_message("错误：过程升温必须是有效数字") return self.identifying = True # 设置辨识中标志 self.log_message("开始测量容积...") def volume_thread(): try: result = measure_volume( self.modbus_client, self.motor, q_in_slm=q_in_val, dt=self.IncrementalPID.dt, p_max=p_max, fit_low=fit_low, fit_high=fit_high, t_delta=t_delta, should_stop=lambda: self.identifying is False, log=self.log_message, on_sample=lambda t, p: self.root.after( 0, lambda p=p: self.current_pressure_var.set(f"{p:.1f} kPa")), ) if result['success']: timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') vol = result['volume_L'] payload_data = result['payload_data'] json_str = json.dumps(payload_data, indent=2, ensure_ascii=False) json_bytes = json_str.encode('utf-8') file_base64 = base64.b64encode(json_bytes).decode('utf-8') filename = f"volume_test_{vol:.2f}L_{timestamp}.json" # 上传到云存储的 ind_data 文件夹（与辨识数据同一位置） upload_payload = { "type": "uploadDataFile", "fileName": filename, "fileBase64": file_base64, "folder": f"{the_folder}/V_config" } try: resp = requests.post(data_record_url, json=upload_payload, timeout=30) resp_json = resp.json() if resp_json.get("success"): self.root.after(0, lambda: self.log_message(f"体积测量成功")) else: self.root.after(0, lambda: self.log_message(f"体积测量失败")) except Exception as e: err_msg = f"体积测量异常: {e}" self.root.after(0, lambda msg=err_msg: self.log_message(msg)) # 把测得的等效体积自动填回容积输入框 self.root.after(0, lambda: self.volume_var.set(f"{vol:.2f}")) self.root.after(0, lambda: self.log_message(f"测量完成，系统等效体积 V = {vol:.4f} L")) else: self.root.after(0, lambda: self.log_message("测量失败：有效数据点不足，无法计算体积")) except Exception as e: import traceback self.log_message(f"容积测量详细错误: {traceback.format_exc()}") self.log_message(f"容积测量失败: {e}") finally: self.identifying = False # 清除辨识中标志 self.log_message("测量结束") thread = threading.Thread(target=volume_thread) thread.daemon = True thread.start() def toggle_connection(self): """切换Modbus连接状态""" if self.modbus_client and self.modbus_client.connect: self.disconnect_plc() else: self.connect_plc() def connect_plc(self): """连接到PLC及电机""" if os.environ.get("REINLOOP_SIMULATION", "").strip().lower() in { "1", "true", "yes", "on"}: # 模拟路径：不创建真实 TCP/RTU 客户端。 self.simulation_mode = True self.modbus_client = SimulatedDevice() self.motor = self.modbus_client self.modbus_client.connect() self.connection_status_var.set("模拟已连接") self.connect_btn.config(text="断开连接") self.log_message("模拟设备已连接（未访问真实 TCP/串口硬件）") return # ---------------- 读取 Modbus TCP 参数 ---------------- ip_address = self.tcp_ip_entry.get().strip() try: tcp_port = int(self.tcp_port_entry.get()) except ValueError: self.log_message("错误: TCP端口必须是整数") return try: pressure_addr = int(self.pressure_addr_entry.get()) except ValueError: self.log_message("错误: 压力寄存器地址必须是整数") return # ---------------- 读取 Modbus RTU 参数 ---------------- motor_port = self.serial_port_var.get() if not motor_port or motor_port == "无可用串口": self.log_message("错误: 请先在下拉框选择有效的端口号！") return try: baudrate = int(self.baudrate_var.get()) rtu_slave = int(self.rtu_slave_entry.get()) databits = int(self.databits_entry.get()) stopbits = int(self.stopbits_entry.get()) except ValueError: self.log_message("错误: 波特率/站号/数据位/停止位必须是整数") return # 校验位 None/Odd/Even -> pymodbus 的 N/O/E parity = {"None": "N", "Odd": "O", "Even": "E"}.get(self.parity_var.get(), "N") try: # Modbus TCP：读压力 self.modbus_client = Easy521ModbusClient(ip_address, port=tcp_port, current_p_addr=pressure_addr) # Modbus RTU：控电机 self.motor = MotorModbusRTUClient( port=motor_port, slave_id=rtu_slave, baudrate=baudrate, bytesize=databits, parity=parity, stopbits=stopbits ) if self.modbus_client.connect(): self.connection_status_var.set("已连接") self.connect_btn.config(text="断开连接") self.log_message(f"成功连接到PLC: {ip_address}:{tcp_port}") else: self.log_message(f"连接PLC失败: {ip_address}:{tcp_port}") return # PLC连不上直接退出，不连电机了 except Exception as e: self.log_message(f"连接错误: {str(e)}") return # ==================== 优化：去除 exit(1) 防闪退 ==================== if not self.motor.connect(): self.log_message(f"电机串口 ({motor_port}) 连接失败，请检查线缆或占用情况！") self.disconnect_plc() # 回滚状态 return time.sleep(1) # 增加短暂延时，等待驱动器接口就绪 if not self.motor.init(): self.log_message("电机初始化失败！") self.motor.disconnect() self.disconnect_plc() # 回滚状态 return self.log_message(f"电机串口 ({motor_port}) 连接并初始化成功！") def disconnect_plc(self): """断开PLC连接""" if self.modbus_client: self.modbus_client.disconnect() self.modbus_client = None self.connection_status_var.set("未连接") self.connect_btn.config(text="连接设备") self.log_message("已断开连接") # 电机可能尚未创建（如 TCP 阶段就失败），加保护避免 AttributeError if getattr(self, "motor", None): self.motor.disconnect() self.motor = None def set_target_pressure(self): """设置目标压力并同步到PLC""" try: target = float(self.target_entry.get()) if 0 <= target <= 300: self.confirmed_target_pressure = target self.log_message(f"目标压力设置为: {target} kPa") # # 更新本地控制器 # self.IncrementalPID.target_pressure = target # # 如果PLC已连接，将目标压力同步写入D42寄存器 # if self.modbus_client and self.modbus_client.connected: # try: # # 安全获取目标地址 # target_addr = self.safe_int_convert(self.target_addr_entry.get(), 42) # # 写入目标压力到PLC # success = self.modbus_client.write_float(target_addr, float(target)) # if success: # self.log_message(f"目标压力设置为: {target} kPa (已同步到PLC)") # # 更新显示 # self.target_pressure_var.set(f"{target:.1f} kPa") # else: # self.log_message(f"目标压力设置为: {target} kPa (但PLC写入失败)") # except Exception as e: # self.log_message(f"目标压力设置为: {target} kPa (但PLC写入错误: {str(e)})") # else: # self.log_message(f"目标压力设置为: {target} kPa (未连接PLC)") # # 更新显示 # self.target_pressure_var.set(f"{target:.1f} kPa") else: self.log_message("错误: 目标压力必须在0-300 kPa范围内") except ValueError: self.log_message("错误: 请输入有效的数字") def set_valve(self): """设置开度给阀门（仅手动模式）""" try: valve = float(self.valve_entry.get()) if 0 <= valve <= 120: self.confirmed_valve = valve self.log_message(f"阀门开度设置为: {valve}%") else: self.log_message("错误: 目标阀开度超出范围") except ValueError: self.log_message("错误: 请输入有效的数字") def toggle_control(self): """开始/停止控制""" if not self.running: self.start_control() else: self.stop_control() def update_pid_parameters(self): """更新PID参数""" try: self.IncrementalPID.kp = float(self.Kp_entry.get()) self.IncrementalPID.ki = float(self.Ki_entry.get()) self.IncrementalPID.kd = float(self.Kd_entry.get()) self.IncrementalPID._calculate_coefficients() self.log_message( f"PID参数更新为: Kp={self.IncrementalPID.kp}, Ki={self.IncrementalPID.ki}, Kd={self.IncrementalPID.kd}") except ValueError as e: self.log_message(f"PID参数输入错误: {e}") def start_control(self): """开始控制循环""" if not self.modbus_client or not self.modbus_client.connect: self.log_message("错误: 请先连接压力表") # self.log_message("错误: 请先连接PLC") return # ======================================================== # 🚀 新增：强制前置校验 # 如果处于 RL 模式，必须确认模型已成功加载，否则绝对不允许启动 # ======================================================== if self.control_mode_var.get() == "RL": if not hasattr(self, 'rl_model') or self.rl_model is None: self.log_message("❌ 启动失败: 强化学习模型未加载！") self.log_message("请先选择工况并点击【加载模型】按钮，然后再点击开始控制。") return # ======================================================== # ===== 消除魔法数字：根据当前压力预置 PID 初始阀位 ===== try: # pressure_addr = self.safe_int_convert(self.pressure_addr_entry.get(), 18) # p_init = self.modbus_client.read_float(pressure_addr) # if p_init is not None: # 这里的公式假设 300kPa 对应 100% 开度 (线性前馈) # 如果你在 env.calculate_feedforward_valve 里有更精确的公式，请替换这里 # initial_valve = max(0.0, min(100.0, (p_init / 300.0) * 100.0)) position_x = self.motor.read_current_position() initial_valve = self.IncrementalPID.init_v(position_x) self.IncrementalPID.output = initial_valve self.log_message(f"预置初始阀位 {initial_valve:.1f}%") # self.log_message(f"初始化：当前压力 {p_init:.1f}kPa，预置初始阀位 {initial_valve:.1f}%") except Exception as e: self.log_message(f"读取初始开度失败，将使用 80% 启动: {e}") self.IncrementalPID.output = 80.0 # ======================================================== self.cached_mode = self.control_mode_var.get() self.cached_collect_data = self.collect_data_var.get() # 提前把字符串转成浮点数存好 flow_str = self.flow_var.get().strip() if not flow_str: self.log_message("错误：请先在控制设置中输入流量") return try: self.cached_flow = float(flow_str) except ValueError: self.log_message("错误：流量输入必须是有效数字") return vol_str = self.volume_var.get() self.cached_volume = float(vol_str) if vol_str else 0.0 dz_str = self.dz_var.get().strip() self.cached_dz = float(dz_str) if dz_str else None # 如果输入了 dz 值，则使用输入的值，否则使用默认值 240 if self.cached_dz is not None: self.IncrementalPID.dead_area = self.cached_dz motor_max_str = self.motor_max_var.get().strip() self.cached_motor_max = float(motor_max_str) if motor_max_str else None self.running = True self.start_btn.config(text="停止控制") self.cycle_count = 0 self.start_time = time.time() self.pressure_data = [] self.target_data = [] self.valve_data = [] self.time_data = [] # --- 新增：初始化数据收集 --- self.episode_data_raw = [] self.current_episode = None self.last_target_rl = None # 记录上一个目标值，用于RL模型 self.last_target_record = None # 记录上一个目标值，用于切分 Episode # 写入M100为True # try: # control_flag_addr = self.safe_int_convert(self.control_flag_addr_entry.get(), 100) # success = self.modbus_client.write_coil(control_flag_addr, True) # if success: # self.log_message(f"已写入控制标志位 M{control_flag_addr} = True") # else: # self.log_message(f"写入控制标志位 M{control_flag_addr} 失败") # except Exception as e: # self.log_message(f"写入控制标志位错误: {str(e)}") # 在单独线程中运行控制循环 # 🌟 1. 挂一块小黑板（共享元组），初始化为 0 self.latest_display_data = (0.0, 0.0, 0.0) # 🌟 2. 告诉一号员工（主线程）：每 100 毫秒去小黑板看一眼数据 self.root.after(50, self._ui_refresh_timer) self.control_thread = threading.Thread(target=self.control_loop, daemon=True) self.control_thread.start() def _ui_refresh_timer(self): """一号员工（主线程）专属：只负责看黑板、画界面""" if not self.running: return # 如果停止了，就不看了 # 从小黑板上读取最新数据 current_pressure, target_pressure, valve_opening = self.latest_display_data # 刷新界面显示 self.current_pressure_var.set(f"{current_pressure:.1f} kPa") self.target_pressure_var.set(f"{target_pressure:.1f} kPa") self.valve_opening_var.set(f"{valve_opening:.1f} %") # 设个闹钟，100毫秒后再次执行自己 self.root.after(50, self._ui_refresh_timer) def stop_control(self): """停止控制循环""" self.running = False self.start_btn.config(text="开始控制") self.log_message("停止控制") # --- 修复：只要有收集到数据就保存，不受复选框当前状态限制 --- # 把最后一个还没闭合的 episode 加入列表 if self.current_episode and len(self.current_episode['pressures']) > 0: self.episode_data_raw.append(self.current_episode) self.current_episode = None if self.episode_data_raw: self._save_and_upload_data() # 写入M100为False # if self.modbus_client and self.modbus_client.connect: # try: # control_flag_addr = self.safe_int_convert(self.control_flag_addr_entry.get(), 100) # success = self.modbus_client.write_coil(control_flag_addr, False) # if success: # self.log_message(f"已写入控制标志位 M{control_flag_addr} = False") # else: # self.log_message(f"写入控制标志位 M{control_flag_addr} 失败") # except Exception as e: # self.log_message(f"写入控制标志位错误: {str(e)}") def _save_and_upload_data(self): """本地保存数据，并(可选)异步回传到服务器""" try: # 1. 本地落盘 vol = self.cached_flow flow = self.cached_flow # base_dir = get_base_path() # save_dir = os.path.join(base_dir, f'data_record/data_{flow}SLM_{vol}L') # os.makedirs(save_dir, exist_ok=True) timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'episode_raw_data_{timestamp}.pkl' # filepath = os.path.join(save_dir, filename) # with open(filepath, 'wb') as f: # pickle.dump(self.episode_data_raw, f) # self.log_message(f"已成功收集并保存 {len(self.episode_data_raw)} 段数据至 {filepath}") # 上传到微信云存储（新建线程防止阻塞 GUI） # 将内存数据序列化为 bytes，再转 base64 data_bytes = pickle.dumps(self.episode_data_raw) file_base64 = base64.b64encode(data_bytes).decode('utf-8') def upload_to_wechat(): try: payload = { "type": "uploadDataFile", "fileName": filename, "fileBase64": file_base64, "folder": f"{the_folder}/data_record/data_{flow}SLM_{vol}L" # 指定存储在云存储的 data_record 文件夹下 } resp = requests.post(data_record_url, json=payload, timeout=30) result = resp.json() if result.get("success"): self.root.after(0, lambda: self.log_message(f"成功收集数据")) else: self.root.after(0, lambda: self.log_message(f"收集数据失败: {result.get('errMsg')}")) except Exception as e: err_msg = f"收集数据异常: {e}" self.root.after(0, lambda msg=err_msg: self.log_message(msg)) threading.Thread(target=upload_to_wechat, daemon=True).start() except Exception as e: self.log_message(f"保存收集数据时发生错误: {e}") finally: self.episode_data_raw = [] # 清空内存 def reset_controller(self): """重置控制器""" if self.running: self.log_message("请先停止控制再重置控制器") return self.IncrementalPID.reset() self.log_message("控制器已重置") def safe_int_convert(self, value, default=0): """安全地将值转换为整数""" try: return int(value) except (ValueError, TypeError): return default def control_loop(self): """控制主循环""" initial_loop = True while self.running: cycle_start = time.perf_counter() # 记录周期开始时间 try: # 安全地获取地址值 # pressure_addr = self.safe_int_convert(self.pressure_addr_entry.get(), 18) # target_addr = self.safe_int_convert(self.target_addr_entry.get(), 42) # valve_addr = self.safe_int_convert(self.valve_addr_entry.get(), 40) current_time = time.time() elapsed_time = current_time - self.start_time if self.start_time else 0 self.time_data.append(elapsed_time) # 读取当前压力值 current_pressure = self.modbus_client.get_current_p() # print(f"t1-读压力用时：{time.perf_counter()-t1}") # current_pressure = self.modbus_client.read_float(pressure_addr) t2 = time.perf_counter() # 建议在高速循环中把这行打印注释掉，否则日志和控制台会刷屏导致软件卡顿 # self.log_message(f"当前压力：{current_pressure}") if initial_loop: target_pressure = float(self.target_entry.get()) set_valve = float(self.valve_entry.get()) initial_loop = False if current_pressure is not None: # plc_target = self.modbus_client.read_float(target_addr) # target_pressure = plc_target if plc_target is not None else float(self.target_entry.get()) # 使用确认后的目标压力（只有点击"设置目标"按钮才会更新） target_pressure = self.confirmed_target_pressure # ======================================================== current_mode = self.cached_mode # current_mode = self.control_mode_var.get() # print(f"t2={time.perf_counter()-t1}") if current_mode == "PID": # 1. 纯 PID 控制模式 self.IncrementalPID.update_pressure_values(current_pressure, target_pressure) valve_opening = self.IncrementalPID.update() xa = 749 * (100 - valve_opening) / 100 success = self.motor.set_position(xa) # success = self.motor.set_position(valve_opening) elif current_mode == "RL": if hasattr(self, 'rl_model') and self.rl_model is not None: t3 = time.perf_counter() volume_val = self.cached_volume flow_rate = self.cached_flow # # --- A. 获取容积 (框里只有数字，直接转 float) --- # volume_str = self.volume_var.get() # volume_val = float(volume_str) if volume_str else 0.0 # # --- B. 获取流量 (框里只有数字，直接转 float) --- # flow_str = self.flow_var.get() # flow_rate = float(flow_str) if flow_str else 0.0 # dz_str = self.dz_var.get().strip() # if dz_str: # self.IncrementalPID.dead_area = float(dz_str) # print(f"t3={time.perf_counter() - t1}") # --- C. 调用 RL 模型预测参数增量 --- if self.last_target_rl is None: self.last_target_rl = target_pressure position_x = self.motor.read_current_position() self.IncrementalPID.output = self.IncrementalPID.init_v(position_x) # self.IncrementalPID.output = self.rl_env.calculate_feedforward_valve(current_pressure) obs = np.array([flow_rate / 100, current_pressure / 100, (target_pressure - current_pressure) / 100], dtype=np.float32) self.log_message(f"obs:{obs}") action, _ = self.rl_model.predict(obs, deterministic=True) # --- D. 更新 PID 参数 --- action_space = self.rl_model.action_space self.Kp_0 = action_space.high[0] self.Ki_0 = action_space.high[1] self.IncrementalPID.kp = self.Kp_0 + action[0] self.IncrementalPID.ki = self.Ki_0 + action[1] # self.IncrementalPID.kp = self.rl_env.Kp_0 + action[0] # self.IncrementalPID.ki = self.rl_env.Ki_0 + action[1] self.log_message( f"Kp={self.IncrementalPID.kp:.4f}, Ki={self.IncrementalPID.ki:.4f}, Kd={self.IncrementalPID.kd:.4f}") self.IncrementalPID._calculate_coefficients() self.root.after(0, self._update_pid_ui, self.IncrementalPID.kp, self.IncrementalPID.ki, self.IncrementalPID.kd) # error = -(target_pressure - current_pressure) # dkp, dki = self.rl_controller.predict(current_pressure, error) # new_kp = max(0.0, min(10.0, 1.0 + dkp)) # new_ki = max(0.0, min(20.0, 0.4 + dki)) # self.IncrementalPID.kp = new_kp # self.IncrementalPID.ki = new_ki # self.IncrementalPID._calculate_coefficients() # self.root.after(0, self._update_pid_ui, new_kp, new_ki) elif self.last_target_rl != target_pressure: # 更新 last_target，确保只在目标压力真正变化时调用一次 RL 模型 self.last_target_rl = target_pressure obs = np.array([flow_rate/100, current_pressure/100, (target_pressure-current_pressure)/100], dtype=np.float32) self.log_message(f"obs: {obs}") action, _ = self.rl_model.predict(obs, deterministic=True) t4 = time.perf_counter() # --- D. 更新 PID 参数 --- # Kp_0 = action_space.high[0] # Ki_0 = action_space.high[1] self.IncrementalPID.kp = self.Kp_0 + action[0] self.IncrementalPID.ki = self.Ki_0 + action[1] # self.IncrementalPID.kp = self.rl_env.Kp_0 + action[0] # self.IncrementalPID.ki = self.rl_env.Ki_0 + action[1] self.log_message(f"Kp={self.IncrementalPID.kp:.4f}, Ki={self.IncrementalPID.ki:.4f}, Kd={self.IncrementalPID.kd:.4f}") self.IncrementalPID._calculate_coefficients() self.root.after(0, self._update_pid_ui, self.IncrementalPID.kp, self.IncrementalPID.ki, self.IncrementalPID.kd) # print(f"t4={time.perf_counter() - t1:.3f}") # 降低 UI 刷新频率：每 10 个控制周期 (0.05秒) 更新一次界面，防止 Tkinter 卡死 # if self.cycle_count % 10 == 0: # self.root.after(0, self._update_pid_ui, new_kp, new_ki) t5 = time.perf_counter() # --- E. 算新开度 --- # 如果输入了电机限幅值，则更新 IncrementalPID.motor_max # motor_max_str = self.motor_max_var.get().strip() # if motor_max_str: # self.IncrementalPID.du_max = float(motor_max_str) * self.IncrementalPID.dt if self.cached_motor_max is not None: self.IncrementalPID.du_max = self.cached_motor_max * self.IncrementalPID.dt else: self.IncrementalPID.get_du_max(target_pressure) self.IncrementalPID.update_pressure_values(current_pressure, target_pressure) valve_opening = self.IncrementalPID.update() xa = self.IncrementalPID.dead_area + (100 - valve_opening) * (750 - self.IncrementalPID.dead_area) / 100 # va = (750 - xa) / 750 * 100 # self.log_message(f"xa: {xa:.4f}") success = self.motor.set_position(xa) # print(f"t5-写开度用时：{time.perf_counter() - t1:.3f}") else: # 极端异常兜底：按理说有前置拦截不会走到这里 self.log_message("⚠️ 致命错误：控制线程中丢失模型实例！正在紧急停机。") self.root.after(0, self.stop_control) # valve_opening = 0.0 # 输出安全阀位 elif current_mode == "MANUAL": # valve_opening = float(self.valve_entry.get()) xa = self.IncrementalPID.dead_area + (100 - set_valve) * (750 - self.IncrementalPID.dead_area) / 100 success = self.motor.set_position(xa) else: # valve_opening = 0.0 self.log_message("错误！") # ======================================================== # ... [此处是原有的获取 valve_opening 并写入 PLC 的代码] ... # 写入阀门开度到PLC # success = self.modbus_client.write_float(valve_addr, float(valve_opening)) # success = self.motor.set_position(valve_opening) # ======================================================== # 新增：训练数据收集逻辑 # ======================================================== t6 = time.perf_counter() # if self.collect_data_var.get(): # vol_str = self.volume_var.get() # flow_str = self.flow_var.get() if self.cached_collect_data: Q_in = self.cached_flow V = self.cached_volume # Q_in = float(flow_str) if flow_str else 0.0 # 根据你的设定转换为标准单位 # # Q_in = float(flow_str) * 1000 if flow_str else 0.0 # 根据你的设定转换为标准单位 # V = float(vol_str) if vol_str else 0.0 # 检查是否需要开启新的 Episode（目标改变或刚启动） if self.current_episode is None or target_pressure != self.last_target_record: if self.current_episode is not None: self.episode_data_raw.append(self.current_episode) self.log_message( f"Episode 结束，已记录 {len(self.current_episode['pressures'])} 个点") self.current_episode = { 'pid': [float(self.IncrementalPID.kp), float(self.IncrementalPID.ki), float(self.IncrementalPID.kd)], 'target_pressure': target_pressure, 'Q_in': Q_in, 'V': V, 'steps': [], 'pressures': [], 'errors': [], 'valves': [] } self.last_target_record = target_pressure self.steady_count = 0 # 记录当前步的数据 error = -(target_pressure - current_pressure) self.current_episode['steps'].append(self.cycle_count) self.current_episode['pressures'].append(current_pressure) self.current_episode['errors'].append(error) self.current_episode['valves'].append(float(valve_opening)) # print(f"t6-记录数据用时：{time.perf_counter() - t1:.3f}") # 可选：判断稳态提前结束 Episode（类似 DataCollector.py 的逻辑） # if 0 <= target_pressure - current_pressure <= 1: # self.steady_count += 1 # else: # self.steady_count = 0 # if self.steady_count > 40: # 假设 40 个 step 为稳态 # ... # ======================================================== # 更新UI并记录数据 self.pressure_data.append(current_pressure) self.target_data.append(target_pressure) self.valve_data.append(valve_opening) # 🌟 2. 把算出来的最新数据写到小黑板上，然后就可以拍拍屁股走人了 self.latest_display_data = (current_pressure, target_pressure, valve_opening) # self.root.after(0, self.update_display, current_pressure, target_pressure, valve_opening) self.cycle_count += 1 # print(f"t6-记录数据用时：{time.perf_counter() - t1:.3f}") else: self.log_message("读取当前压力失败，检查地址和连接") time.sleep(self.IncrementalPID.dt) # 读取失败时短暂等待 except Exception as e: self.log_message(f"控制循环错误: {str(e)}") import traceback self.log_message(f"详细错误: {traceback.format_exc()}") time.sleep(self.IncrementalPID.dt) # 控制周期时间补偿 (保持控制频率恒定) # print(f"cycle time：{time.perf_counter() - cycle_start}") elapsed_time = time.perf_counter() - cycle_start sleep_time = max(0.001, self.IncrementalPID.dt - elapsed_time) # 确保最小等待1ms time.sleep(sleep_time) total_time = time.perf_counter() - cycle_start print(f"total time：{total_time}") if total_time > 0.11: print(f"=================================超时！本循环用时{total_time}") # self.log_message(f"超时！本循环用时{total_time}") print("\n") def update_display(self, current_pressure, target_pressure, valve_opening): """更新显示并记录数据 (带滚动窗口限制)""" self.current_pressure_var.set(f"{current_pressure:.1f} kPa") self.target_pressure_var.set(f"{target_pressure:.1f} kPa") self.valve_opening_var.set(f"{valve_opening:.1f} %") # 1. 记录数据用于绘图 current_time = time.time() if self.start_time is not None: elapsed_time = current_time - self.start_time else: self.start_time = current_time elapsed_time = 0 self.time_data.append(elapsed_time) self.pressure_data.append(current_pressure) self.target_data.append(target_pressure) self.valve_data.append(valve_opening) # ======================================================== # 2. 新增：限制绘图数据的最大长度（只保留最近10分钟） # 控制周期 0.05s，10分钟 = 600秒 = 12000个控制周期 # ======================================================== # max_points = 12000 # if len(self.time_data) > max_points: # print(1111) # # 列表切片，丢弃最前面的老点，只保留最后 12000 个新点 # self.time_data = self.time_data[-max_points:] # self.pressure_data = self.pressure_data[-max_points:] # self.target_data = self.target_data[-max_points:] # self.valve_data = self.valve_data[-max_points:] def plot_control_data(self): """绘制控制数据曲线 - 修复白屏问题""" if not self.pressure_data: self.log_message("没有可绘制的数据") return # 防止重复点击 if self.is_plotting: return self.is_plotting = True self.plot_btn.config(state=tk.DISABLED) self.log_message("正在生成图表...") # 在新线程中创建绘图窗口 plot_thread = threading.Thread(target=self._create_plot_window, daemon=True) plot_thread.start() def _create_plot_window(self): """在新线程中创建绘图窗口""" try: # 确保matplotlib使用正确的设置 matplotlib.use('TkAgg') # plt.rcParams['font.sans-serif'] = ['SimHei'] # plt.rcParams['axes.unicode_minus'] = False # 在主线程中创建窗口 self.root.after(0, self._safe_create_plot_window) except Exception as e: self.root.after(0, self._plot_error, str(e)) def _safe_create_plot_window(self): """安全创建绘图窗口（在主线程中执行）""" try: # 创建新的Toplevel窗口 plot_window = tk.Toplevel(self.root) plot_window.title("控制数据曲线图") plot_window.geometry("1100x800") # 添加加载提示 loading_label = ttk.Label(plot_window, text="正在加载图表...", font=("Arial", 12)) loading_label.pack(pady=20) plot_window.update() # 创建主框架 main_frame = ttk.Frame(plot_window) main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # 创建控制面板 control_frame = ttk.Frame(main_frame) control_frame.pack(fill=tk.X, pady=(0, 10)) # X轴范围控制 ttk.Label(control_frame, text="时间轴范围 (秒):").pack(side=tk.LEFT, padx=(0, 5)) self.x_min_var = tk.StringVar(value="0") x_min_entry = ttk.Entry(control_frame, textvariable=self.x_min_var, width=10) x_min_entry.pack(side=tk.LEFT, padx=5) ttk.Label(control_frame, text="到").pack(side=tk.LEFT, padx=5) if self.time_data: self.x_max_var = tk.StringVar(value=f"{max(self.time_data):.1f}") else: self.x_max_var = tk.StringVar(value="10") x_max_entry = ttk.Entry(control_frame, textvariable=self.x_max_var, width=10) x_max_entry.pack(side=tk.LEFT, padx=5) ttk.Button(control_frame, text="应用", command=lambda: self._apply_x_limits()).pack(side=tk.LEFT, padx=10) ttk.Button(control_frame, text="重置", command=lambda: self._reset_view()).pack(side=tk.LEFT, padx=5) ttk.Button(control_frame, text="全部", command=lambda: self._show_all_data()).pack(side=tk.LEFT, padx=5) ttk.Button(control_frame, text="最后30秒", command=lambda: self._zoom_last_n_seconds(30)).pack(side=tk.LEFT, padx=5) # 创建图形 fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 7), dpi=100) # 确保有足够的数据 if not self.time_data or len(self.time_data) < 2: loading_label.config(text="数据不足，无法绘制图表") self._reenable_plot_button() return # 压力曲线 ax1.plot(self.time_data, self.pressure_data, 'b-o', linewidth=1.5, markersize=3, alpha=0.8, label='实际压力') ax1.plot(self.time_data, self.target_data, 'r--', linewidth=1.5, alpha=0.8, label='目标压力') ax1.set_ylabel('压力 (kPa)', fontsize=12) ax1.set_title('压力控制性能', fontsize=14, fontweight='bold') ax1.legend(loc='upper right', fontsize=10) ax1.grid(True, alpha=0.3) # 阀门开度曲线 ax2.plot(self.time_data, self.valve_data, 'm-o', linewidth=1.5, markersize=3, alpha=0.8, label='实际阀门指令') ax2.set_xlabel('时间 (秒)', fontsize=12) ax2.set_ylabel('阀门开度 (%)', fontsize=12) ax2.legend(loc='upper right', fontsize=10) ax2.set_ylim([0, 105]) ax2.grid(True, alpha=0.3) # 共享X轴 ax2.sharex(ax1) plt.tight_layout() # 移除加载提示 loading_label.destroy() # 创建画布 canvas_frame = ttk.Frame(main_frame) canvas_frame.pack(fill=tk.BOTH, expand=True) canvas = FigureCanvasTkAgg(fig, master=canvas_frame) canvas.draw() # 添加导航工具栏 toolbar_frame = ttk.Frame(canvas_frame) toolbar_frame.pack(fill=tk.X, pady=(0, 5)) toolbar = NavigationToolbar2Tk(canvas, toolbar_frame) toolbar.update() # 将画布放置到窗口中 canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) # 添加提示标签 hint_label = ttk.Label(canvas_frame, text="提示: 使用工具栏缩放/平移 | 拖动矩形区域可局部放大", font=("Arial", 9), foreground="gray") hint_label.pack(side=tk.BOTTOM, pady=(5, 0)) # 存储图表对象 self.current_fig = fig self.current_ax1 = ax1 self.current_ax2 = ax2 self.current_canvas = canvas # 配置窗口关闭事件 def on_closing(): try: plt.close(fig) plot_window.destroy() self.current_fig = None self.current_ax1 = None self.current_ax2 = None self.current_canvas = None except: pass finally: self.is_plotting = False self._reenable_plot_button() plot_window.protocol("WM_DELETE_WINDOW", on_closing) # 确保窗口正确显示 plot_window.update() plot_window.deiconify() self.log_message("图表已生成") except Exception as e: self.log_message(f"创建图表时出错: {str(e)}") import traceback traceback.print_exc() finally: self.is_plotting = False self._reenable_plot_button() def _apply_x_limits(self): """应用X轴范围限制""" if not self.current_canvas or not self.current_ax1: return try: x_min = float(self.x_min_var.get()) x_max = float(self.x_max_var.get()) if x_min >= x_max: return self.current_ax1.set_xlim([x_min, x_max]) self.current_ax2.set_xlim([x_min, x_max]) self.current_canvas.draw() except ValueError: pass def _reset_view(self): """重置视图""" if not self.current_canvas or not self.current_ax1 or not self.time_data: return x_min = min(self.time_data) x_max = max(self.time_data) x_range = x_max - x_min margin = x_range * 0.05 if x_range > 0 else 0.1 self.current_ax1.set_xlim([x_min - margin, x_max + margin]) self.current_ax2.set_xlim([x_min - margin, x_max + margin]) self.x_min_var.set(f"{x_min - margin:.1f}") self.x_max_var.set(f"{x_max + margin:.1f}") self.current_canvas.draw() def _show_all_data(self): """显示所有数据""" if not self.current_canvas or not self.current_ax1 or not self.time_data: return x_min = min(self.time_data) x_max = max(self.time_data) self.current_ax1.set_xlim([x_min, x_max]) self.current_ax2.set_xlim([x_min, x_max]) self.x_min_var.set(f"{x_min:.1f}") self.x_max_var.set(f"{x_max:.1f}") self.current_canvas.draw() def _zoom_last_n_seconds(self, n_seconds): """缩放到最后N秒的数据""" if not self.current_canvas or not self.current_ax1 or not self.time_data: return x_max = max(self.time_data) x_min = max(0, x_max - n_seconds) self.current_ax1.set_xlim([x_min, x_max]) self.current_ax2.set_xlim([x_min, x_max]) self.x_min_var.set(f"{x_min:.1f}") self.x_max_var.set(f"{x_max:.1f}") self.current_canvas.draw() def _plot_error(self, error_msg): """处理绘图错误""" self.log_message(f"绘图错误: {error_msg}") self._reenable_plot_button() def _reenable_plot_button(self): """重新启用绘制按钮""" if self.plot_btn and self.plot_btn.winfo_exists(): self.plot_btn.config(state=tk.NORMAL)