Compare commits

2 Commits
Author SHA1 Message Date
louis bff24e8301 新增一个py模块,在稳态后记录、输出到终端:
该次调节用的kp,ki,kd参数以及控制效果如何,包括调节时间(调节至稳态花的时间,稳态用两个条件判断:用滑动时间窗口的方差/标准差判断,或者用滑动窗口均值介于目标值上下2%误差带之间)、超调量(记录绝对数值和相对目标值的百分比)、该次控制所设置的控制范围。
2026-08-14 15:35:05 +08:00
louis 9260fe20a8 删除"111.py"和"企业微信截图" 2026-08-14 14:46:33 +08:00
11 changed files with 185 additions and 7 deletions
-7
View File
@@ -1,7 +0,0 @@
import os
# 示例 3: 末尾路径为空
path = os.path.join("/home/user", "documents", "")
print(path) # 输出: /home/user/documents/
path = os.path.join("/home/user", "documents")
print(path) # 输出: /home/user/documents
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+8
View File
@@ -46,6 +46,14 @@ OPENING_MIN_PCT = 0.0
OPENING_MAX_PCT = 100.0
MAX_OPENING_RATE_PCT_S = 20.0 # 每秒最多改变 20% 开度
# ---------------------------------------------------------------------------
# 稳态判定(用于控制性能报告)
# ---------------------------------------------------------------------------
STEADY_STATE_WINDOW_S = 2.0 # 滑动时间窗口长度
STEADY_STATE_BAND_PCT = 2.0 # 滑窗均值相对目标的允许误差带(±2%)
STEADY_STATE_STD_PCT = 2.0 # 滑窗标准差阈值(相对目标百分比)
# ---------------------------------------------------------------------------
# 阀门执行机构
# ---------------------------------------------------------------------------
+27
View File
@@ -17,6 +17,7 @@ import config
from controllers import IncrementalPID
from data_logger import FlowRunRecorder
from flow_control import FlowControlFault, FlowControlLoop
from performance_reporter import PerformanceReporter
def build_logger():
@@ -133,6 +134,7 @@ def run(target_flow_slm):
connected = False
recorder = None
reporter = None
exit_code = 0
try:
logger.info(
@@ -162,6 +164,23 @@ def run(target_flow_slm):
{"motor_closed_position": config.MOTOR_CLOSED_POSITION},
)
# 控制前读取一次当前流量,作为超调量方向的基准。
try:
initial_flow_slm = hardware.get_flow(config.FLOW_INPUT_CHANNEL)
except Exception:
initial_flow_slm = None
reporter = PerformanceReporter(
controller.pid,
window_s=config.STEADY_STATE_WINDOW_S,
band_pct=config.STEADY_STATE_BAND_PCT,
std_pct=config.STEADY_STATE_STD_PCT,
flow_range_min_slm=config.TARGET_FLOW_MIN_SLM,
flow_range_max_slm=config.TARGET_FLOW_MAX_SLM,
initial_flow_slm=initial_flow_slm,
logger=logger,
)
controller.start(initial_opening=0.0)
recorder = FlowRunRecorder(
Path(__file__).resolve().parent / config.DATA_DIRECTORY,
@@ -176,6 +195,7 @@ def run(target_flow_slm):
now = time.perf_counter()
result = controller.step(now=now)
recorder.record(result)
reporter.observe(result)
if now >= next_status_time or result.status != "OK":
logger.info(format_status(result))
@@ -221,6 +241,13 @@ def run(target_flow_slm):
recorder.close()
exit_code = max(exit_code, 7)
logger.exception("保存控制数据曲线图失败;CSV 数据仍保留")
if reporter is not None:
try:
reporter.finalize()
except Exception:
logger.exception("生成控制性能报告时发生异常")
logger.info("程序结束,退出码=%d", exit_code)
return exit_code
+150
View File
@@ -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)
Binary file not shown.

Before

Width:  |  Height:  |  Size: 21 KiB