215 lines
7.1 KiB
Python
215 lines
7.1 KiB
Python
import numpy as np
|
||
import time
|
||
|
||
|
||
def measure_volume(conn_mgr,
|
||
q_in_slm=50.0, dt=0.1,
|
||
xa=1000, p_max=200.0,
|
||
fit_low=50.0, fit_high=150.0,
|
||
T_delta=30.0,
|
||
should_stop=None, log=print, on_sample=None):
|
||
"""充气升压测试，辨识系统等效体积 V（可直接被 GUI 导入调用）。
|
||
|
||
连接由调用方负责：传入已连接的 ConnectionManager。
|
||
本函数不创建客户端、不调用 exit()、不画图、不阻塞，只跑测试并返回结果。
|
||
|
||
参数:
|
||
conn_mgr : 已连接的 ConnectionManager（需有 read_pressure() / set_motor_position()）
|
||
q_in_slm : 进气流量设定 (SLM)
|
||
dt : 控制/采样周期 (秒)
|
||
xa : 阀门全开对应的电机位置指令
|
||
p_max : 升压上限，超过即停止并关阀 (kPa)
|
||
fit_low/high : 用于线性拟合的压力区间 (kPa)
|
||
t_std : 流量计标况温度 (K)
|
||
t_tank : 充气时估计气体温度 (K)
|
||
should_stop : 可选回调，返回 True 时提前中止（供 GUI 停止按钮用）
|
||
log : 日志回调，默认 print（GUI 可传入 self.log_message）
|
||
on_sample : 可选回调 on_sample(t, pressure)，每个采样点调用一次（供 GUI 刷新界面）
|
||
|
||
返回:
|
||
dict: {
|
||
'record_time': [...], 'p_actual': [...],
|
||
'slope': float | None, 'intercept': float | None,
|
||
'c1': float | None, 'volume_L': float | None,
|
||
'valid_points': int, 'payload_data': dict | None,
|
||
'success': bool
|
||
}
|
||
"""
|
||
p_actual = []
|
||
record_time = []
|
||
|
||
conn_mgr.set_motor_position(xa)
|
||
# time.sleep(5) # 等待压力稳定
|
||
|
||
begin_time = time.perf_counter()
|
||
current_pressure = conn_mgr.read_pressure()
|
||
while True:
|
||
if should_stop is not None and should_stop():
|
||
log("测试被手动中止")
|
||
conn_mgr.set_motor_position(0)
|
||
break
|
||
|
||
start_time = time.perf_counter()
|
||
if current_pressure is not None and current_pressure > p_max:
|
||
conn_mgr.set_motor_position(0)
|
||
break
|
||
|
||
current_pressure = conn_mgr.read_pressure()
|
||
t = time.perf_counter() - begin_time
|
||
if current_pressure is not None:
|
||
p_actual.append(current_pressure)
|
||
record_time.append(t)
|
||
print(f"time:{t:.2f}, current_pressure:{current_pressure}")
|
||
if on_sample is not None:
|
||
on_sample(t, current_pressure)
|
||
|
||
i += 1
|
||
|
||
cycle_time = time.perf_counter() - start_time
|
||
time.sleep(max(dt - cycle_time, 0.001))
|
||
|
||
# ==========================================
|
||
# 自动计算 升压速率(dP/dt) 与 c1、等效体积 V
|
||
# ==========================================
|
||
valid_times = []
|
||
valid_pressures = []
|
||
for tt, p in zip(record_time, p_actual):
|
||
if fit_low <= p <= fit_high:
|
||
valid_times.append(tt)
|
||
valid_pressures.append(p)
|
||
|
||
result = {
|
||
'record_time': record_time,
|
||
'p_actual': p_actual,
|
||
'slope': None,
|
||
'intercept': None,
|
||
'c1': None,
|
||
'volume_L': None,
|
||
'valid_points': len(valid_times),
|
||
'payload_data': None,
|
||
'success': False,
|
||
}
|
||
|
||
log("=" * 40)
|
||
log("物理参数辨识结果")
|
||
log("=" * 40)
|
||
|
||
t_std = 293.15
|
||
t_tank = t_std + T_delta
|
||
|
||
if len(valid_times) > 1:
|
||
slope, intercept = np.polyfit(valid_times, valid_pressures, 1)
|
||
c1 = slope / q_in_slm
|
||
P_atm = 101.325
|
||
volume_L = P_atm / (60 * c1) * (t_tank / t_std)
|
||
|
||
# 打包json上传到云端
|
||
payload_data = {
|
||
"slope": slope,
|
||
"intercept": intercept,
|
||
"c1": c1,
|
||
"volume_L": volume_L,
|
||
"valid_points": len(valid_times),
|
||
"q_in_slm": q_in_slm,
|
||
}
|
||
|
||
result.update({
|
||
'slope': float(slope),
|
||
'intercept': float(intercept),
|
||
'c1': float(c1),
|
||
'volume_L': float(volume_L),
|
||
'payload_data': payload_data,
|
||
'success': True,
|
||
})
|
||
|
||
print(f"有效数据点数量: {len(valid_times)}")
|
||
print(f"实测升压速率 (dP/dt) : {slope:.4f} kPa/s")
|
||
print(f"进气流量设定 (q_in) : {q_in_slm} SLM")
|
||
print(f"最终进气增益 (c1) : {c1:.4f}")
|
||
print(f"系统真实等效体积 (V) : {volume_L:.4f} L")
|
||
|
||
|
||
|
||
else:
|
||
log(f"警告：{fit_low:.0f}~{fit_high:.0f}kPa 区间内的数据点太少，无法计算斜率！")
|
||
log("=" * 40)
|
||
|
||
return result
|
||
|
||
|
||
def main():
|
||
"""独立运行入口：自建连接、跑测试、画图验证。"""
|
||
import matplotlib.pyplot as plt
|
||
from PcControl import Easy521ModbusClient, MotorModbusRTUClient
|
||
import os
|
||
from core.simulated_device import SimulatedDevice
|
||
|
||
q_in_slm = 50.0
|
||
|
||
if os.environ.get("REINLOOP_SIMULATION", "").strip().lower() in {
|
||
"1", "true", "yes", "on"}:
|
||
device = SimulatedDevice()
|
||
device.connect()
|
||
result = measure_volume(device, q_in_slm=q_in_slm)
|
||
device.disconnect()
|
||
record_time = result['record_time']
|
||
p_actual = result['p_actual']
|
||
plt.figure(figsize=(10, 6))
|
||
plt.plot(record_time, p_actual, 'b.-', label='Simulated P (kPa)')
|
||
plt.title('Simulated Pressure Rise Test')
|
||
plt.xlabel('Time (s)')
|
||
plt.ylabel('Pressure (kPa)')
|
||
plt.grid(True)
|
||
plt.legend()
|
||
plt.tight_layout()
|
||
plt.show()
|
||
return
|
||
|
||
modbus_client = Easy521ModbusClient()
|
||
if modbus_client.connect():
|
||
print("成功连接到PLC")
|
||
modbus_client.start_control()
|
||
|
||
motor = MotorModbusRTUClient()
|
||
if not motor.connect():
|
||
print("电机连接失败，退出。")
|
||
return
|
||
time.sleep(1) # 增加短暂延时，等待驱动器接口就绪
|
||
if not motor.init():
|
||
print("电机初始化失败，退出。")
|
||
motor.disconnect()
|
||
return
|
||
|
||
result = measure_volume(modbus_client, motor, q_in_slm=q_in_slm)
|
||
|
||
modbus_client.stop_control()
|
||
motor.disconnect()
|
||
|
||
record_time = result['record_time']
|
||
p_actual = result['p_actual']
|
||
|
||
# ==========================================
|
||
# 画图验证
|
||
# ==========================================
|
||
plt.figure(figsize=(10, 6))
|
||
plt.plot(record_time, p_actual, 'b.-', label='Actual P (kPa)')
|
||
|
||
if result['success']:
|
||
slope = result['slope']
|
||
intercept = result['intercept']
|
||
valid_times = [t for t, p in zip(record_time, p_actual) if 50 <= p <= 150]
|
||
ideal_p = [slope * t + intercept for t in valid_times]
|
||
plt.plot(valid_times, ideal_p, 'r--', linewidth=2, label=f'Linear Fit (slope={slope:.1f})')
|
||
|
||
plt.title('Pressure Rise Test')
|
||
plt.xlabel('Time (s)')
|
||
plt.ylabel('Pressure (kPa)')
|
||
plt.grid(True)
|
||
plt.legend()
|
||
plt.tight_layout()
|
||
plt.show()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|