From 14291ee98451294c46e9f98a6d162f05c81a7de6 Mon Sep 17 00:00:00 2001 From: louissun Date: Fri, 31 Jul 2026 11:29:10 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E6=A8=A1=E6=8B=9F=E7=A1=AC?= =?UTF-8?q?=E4=BB=B6=E6=B5=8B=E8=AF=95=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 19 +++++- ReinLoop/core/connection_manager.py | 49 +++++++++----- ReinLoop/core/simulated_device.py | 86 +++++++++++++++++++++++++ ReinLoop/get_V.py | 31 +++++++-- ReinLoop/ind_collector.py | 26 ++++++-- ReinLoop/tests/test_simulated_device.py | 40 ++++++++++++ ReinLoop/tool/gui.py | 25 +++++-- 7 files changed, 242 insertions(+), 34 deletions(-) create mode 100644 ReinLoop/core/simulated_device.py create mode 100644 ReinLoop/tests/test_simulated_device.py diff --git a/README.md b/README.md index 8d05f44..dbbec71 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,23 @@ python main.py 客户端服务地址和设备目录由 `api.py` 配置。控制数据、辨识数据和容积测量结果经 Server 上传；许可证、模型、公司和产线操作由 ControlPanel 完成。 +#### 无硬件模拟模式 + +需要在不连接 MT2-AM8、PLC 或串口电机的情况下测试控制流程时，在启动客户端前设置： + +```powershell +$env:REINLOOP_SIMULATION="1" +python main.py +``` + +该模式使用内存中的压力、流量和阀门模型，不创建真实 TCP/串口连接。默认值为空，客户端仍使用真实 Modbus 硬件路径。旧版 `tool/gui.py`、`get_V.py` 和 `ind_collector.py` 也支持同一环境变量。 + +恢复真实硬件模式： +```powershell +Remove-Item Env:REINLOOP_SIMULATION +python main.py +``` + ## 部署与数据 - 新生产环境使用 PostgreSQL：设置 `NODE_ENV=production` 与 `DATABASE_URL`；迁移可通过 `npm run migrate` 重复执行。 @@ -95,4 +112,4 @@ npm test - 稳定压力数组 JSON 的最终结构、两套参数的正式默认值、单位与范围仍需产品或算法确认。 - 图像是否通过目前由人工判断，尚未提供自动判定阈值或算法。 -- Electron 的第三方构建依赖存在已知安全告警；未执行可能造成破坏性升级的 `npm audit fix --force`。 \ No newline at end of file +- Electron 的第三方构建依赖存在已知安全告警；未执行可能造成破坏性升级的 `npm audit fix --force`。 diff --git a/ReinLoop/core/connection_manager.py b/ReinLoop/core/connection_manager.py index 98aea09..72a3f43 100644 --- a/ReinLoop/core/connection_manager.py +++ b/ReinLoop/core/connection_manager.py @@ -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: diff --git a/ReinLoop/core/simulated_device.py b/ReinLoop/core/simulated_device.py new file mode 100644 index 0000000..5afa4ab --- /dev/null +++ b/ReinLoop/core/simulated_device.py @@ -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 + diff --git a/ReinLoop/get_V.py b/ReinLoop/get_V.py index 74ae184..7467f59 100644 --- a/ReinLoop/get_V.py +++ b/ReinLoop/get_V.py @@ -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() diff --git a/ReinLoop/ind_collector.py b/ReinLoop/ind_collector.py index 24c4b3d..34e425a 100644 --- a/ReinLoop/ind_collector.py +++ b/ReinLoop/ind_collector.py @@ -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 diff --git a/ReinLoop/tests/test_simulated_device.py b/ReinLoop/tests/test_simulated_device.py new file mode 100644 index 0000000..370685d --- /dev/null +++ b/ReinLoop/tests/test_simulated_device.py @@ -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() diff --git a/ReinLoop/tool/gui.py b/ReinLoop/tool/gui.py index ba05fec..df19d81 100644 --- a/ReinLoop/tool/gui.py +++ b/ReinLoop/tool/gui.py @@ -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())