174 lines
6.2 KiB
Python
174 lines
6.2 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,
|
||
sample_period_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.sample_period_s = float(sample_period_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.steady_mean = None
|
||
self.steady_std = None
|
||
self.steady_error_slm = 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
|
||
# 采样时间戳会有轻微抖动。旧样本按 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]
|
||
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.steady_mean = mean
|
||
self.steady_std = std
|
||
self.steady_error_slm = target - mean
|
||
self._report()
|
||
|
||
def finalize(self):
|
||
"""运行结束调用；若始终未达到稳态则补一条提示日志。"""
|
||
if not self._reached and self.target_flow_slm is not None:
|
||
self._log("info", "本次运行未达到稳态，未生成控制性能报告")
|
||
|
||
@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_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)
|