louis
c1a4a2a229
新增可配置开环扫点、多 CSV 前馈表拟合及结果文件生成；补充闭环停止、手动开度安全互锁和 PID 未启用状态显示，并同步更新自动化测试、使用文档、打包配置与 Windows 可执行程序。
361 lines
12 KiB
Python
361 lines
12 KiB
Python
import logging
|
||
from pathlib import Path
|
||
from queue import Empty
|
||
import time
|
||
|
||
import pytest
|
||
|
||
import config
|
||
from control_runtime import (
|
||
ControlWorker,
|
||
build_open_loop_points,
|
||
load_model,
|
||
validate_device_settings,
|
||
validate_opening,
|
||
validate_open_loop_settings,
|
||
validate_target,
|
||
)
|
||
|
||
|
||
class FakeHardware:
|
||
def __init__(self, *, connect_ok=True):
|
||
self.connect_ok = connect_ok
|
||
self.connected = False
|
||
self.disconnect_count = 0
|
||
self.motor_commands = []
|
||
|
||
def connect(self):
|
||
self.connected = self.connect_ok
|
||
return self.connect_ok
|
||
|
||
def disconnect(self):
|
||
self.connected = False
|
||
self.disconnect_count += 1
|
||
|
||
def get_flow(self, _address):
|
||
return 20.0
|
||
|
||
def get_pressure(self, address):
|
||
if address == config.PRESSURE_BEFORE_ADDR:
|
||
return 200.0
|
||
return 100.0
|
||
|
||
def set_motor_position(self, position, channel=None):
|
||
self.motor_commands.append((float(position), channel))
|
||
return True
|
||
|
||
|
||
class OverpressureHardware(FakeHardware):
|
||
def get_pressure(self, address):
|
||
if address == config.PRESSURE_BEFORE_ADDR:
|
||
return config.MAX_PRESSURE_KPA + 1.0
|
||
return 100.0
|
||
|
||
|
||
class MemoryOpenLoopRecorder:
|
||
def __init__(self):
|
||
self.csv_path = Path("memory_open_loop.csv")
|
||
self.image_path = Path("memory_open_loop.png")
|
||
self.sample_count = 0
|
||
self.rows = []
|
||
self.finalize_count = 0
|
||
|
||
def record(self, **row):
|
||
self.rows.append(row)
|
||
self.sample_count += 1
|
||
|
||
def finalize(self):
|
||
self.finalize_count += 1
|
||
return self.image_path if self.sample_count else None
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
|
||
def wait_for_event(worker, predicate, timeout=2.0):
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
event = worker.events.get(timeout=0.1)
|
||
except Empty:
|
||
continue
|
||
if predicate(event):
|
||
return event
|
||
raise AssertionError("等待后台线程事件超时")
|
||
|
||
|
||
def make_logger():
|
||
logger = logging.getLogger(f"test-control-runtime-{time.monotonic_ns()}")
|
||
logger.addHandler(logging.NullHandler())
|
||
return logger
|
||
|
||
|
||
def test_validate_ui_values():
|
||
settings = validate_device_settings(" 192.168.1.88 ", "502", "1")
|
||
assert settings.host == "192.168.1.88"
|
||
assert settings.port == 502
|
||
assert settings.slave_id == 1
|
||
assert validate_target("50.5") == 50.5
|
||
assert validate_opening("25.5") == 25.5
|
||
with pytest.raises(ValueError, match="0～100"):
|
||
validate_opening("100.1")
|
||
|
||
|
||
def test_builtin_model_is_valve_model_004():
|
||
info = load_model()
|
||
assert info.is_builtin is True
|
||
assert info.path.name == "valve_model_004.json"
|
||
assert info.point_count == 80
|
||
assert info.model.motor_open == config.MOTOR_OPEN_POSITION
|
||
assert info.model.motor_closed == config.MOTOR_CLOSED_POSITION
|
||
|
||
|
||
def test_validate_and_build_gui_open_loop_points():
|
||
settings = validate_open_loop_settings(
|
||
min_opening_pct="0",
|
||
max_opening_pct="10",
|
||
opening_step_pct="5",
|
||
random_seed="42",
|
||
include_fine_scan=False,
|
||
)
|
||
points = build_open_loop_points(settings)
|
||
assert sorted(opening for opening, _position in points) == [0.0, 5.0, 10.0]
|
||
assert all(
|
||
config.MOTOR_OPEN_POSITION <= position <= config.MOTOR_CLOSED_POSITION
|
||
for _opening, position in points
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="开度范围"):
|
||
validate_open_loop_settings(
|
||
min_opening_pct=50,
|
||
max_opening_pct=50,
|
||
include_fine_scan=False,
|
||
)
|
||
|
||
|
||
def test_worker_connect_control_and_safe_disconnect():
|
||
hardware = FakeHardware()
|
||
worker = ControlWorker(make_logger(), hardware_factory=lambda _settings: hardware)
|
||
worker.start()
|
||
wait_for_event(
|
||
worker,
|
||
lambda event: event.get("kind") == "state"
|
||
and event.get("state") == "DISCONNECTED",
|
||
)
|
||
|
||
settings = validate_device_settings("192.168.1.88", 502, 1)
|
||
worker.request_connect(settings)
|
||
wait_for_event(
|
||
worker,
|
||
lambda event: event.get("kind") == "state"
|
||
and event.get("state") == "MONITORING",
|
||
)
|
||
assert hardware.connected is True
|
||
assert hardware.motor_commands[0] == (
|
||
config.MOTOR_OPEN_POSITION,
|
||
config.MOTOR_OUTPUT_ADDR,
|
||
)
|
||
|
||
worker.request_target(50.0)
|
||
wait_for_event(
|
||
worker,
|
||
lambda event: event.get("kind") == "target_applied",
|
||
)
|
||
telemetry = wait_for_event(
|
||
worker,
|
||
lambda event: event.get("kind") == "telemetry"
|
||
and event.get("target_flow_slm") == 50.0,
|
||
)
|
||
assert telemetry["measured_flow_slm"] == 20.0
|
||
assert 0.0 <= telemetry["opening_pct"] <= 100.0
|
||
assert telemetry["pid_mode"] in {"FAR", "NEAR"}
|
||
|
||
worker.request_disconnect()
|
||
wait_for_event(
|
||
worker,
|
||
lambda event: event.get("kind") == "state"
|
||
and event.get("state") == "DISCONNECTED",
|
||
)
|
||
assert hardware.connected is False
|
||
assert hardware.motor_commands[-1] == (
|
||
config.MOTOR_OPEN_POSITION,
|
||
config.MOTOR_OUTPUT_ADDR,
|
||
)
|
||
assert hardware.disconnect_count == 1
|
||
|
||
worker.request_shutdown()
|
||
wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
|
||
worker.thread.join(timeout=1.0)
|
||
assert not worker.is_alive()
|
||
|
||
|
||
def test_worker_runs_open_loop_scan_and_keeps_connection():
|
||
hardware = FakeHardware()
|
||
recorder = MemoryOpenLoopRecorder()
|
||
worker = ControlWorker(
|
||
make_logger(),
|
||
hardware_factory=lambda _settings: hardware,
|
||
open_loop_recorder_factory=lambda: recorder,
|
||
)
|
||
worker.start()
|
||
wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
|
||
worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
|
||
wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
|
||
|
||
settings = validate_open_loop_settings(
|
||
min_opening_pct=0,
|
||
max_opening_pct=1,
|
||
opening_step_pct=1,
|
||
random_seed=1,
|
||
sample_period_s=0.005,
|
||
steady_window_s=0.01,
|
||
max_wait_s=0.03,
|
||
include_fine_scan=False,
|
||
)
|
||
worker.request_open_loop_start(settings)
|
||
started = wait_for_event(
|
||
worker, lambda event: event.get("kind") == "open_loop_started"
|
||
)
|
||
assert started["total_steps"] == 2
|
||
finished = wait_for_event(
|
||
worker, lambda event: event.get("kind") == "open_loop_finished"
|
||
)
|
||
assert finished["completed"] is True
|
||
assert finished["sample_count"] == recorder.sample_count
|
||
assert recorder.sample_count >= 4
|
||
assert recorder.finalize_count == 1
|
||
wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
|
||
assert hardware.connected is True
|
||
assert hardware.motor_commands[-1] == (
|
||
config.MOTOR_OPEN_POSITION,
|
||
config.MOTOR_OUTPUT_ADDR,
|
||
)
|
||
|
||
worker.request_shutdown()
|
||
wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
|
||
worker.thread.join(timeout=1.0)
|
||
assert not worker.is_alive()
|
||
|
||
|
||
def test_worker_stops_closed_loop_at_full_open_and_accepts_manual_opening():
|
||
hardware = FakeHardware()
|
||
worker = ControlWorker(make_logger(), hardware_factory=lambda _settings: hardware)
|
||
worker.start()
|
||
wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
|
||
worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
|
||
wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
|
||
|
||
worker.request_target(50.0)
|
||
wait_for_event(worker, lambda event: event.get("kind") == "target_applied")
|
||
worker.request_manual_opening(25.0)
|
||
warning = wait_for_event(
|
||
worker,
|
||
lambda event: event.get("kind") == "warning"
|
||
and "先停止流量控制" in event.get("message", ""),
|
||
)
|
||
assert "先停止流量控制" in warning["message"]
|
||
|
||
worker.request_stop_control()
|
||
stopped = wait_for_event(
|
||
worker, lambda event: event.get("kind") == "control_stopped"
|
||
)
|
||
assert stopped["opening_pct"] == 100.0
|
||
assert stopped["motor_position"] == config.MOTOR_OPEN_POSITION
|
||
wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
|
||
assert hardware.motor_commands[-1] == (
|
||
config.MOTOR_OPEN_POSITION,
|
||
config.MOTOR_OUTPUT_ADDR,
|
||
)
|
||
|
||
worker.request_manual_opening(25.0)
|
||
manual = wait_for_event(
|
||
worker, lambda event: event.get("kind") == "manual_opening_applied"
|
||
)
|
||
assert manual["opening_pct"] == 25.0
|
||
assert manual["motor_position"] == pytest.approx(950.0)
|
||
wait_for_event(
|
||
worker,
|
||
lambda event: event.get("kind") == "telemetry"
|
||
and event.get("pid_mode") == "未启用",
|
||
)
|
||
assert hardware.motor_commands[-1] == (950.0, config.MOTOR_OUTPUT_ADDR)
|
||
|
||
worker.request_shutdown()
|
||
wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
|
||
worker.thread.join(timeout=1.0)
|
||
assert not worker.is_alive()
|
||
|
||
|
||
def test_disconnect_during_open_loop_finalizes_partial_data():
|
||
hardware = FakeHardware()
|
||
recorder = MemoryOpenLoopRecorder()
|
||
worker = ControlWorker(
|
||
make_logger(),
|
||
hardware_factory=lambda _settings: hardware,
|
||
open_loop_recorder_factory=lambda: recorder,
|
||
)
|
||
worker.start()
|
||
wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
|
||
worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
|
||
wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
|
||
settings = validate_open_loop_settings(
|
||
min_opening_pct=0,
|
||
max_opening_pct=100,
|
||
opening_step_pct=1,
|
||
sample_period_s=0.02,
|
||
steady_window_s=1,
|
||
max_wait_s=1,
|
||
include_fine_scan=False,
|
||
)
|
||
worker.request_open_loop_start(settings)
|
||
wait_for_event(worker, lambda event: event.get("kind") == "open_loop_started")
|
||
worker.request_disconnect()
|
||
finished = wait_for_event(
|
||
worker, lambda event: event.get("kind") == "open_loop_finished"
|
||
)
|
||
assert finished["completed"] is False
|
||
assert finished["reason"] == "设备断开或程序退出"
|
||
assert recorder.finalize_count == 1
|
||
wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
|
||
assert hardware.connected is False
|
||
|
||
worker.request_shutdown()
|
||
wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
|
||
worker.thread.join(timeout=1.0)
|
||
assert not worker.is_alive()
|
||
|
||
|
||
def test_worker_connection_failure_reports_fault_without_hardware_access():
|
||
hardware = FakeHardware(connect_ok=False)
|
||
worker = ControlWorker(make_logger(), hardware_factory=lambda _settings: hardware)
|
||
worker.start()
|
||
wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
|
||
worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
|
||
fault = wait_for_event(worker, lambda event: event.get("kind") == "fault")
|
||
assert "无法连接 PLC" in fault["message"]
|
||
assert hardware.motor_commands == []
|
||
assert hardware.disconnect_count == 1
|
||
|
||
worker.request_shutdown()
|
||
wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
|
||
worker.thread.join(timeout=1.0)
|
||
assert not worker.is_alive()
|
||
|
||
|
||
def test_monitoring_stage_overpressure_triggers_safe_disconnect():
|
||
hardware = OverpressureHardware()
|
||
worker = ControlWorker(make_logger(), hardware_factory=lambda _settings: hardware)
|
||
worker.start()
|
||
wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
|
||
worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
|
||
wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
|
||
fault = wait_for_event(worker, lambda event: event.get("kind") == "fault")
|
||
assert "超过上限" in fault["message"]
|
||
assert hardware.connected is False
|
||
assert hardware.motor_commands[-1][0] == config.MOTOR_OPEN_POSITION
|
||
|
||
worker.request_shutdown()
|
||
wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
|
||
worker.thread.join(timeout=1.0)
|
||
assert not worker.is_alive()
|