2038 lines
96 KiB
Python
2038 lines
96 KiB
Python
# 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("<Button-1>", 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("<Enter>", _on_enter)
|
||
btn.bind("<Leave>", _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)
|
||
|