新增一个py模块，在稳态后记录、输出到终端：

该次调节用的kp，ki，kd参数以及控制效果如何，包括调节时间（调节至稳态花的时间，稳态用两个条件判断：用滑动时间窗口的方差/标准差判断，或者用滑动窗口均值介于目标值上下2%误差带之间）、超调量（记录绝对数值和相对目标值的百分比）、该次控制所设置的控制范围。
2026-08-14 15:35:05 +08:00
parent 9260fe20a8
commit bff24e8301
9 changed files with 185 additions and 0 deletions
@@ -0,0 +1,150 @@
+"""控制性能报告：在达到稳态后输出 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)