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

新增模拟硬件测试功能

Browse Source
This commit is contained in:
louis
2026-07-31 11:29:10 +08:00
parent 692d8dd18a
commit 14291ee984
7 changed files with 242 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ReinLoop/core/connection_manager.py
+32 -17
View File
@@ -4,15 +4,21 @@
纯业务逻辑，无 UI 依赖。通过回调与 UI 层通信。
"""
import time
from PcControl import MT2AM8Client
import os
import time
from PcControl import MT2AM8Client
from core.simulated_device import SimulatedDevice
class ConnectionManager:
"""管理 MT2-AM8 模块连接的生命周期"""
def __init__(self):
self.modbus_client = None # MT2AM8Client (TCP 读压力/流量，控电机)
def __init__(self, simulation=None):
self.modbus_client = None # MT2AM8Client (TCP 读压力/流量，控电机)
if simulation is None:
simulation = os.environ.get("REINLOOP_SIMULATION", "").strip().lower() \
in {"1", "true", "yes", "on"}
self.simulation = bool(simulation)
self._pressure_addr = 0 # 压力传感器模拟量通道地址
self._flowmeter_addr = None # 流量计模拟量通道地址（None=使用手动输入）
self._motor_addr = 0 # 电机模拟量输出通道地址
@@ -61,19 +67,28 @@ class ConnectionManager:
self._motor_addr = motor_addr
self._flowmeter_addr = flowmeter_addr
# 创建 MT2-AM8 客户端
self.modbus_client = MT2AM8Client(
host=tcp_ip,
port=tcp_port,
pressure_range=pressure_range,
flow_range=flow_range,
)
if not self.modbus_client.connect():
self.log(f"连接 MT2-AM8 模块失败: {tcp_ip}:{tcp_port}")
return False
self.log(f"成功连接到 MT2-AM8 模块: {tcp_ip}:{tcp_port}")
if self.simulation:
self.modbus_client = SimulatedDevice(
pressure_range=pressure_range,
flow_range=flow_range,
)
if not self.modbus_client.connect():
return False
self.log("模拟设备已连接（未访问真实 TCP/串口硬件）")
else:
# 真实硬件路径：保留原有 MT2-AM8 Modbus TCP 实现。
self.modbus_client = MT2AM8Client(
host=tcp_ip,
port=tcp_port,
pressure_range=pressure_range,
flow_range=flow_range,
)
if not self.modbus_client.connect():
self.log(f"连接 MT2-AM8 模块失败: {tcp_ip}:{tcp_port}")
return False
self.log(f"成功连接到 MT2-AM8 模块: {tcp_ip}:{tcp_port}")
self.log(f"压力地址: {pressure_addr}, 电机地址: {motor_addr}, 流量计地址: {flowmeter_addr}")
if self._on_status_change:
ReinLoop/core/simulated_device.py
+86
View File
@@ -0,0 +1,86 @@
"""Deterministic in-memory device used for hardware-free ReinLoop tests.
The public methods intentionally mirror the MT2AM8 client and the small
subset of the legacy PLC/motor clients used by ``tool/gui.py``. No sockets,
serial ports, or Modbus objects are created here.
"""
import time
class SimulatedDevice:
"""A small pressure/flow/valve model for offline control-loop testing."""
def __init__(self, pressure_range=400.0, flow_range=100.0,
initial_pressure=0.0):
self.pressure_range = float(pressure_range)
self.flow_range = float(flow_range)
self.pressure = float(initial_pressure)
self.flow = min(self.flow_range, 50.0)
self.position = 1000.0
self.connected = False
self.last_update = time.monotonic()
self.command_log = []
def connect(self):
self.connected = True
self.last_update = time.monotonic()
print("模拟设备连接成功（未访问真实硬件）")
return True
def disconnect(self):
self.connected = False
print("模拟设备已断开")
def _advance(self):
now = time.monotonic()
dt = max(0.0, min(now - self.last_update, 0.5))
self.last_update = now
if not self.connected or dt == 0:
return
# MT2AM8 的行程控制在当前程序中是反向阀位：较小行程表示更大
# 进气。模型因此让压力向由行程决定的目标值缓慢靠近。
opening = max(0.0, min(1.0, 1.0 - self.position / 1000.0))
target_pressure = self.pressure_range * opening
response = min(1.0, dt / 8.0)
self.pressure += (target_pressure - self.pressure) * response
def get_pressure(self, channel=0):
self._advance()
return max(0.0, self.pressure)
def get_flow(self, channel=0):
self._advance()
opening = max(0.0, min(1.0, 1.0 - self.position / 1000.0))
return self.flow * (0.25 + 0.75 * opening)
def set_motor_position(self, position, channel=0):
self._advance()
position = float(position)
if not 0.0 <= position <= 1000.0:
return False
self.position = position
self.command_log.append({"position": position, "channel": channel,
"time": time.monotonic()})
return True
# Compatibility methods used by the legacy GUI and standalone helpers.
def get_current_p(self):
return self.get_pressure(0)
def read_current_position(self):
return self.position
def set_position(self, position):
return self.set_motor_position(position)
def init(self):
return True
def start_control(self):
return True
def stop_control(self):
return True
ReinLoop/get_V.py
+26 -5
View File
@@ -137,14 +137,35 @@ def measure_volume(conn_mgr,
return result
def main():
"""独立运行入口：自建连接、跑测试、画图验证。"""
import matplotlib.pyplot as plt
from PcControl import Easy521ModbusClient, MotorModbusRTUClient
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
modbus_client = Easy521ModbusClient()
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()
ReinLoop/ind_collector.py
+21 -5
View File
@@ -204,17 +204,33 @@ def collect_data_with_prbs(conn_mgr,
}
# ==================== 6. 主程序 ====================
def main():
"""独立运行入口：自建连接、采集、存盘。"""
from PcControl import Easy521ModbusClient, MotorModbusRTUClient
def main():
"""独立运行入口：自建连接、采集、存盘。"""
from PcControl import Easy521ModbusClient, MotorModbusRTUClient
import os
from core.simulated_device import SimulatedDevice
dt = 0.1
n_order = 7 # 码元数 127
t_c = 5 # 码元周期 5 秒
levels = [10, 20, 30, 40, 50, 60, 70, 80] # 8 个电平，对应 group_bits=3
# 连接 PLC
modbus_client = Easy521ModbusClient()
if os.environ.get("REINLOOP_SIMULATION", "").strip().lower() in {
"1", "true", "yes", "on"}:
device = SimulatedDevice()
device.connect()
q_in_val = float(input("请输入实验时的流量 (SLM): "))
try:
collect_data_with_prbs(device,
q_in_val=q_in_val, dt=dt, n_order=n_order,
t_c=t_c, levels=levels, save_dir="test_data",
repeat=2)
finally:
device.disconnect()
return
# 真实硬件路径：保留原有 PLC/电机 Modbus TCP/RTU 连接代码。
modbus_client = Easy521ModbusClient()
if not modbus_client.connect():
print("无法连接 PLC，退出")
return
ReinLoop/tests/test_simulated_device.py
+40
View File
@@ -0,0 +1,40 @@
import importlib.util
import time
import unittest
from pathlib import Path
MODULE_PATH = Path(__file__).resolve().parents[1] / "core" / "simulated_device.py"
SPEC = importlib.util.spec_from_file_location("simulated_device_under_test", MODULE_PATH)
SIMULATED = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(SIMULATED)
class SimulatedDeviceTests(unittest.TestCase):
def test_connect_read_and_write_without_hardware(self):
device = SIMULATED.SimulatedDevice()
self.assertTrue(device.connect())
self.assertTrue(device.set_motor_position(0))
self.assertIsInstance(device.get_pressure(), float)
self.assertIsInstance(device.get_flow(), float)
self.assertEqual(len(device.command_log), 1)
device.disconnect()
self.assertFalse(device.connected)
def test_pressure_model_responds_to_valve_position(self):
device = SIMULATED.SimulatedDevice()
device.connect()
device.set_motor_position(0)
time.sleep(0.02)
open_pressure = device.get_pressure()
device.set_motor_position(1000)
time.sleep(0.02)
closed_pressure = device.get_pressure()
self.assertGreaterEqual(open_pressure, 0.0)
self.assertGreaterEqual(closed_pressure, 0.0)
self.assertEqual(device.read_current_position(), 1000.0)
device.disconnect()
if __name__ == "__main__":
unittest.main()
ReinLoop/tool/gui.py
+19 -6
View File
@@ -26,7 +26,7 @@ import serial.tools.list_ports
from stable_baselines3 import SAC
import torch
# from PressureEnv import CustomPressureEnv
from PcControl import Easy521ModbusClient, MotorModbusRTUClient
from PcControl import Easy521ModbusClient, MotorModbusRTUClient
# from zzp import SECRET_KEY
# from PcControl import PressureModbusRTUClient, MotorModbusRTUClient
from controllers import IncrementalPID
@@ -35,8 +35,9 @@ 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 api import base_url, data_record_url, the_folder
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)
@@ -1127,9 +1128,21 @@ class ControlGUI:
else:
self.connect_plc()
def connect_plc(self):
"""连接到PLC及电机"""
# ---------------- 读取 Modbus TCP 参数 ----------------
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())