diff --git a/ReinLoop/core/identification.py b/ReinLoop/core/identification.py index bbfb0d1..a4d0a2a 100644 --- a/ReinLoop/core/identification.py +++ b/ReinLoop/core/identification.py @@ -108,6 +108,34 @@ class IdentificationManager: This is an independent pre-scan. It does not replace or modify the subsequent PRBS collection performed by ``collect_data_with_prbs``. """ + if getattr(conn_mgr, "simulation", False): + stable_pressure_records = [ + {"distance": 1000, "pressure": 400}, + {"distance": 900, "pressure": 361}, + {"distance": 800, "pressure": 322}, + {"distance": 700, "pressure": 283}, + {"distance": 600, "pressure": 244}, + {"distance": 500, "pressure": 205}, + {"distance": 400, "pressure": 166}, + {"distance": 300, "pressure": 127}, + {"distance": 200, "pressure": 88}, + {"distance": 100, "pressure": 49}, + {"distance": 0, "pressure": 10}, + ] + timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') + filename = f"travel_stability_pressures_{timestamp}.json" + payload = {"stable_pressures": stable_pressure_records} + uploaded = self._upload_to_server( + json.dumps(payload, ensure_ascii=False, indent=2), + filename, + f"{the_folder}/ind_data", + ) + if uploaded: + self.log("行程稳态压力 JSON 上传成功,继续执行 PRBS 辨识") + else: + self.log("行程稳态压力 JSON 上传失败,继续执行 PRBS 辨识") + return payload + distances = list(range(1000, -1, -100)) settings = { "min_wait_time": 5.0, diff --git a/ReinLoop/core/simulated_device.py b/ReinLoop/core/simulated_device.py index 5afa4ab..dd7a3b8 100644 --- a/ReinLoop/core/simulated_device.py +++ b/ReinLoop/core/simulated_device.py @@ -39,10 +39,10 @@ class SimulatedDevice: 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 + # 行程与阀门开度反向:行程越大,阀门开度越小。阀门开度减小 + # 时系统压力升高;默认量程下行程 0/1000 分别对应 10/400 kPa。 + travel_ratio = max(0.0, min(1.0, self.position / 1000.0)) + target_pressure = 10.0 + (self.pressure_range - 10.0) * travel_ratio response = min(1.0, dt / 8.0) self.pressure += (target_pressure - self.pressure) * response diff --git a/ReinLoop/tests/test_simulated_device.py b/ReinLoop/tests/test_simulated_device.py index 370685d..4a80b6e 100644 --- a/ReinLoop/tests/test_simulated_device.py +++ b/ReinLoop/tests/test_simulated_device.py @@ -1,7 +1,7 @@ import importlib.util -import time import unittest from pathlib import Path +from unittest.mock import patch MODULE_PATH = Path(__file__).resolve().parents[1] / "core" / "simulated_device.py" @@ -22,18 +22,40 @@ class SimulatedDeviceTests(unittest.TestCase): 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() + class FakeClock: + def __init__(self): + self.now = 0.0 + + def monotonic(self): + return self.now + + def advance(self, duration): + self.now += duration + + clock = FakeClock() + with patch.object(SIMULATED.time, "monotonic", clock.monotonic): + device = SIMULATED.SimulatedDevice() + device.connect() + + def settled_pressure(position): + device.set_motor_position(position) + pressure = None + for _ in range(1200): + clock.advance(0.1) + pressure = device.get_pressure() + return pressure + + open_pressure = settled_pressure(0) + half_open_pressure = settled_pressure(500) + closed_pressure = settled_pressure(1000) + + self.assertAlmostEqual(open_pressure, 10.0, delta=0.1) + self.assertAlmostEqual(half_open_pressure, 205.0, delta=0.1) + self.assertAlmostEqual(closed_pressure, 400.0, delta=0.1) + self.assertGreater(closed_pressure, half_open_pressure) + self.assertGreater(half_open_pressure, open_pressure) + self.assertEqual(device.read_current_position(), 1000.0) + device.disconnect() if __name__ == "__main__":