louis
cfeedcf887
- 前馈冻结改用可配置的 3 秒流量绝对误差滑窗，修复采样抖动造成的重复解冻，并保留下游扰动后的自动更新 - 支持非单调阀特性反解、最小可测面积以下直接全关，以及闭阀端精细扫点与辨识开关 - 调整双压力判稳、最长等待时间、在线入口全开收尾和闭环四联图 - 更新配置、README、阀模型及离线测试，归档本轮实验数据与诊断产物
244 lines
9.3 KiB
Python
244 lines
9.3 KiB
Python
"""控制性能报告：在达到稳态后输出 PID 参数、调节时间、超调量和控制范围。
|
||
|
||
本模块只负责观测每个控制周期产出的 ControlStepResult，用滑动时间窗口判断
|
||
稳态，并在首次进入稳态时打印一份一次性报告；不参与任何控制计算。
|
||
"""
|
||
|
||
from collections import deque
|
||
import math
|
||
import statistics
|
||
|
||
|
||
class PerformanceReporter:
|
||
"""逐周期观测流量，判定稳态并输出控制性能报告。
|
||
|
||
稳态判定采用两个条件同时满足（AND）：
|
||
1. 流量滑窗均值落在目标流量 ±band_pct% 误差带内；
|
||
2. 阀前、阀后压力的窗口样本均位于各自窗口均值
|
||
±pressure_band_kpa 范围内。
|
||
报告只在首次判定稳态时输出一次。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
pid,
|
||
*,
|
||
window_s,
|
||
sample_period_s,
|
||
band_pct,
|
||
pressure_band_kpa,
|
||
flow_range_min_slm,
|
||
flow_range_max_slm,
|
||
initial_flow_slm=None,
|
||
logger=None,
|
||
):
|
||
self.pid = pid
|
||
self.window_s = float(window_s)
|
||
self.sample_period_s = float(sample_period_s)
|
||
self.band_pct = float(band_pct)
|
||
self.pressure_band_kpa = float(pressure_band_kpa)
|
||
self.flow_range_min_slm = float(flow_range_min_slm)
|
||
self.flow_range_max_slm = float(flow_range_max_slm)
|
||
self.initial_flow_slm = (
|
||
None if initial_flow_slm is None else float(initial_flow_slm)
|
||
)
|
||
self.logger = logger
|
||
|
||
self.target_flow_slm = None
|
||
self.settling_time_s = None
|
||
self.steady_mean = None
|
||
self.steady_std = None
|
||
self.steady_error_slm = None
|
||
self.steady_pressure_before_mean = None
|
||
self.steady_pressure_after_mean = None
|
||
self.steady_pressure_before_max_deviation = None
|
||
self.steady_pressure_after_max_deviation = None
|
||
self._first_timestamp = None
|
||
self._peak_flow_slm = None
|
||
self._min_flow_slm = None
|
||
# 元素为 (elapsed_s, flow_slm, pressure_before_kpa, pressure_after_kpa)
|
||
self._window = deque()
|
||
self._reached = False
|
||
|
||
def observe(self, result):
|
||
"""喂入一个 ControlStepResult；达到稳态时自动输出一次报告。"""
|
||
if self._reached:
|
||
return
|
||
if result.status != "OK" or result.measured_flow_slm is None:
|
||
return
|
||
|
||
flow = float(result.measured_flow_slm)
|
||
pressure_before = result.pressure_before_kpa
|
||
pressure_after = result.pressure_kpa
|
||
if pressure_before is None or pressure_after is None:
|
||
return
|
||
pressure_before = float(pressure_before)
|
||
pressure_after = float(pressure_after)
|
||
if not all(
|
||
math.isfinite(value)
|
||
for value in (flow, pressure_before, pressure_after)
|
||
):
|
||
return
|
||
target = float(result.target_flow_slm)
|
||
if target <= 0.0:
|
||
return
|
||
self.target_flow_slm = target
|
||
|
||
if self._first_timestamp is None:
|
||
self._first_timestamp = float(result.timestamp)
|
||
elapsed_s = max(0.0, float(result.timestamp) - self._first_timestamp)
|
||
|
||
if self.initial_flow_slm is None:
|
||
self.initial_flow_slm = flow
|
||
if self._peak_flow_slm is None or flow > self._peak_flow_slm:
|
||
self._peak_flow_slm = flow
|
||
if self._min_flow_slm is None or flow < self._min_flow_slm:
|
||
self._min_flow_slm = flow
|
||
|
||
self._window.append(
|
||
(elapsed_s, flow, pressure_before, pressure_after)
|
||
)
|
||
cutoff = elapsed_s - self.window_s
|
||
while self._window and self._window[0][0] < cutoff:
|
||
self._window.popleft()
|
||
|
||
if len(self._window) < 2:
|
||
return
|
||
# 采样时间戳会有轻微抖动。旧样本按 window_s 剔除后，窗口跨度通常会
|
||
# 比 window_s 少一个采样周期；若仍严格要求完整 window_s，判稳可能
|
||
# 只能依赖浮点时间戳碰巧相等。允许一个采样周期的覆盖余量。
|
||
min_span = max(0.0, self.window_s - self.sample_period_s)
|
||
if self._window[-1][0] - self._window[0][0] < min_span:
|
||
return
|
||
|
||
flows = [item[1] for item in self._window]
|
||
pressures_before = [item[2] for item in self._window]
|
||
pressures_after = [item[3] for item in self._window]
|
||
mean = statistics.mean(flows)
|
||
std = statistics.pstdev(flows)
|
||
pressure_before_mean = statistics.mean(pressures_before)
|
||
pressure_after_mean = statistics.mean(pressures_after)
|
||
pressure_before_max_deviation = max(
|
||
abs(value - pressure_before_mean) for value in pressures_before
|
||
)
|
||
pressure_after_max_deviation = max(
|
||
abs(value - pressure_after_mean) for value in pressures_after
|
||
)
|
||
|
||
band = self.band_pct / 100.0 * target
|
||
pressure_stable = (
|
||
pressure_before_max_deviation <= self.pressure_band_kpa
|
||
and pressure_after_max_deviation <= self.pressure_band_kpa
|
||
)
|
||
if abs(mean - target) <= band and pressure_stable:
|
||
self._reached = True
|
||
self.settling_time_s = elapsed_s
|
||
self.steady_mean = mean
|
||
self.steady_std = std
|
||
self.steady_error_slm = target - mean
|
||
self.steady_pressure_before_mean = pressure_before_mean
|
||
self.steady_pressure_after_mean = pressure_after_mean
|
||
self.steady_pressure_before_max_deviation = (
|
||
pressure_before_max_deviation
|
||
)
|
||
self.steady_pressure_after_max_deviation = (
|
||
pressure_after_max_deviation
|
||
)
|
||
self._report()
|
||
|
||
def finalize(self):
|
||
"""运行结束调用；若始终未达到稳态则补一条提示日志。"""
|
||
if not self._reached and self.target_flow_slm is not None:
|
||
self._log("info", "本次运行未达到稳态，未生成控制性能报告")
|
||
|
||
def reset(self):
|
||
"""清空判稳状态，使切换目标后能重新判定稳态并再次输出报告。
|
||
|
||
``initial_flow_slm`` 一并置空，观察器会在下一个有效样本上重新把它
|
||
初始化为切换目标时刻的实测流量，从而保证新目标的超调方向判断正确。
|
||
"""
|
||
self._reached = False
|
||
self.target_flow_slm = None
|
||
self.settling_time_s = None
|
||
self.steady_mean = None
|
||
self.steady_std = None
|
||
self.steady_error_slm = None
|
||
self.steady_pressure_before_mean = None
|
||
self.steady_pressure_after_mean = None
|
||
self.steady_pressure_before_max_deviation = None
|
||
self.steady_pressure_after_max_deviation = None
|
||
self._first_timestamp = None
|
||
self._peak_flow_slm = None
|
||
self._min_flow_slm = None
|
||
self.initial_flow_slm = None
|
||
self._window.clear()
|
||
|
||
@property
|
||
def steady_reached(self):
|
||
"""是否已判定达到稳态（供主循环决定是否自动停止）。"""
|
||
return self._reached
|
||
|
||
def _report(self):
|
||
target = self.target_flow_slm
|
||
initial = self.initial_flow_slm
|
||
peak = self._peak_flow_slm
|
||
minimum = self._min_flow_slm
|
||
|
||
if initial < target:
|
||
overshoot_abs = max(0.0, peak - target)
|
||
extremum_text = f"峰值 {peak:.3f} SLM"
|
||
direction = "超出目标"
|
||
elif initial > target:
|
||
overshoot_abs = max(0.0, target - minimum)
|
||
extremum_text = f"谷值 {minimum:.3f} SLM"
|
||
direction = "低于目标"
|
||
else:
|
||
overshoot_abs = 0.0
|
||
extremum_text = f"峰值 {peak:.3f} SLM"
|
||
direction = "无"
|
||
|
||
overshoot_pct = overshoot_abs / target * 100.0
|
||
|
||
lines = [
|
||
"===== 控制性能报告 =====",
|
||
(
|
||
f"PID 参数: Kp={self.pid.kp:.3f} Ki={self.pid.ki:.3f} "
|
||
f"Kd={self.pid.kd:.3f}"
|
||
),
|
||
f"目标流量: {target:.3f} SLM",
|
||
(
|
||
f"控制范围: {self.flow_range_min_slm:.3f} ~ "
|
||
f"{self.flow_range_max_slm:.3f} SLM"
|
||
),
|
||
f"调节时间(至稳态): {self.settling_time_s:.3f} s",
|
||
f"稳态流量(窗口均值): {self.steady_mean:.3f} SLM",
|
||
f"稳态标准差(窗口): {self.steady_std:.3f} SLM",
|
||
(
|
||
"阀前压力(窗口均值/最大偏差): "
|
||
f"{self.steady_pressure_before_mean:.3f} kPa / "
|
||
f"±{self.steady_pressure_before_max_deviation:.3f} kPa"
|
||
),
|
||
(
|
||
"阀后压力(窗口均值/最大偏差): "
|
||
f"{self.steady_pressure_after_mean:.3f} kPa / "
|
||
f"±{self.steady_pressure_after_max_deviation:.3f} kPa"
|
||
),
|
||
(
|
||
f"稳态误差: {self.steady_error_slm:+.3f} SLM "
|
||
f"(目标 - 稳态均值)"
|
||
),
|
||
f"初始流量: {initial:.3f} SLM",
|
||
(
|
||
f"超调量: {extremum_text}，{direction} "
|
||
f"{overshoot_abs:+.3f} SLM ({overshoot_pct:+.2f}%)"
|
||
),
|
||
"========================",
|
||
]
|
||
self._log("info", "\n".join(lines))
|
||
|
||
def _log(self, level, message, *args):
|
||
if self.logger is not None:
|
||
getattr(self.logger, level)(message, *args)
|
||
else:
|
||
print(message)
|