louis
bff24e8301
该次调节用的kp，ki，kd参数以及控制效果如何，包括调节时间（调节至稳态花的时间，稳态用两个条件判断：用滑动时间窗口的方差/标准差判断，或者用滑动窗口均值介于目标值上下2%误差带之间）、超调量（记录绝对数值和相对目标值的百分比）、该次控制所设置的控制范围。
151 lines
5.1 KiB
Python
151 lines
5.1 KiB
Python
"""控制性能报告：在达到稳态后输出 PID 参数、调节时间、超调量和控制范围。
|
||
|
||
本模块只负责观测每个控制周期产出的 ControlStepResult，用滑动时间窗口判断
|
||
稳态，并在首次进入稳态时打印一份一次性报告；不参与任何控制计算。
|
||
"""
|
||
|
||
from collections import deque
|
||
import math
|
||
import statistics
|
||
|
||
|
||
class PerformanceReporter:
|
||
"""逐周期观测流量，判定稳态并输出控制性能报告。
|
||
|
||
稳态判定采用两个条件同时满足（AND）：
|
||
1. 滑窗均值落在目标流量 ±band_pct% 误差带内；
|
||
2. 滑窗标准差低于 std_pct% 目标流量阈值。
|
||
报告只在首次判定稳态时输出一次。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
pid,
|
||
*,
|
||
window_s,
|
||
band_pct,
|
||
std_pct,
|
||
flow_range_min_slm,
|
||
flow_range_max_slm,
|
||
initial_flow_slm=None,
|
||
logger=None,
|
||
):
|
||
self.pid = pid
|
||
self.window_s = float(window_s)
|
||
self.band_pct = float(band_pct)
|
||
self.std_pct = float(std_pct)
|
||
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._first_timestamp = None
|
||
self._peak_flow_slm = None
|
||
self._min_flow_slm = None
|
||
self._window = deque() # 元素为 (elapsed_s, flow_slm)
|
||
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)
|
||
if not math.isfinite(flow):
|
||
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))
|
||
cutoff = elapsed_s - self.window_s
|
||
while self._window and self._window[0][0] < cutoff:
|
||
self._window.popleft()
|
||
|
||
if len(self._window) < 2:
|
||
return
|
||
if self._window[-1][0] - self._window[0][0] < self.window_s:
|
||
return
|
||
|
||
flows = [item[1] for item in self._window]
|
||
mean = statistics.mean(flows)
|
||
std = statistics.pstdev(flows)
|
||
|
||
band = self.band_pct / 100.0 * target
|
||
std_threshold = self.std_pct / 100.0 * target
|
||
if abs(mean - target) <= band and std <= std_threshold:
|
||
self._reached = True
|
||
self.settling_time_s = elapsed_s
|
||
self._report()
|
||
|
||
def finalize(self):
|
||
"""运行结束调用；若始终未达到稳态则补一条提示日志。"""
|
||
if not self._reached and self.target_flow_slm is not None:
|
||
self._log("info", "本次运行未达到稳态，未生成控制性能报告")
|
||
|
||
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"初始流量: {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)
|