定时导出控制器运行数据 → 自动生成每日控制效果评估文档每天早上 8 点工艺工程师打开电脑的第一件事是翻看昨天 30 个控制回路各自表现如何。如果靠人工从 DCS 里一条条翻趋势、算超调、判振荡两个小时就搭进去了——而这原本是脚本 30 秒就能干完的活。—— 基于哈尔滨工程大学《工业过程控制》第 5 章PID 控制、第 6 章单回路控制系统设计一、实际应用场景描述在石化、煤化工、精细化工、热电、制药等流程工业里DCS/SCADA 系统每天都会默默记录下成百上千个控制回路的 PV/SP/OP 数据。这些数据是工艺工程师判断昨天这炉子稳不稳、那台泵是不是又振荡了的第一手证据┌──────────────────────────────────────────────────────┐│ 每日控制效果评估系统架构 ││ ││ [DCS / SCADA 历史库] ││ • 霍尼韦尔 Experion PKS ││ • 横河 Centum VP ││ • 西门子 PCS 7 ││ • 中控 ECS-700 ││ │ ││ ▼ (每日定时导出 CSV) ││ ┌──────────────────────────────────┐ ││ │ 数据加载与校验模块 │ ││ │ • 读取 CSV/数据库 │ ││ │ • 按回路拆分 PV/SP/OP │ ││ │ • 坏点剔除通讯中断、超限 │ ││ └──────────────┬───────────────────────┘ ││ │ ││ ▼ ││ ┌──────────────────────────────────┐ ││ │ 指标计算引擎核心 │ ││ │ ┌────────────┐ ┌────────────┐ │ ││ │ │ ISE/IAE/ │ │ 超调量/ │ │ ││ │ │ ITAE 积分 │ │ 稳态误差 │ │ ││ │ └────────────┘ └────────────┘ │ ││ │ ┌────────────┐ ┌────────────┐ │ ││ │ │ 调节时间 │ │ 振荡周期 │ │ ││ │ │ 计算 │ │ 计数 │ │ ││ │ └────────────┘ └────────────┘ │ ││ └──────────────┬───────────────────────┘ ││ │ ││ ▼ ││ ┌──────────────────────────────────┐ ││ │ 综合评级与处置建议 │ ││ │ • 优秀/良好/一般/较差 四级 │ ││ │ • 自动给出处置建议 │ ││ │ • 积分饱和/持续振荡识别 │ ││ └──────────────┬───────────────────────┘ ││ │ ││ ▼ ││ ┌──────────────────────────────────┐ ││ │ 报告输出 │ ││ │ • CSV → Excel 归档分析 │ ││ │ • TXT → 邮件正文/值班记录 │ ││ │ • 定时调度crontab/任务计划 │ ││ └──────────────────────────────────┘ ││ ││ 典型使用流程: ││ 1. DCS 每日 23:55 自动导出昨日历史数据到共享目录 ││ 2. 本脚本 00:05 定时触发加载 CSV ││ 3. 逐回路计算 ISE/IAE/ITAE/超调/稳态误差/振荡 ││ 4. 按阈值自动评级生成 CSV TXT 报告 ││ 5. TXT 报告自动邮件发送给当班工程师 ││ 6. 工程师上班第一件事看报告抓较差回路处置 │└──────────────────────────────────────────────────────┘典型应用场景场景 回路数量 采样频率 控制难点乙烯裂解装置 80~120 个 1Hz 强耦合、大滞后常减压蒸馏 50~80 个 1Hz 多回路级联锅炉汽包水位 3~5 个 5Hz 虚假水位、快动态精馏塔 20~40 个 1Hz 非线性、约束控制制药发酵罐 10~20 个 0.1Hz 慢动态、严格合规二、引入痛点2.1 现场的真实困境场景 现场反馈 根因翻不完 300 个回路每个翻 30 秒趋势半天就没了 纯人工无自动化工具算不准 凭眼睛看超调10 个人估出 10 个值 缺乏统一量化指标记不住 上周这个回路还行这周怎么又抖了 无历史趋势对比报不上 发现振荡了打电话给工艺人家早下班了 信息传递延迟标准乱 啥叫稳态3% 算不算差各说各的 无统一评估阈值留不下 邮件里说一嘴就完了半年后查不到 无结构化归档2.2 核心矛盾每日控制效果评估的核心矛盾是海量运行数据与有限人工工时之间的不匹配。 一个中型化工厂每天产生的控制器数据以 GB 计而工艺工程师每天能花在巡检上的时间不超过 2 小时——必须靠程序把看趋势、算指标、判好坏这三件事自动化。2.3 三种解决思路对比思路 核心思想 优势 劣势DCS 自带报表 用 DCS 内置报表工具 无需额外软件集成度高 灵活性差跨系统难定制成本高Excel 宏 VBA 读取 CSV 画图 工程师熟悉上手快 性能差难维护跨平台弱Python 脚本 独立程序定时跑 灵活强大易版本管理可对接数据库 需一点编程基础本文解决三、核心逻辑讲解3.1 理论基础PID 控制性能指标本工具基于哈工程《工业过程控制》第 5 章PID 控制与第 6 章单回路控制系统设计① 三大积分性能指标是衡量控制品质的黄金三件套教材 P112 起详细推导ISE \int_0^\infty e^2(t)\,dt \qquad \text{平方误差积分}IAE \int_0^\infty |e(t)|\,dt \qquad \text{绝对误差积分}ITAE \int_0^\infty t\,|e(t)|\,dt \qquad \text{时间加权绝对误差}其中 e(t) SP - PV 偏差。三者的区别指标 对大误差的惩罚 对晚消除误差的惩罚 工程含义ISE 极强平方 无 对大偏差零容忍适合安全关键回路IAE 线性 无 直观与平均误差相关ITAE 线性 极强乘 t 惩罚迟迟不收敛最适合评估调节时间离散化实现采样周期 \Delta t ISE \approx \sum_{k0}^{N} e_k^2 \cdot \Delta tIAE \approx \sum_{k0}^{N} |e_k| \cdot \Delta tITAE \approx \sum_{k0}^{N} (k\cdot\Delta t) \cdot |e_k| \cdot \Delta t② 超调量与稳态误差\sigma\% \frac{PV_{max} - SP}{SP} \times 100\% \quad (\text{仅统计正向超调})稳态误差取最后 20% 数据的平均偏差归一化e_{ss}\% \frac{1}{N_{tail}}\sum_{kN-N_{tail}}^{N} |e_k| \cdot \frac{100\%}{SP}③ 调节时间进入 \pm 2\% 误差带后不再跑出的最早时刻。这是 ITAE 指标的物理对应量。④ 振荡识别误差序列过零次数的一半即为振荡半周期数。连续多个周期 持续振荡是 PID 参数不当的典型症状。3.2 综合评级逻辑综合评级决策树借鉴哈工程教材分级思路:基础等级 优秀EXCELLENTif 超调量 15%:等级 较差POOR原因 超调过大(xx%15%)elif 超调量 5%:if 等级仍是优秀: 等级 一般FAIR原因 超调偏高(xx%5%)if 稳态误差 3%:等级 较差POOR原因 稳态误差大(xx%3%)elif 稳态误差 1%:if 等级 优秀: 等级 良好GOOD原因 稳态误差偏高(xx%1%)if 振荡周期数 3:if 等级 优秀: 等级 良好GOOD原因 存在持续振荡(N周期)if 末段25%误差均值/SP 5%:if 等级 优秀: 等级 良好GOOD原因 误差长期未归零(疑似积分饱和)if 没有触发任何降级:原因 各指标均在优秀范围内输出: (等级, 原因)3.3 定时调度架构定时调度方案对比:┌─────────────────────────────────────────────────────┐│ 方案A: 操作系统级定时推荐生产环境 ││ • Linux: crontab -e ││ → 55 23 * * * /usr/bin/python3 /path/to/script ││ • Windows: 任务计划程序 ││ → 触发器: 每天 23:55 ││ → 操作: 启动 python.exe 脚本路径 ││ 优点: 系统级不依赖 Python 进程常驻 ││ 缺点: 配置略繁琐 │├─────────────────────────────────────────────────────┤│ 方案B: Python 内循环适合测试/小项目 ││ • scheduler.run_daily(hour23, minute55) ││ • 内部 while True time.sleep() ││ 优点: 零配置开箱即用 ││ 缺点: 进程需常驻重启后需手动拉起 │├─────────────────────────────────────────────────────┤│ 方案C: 容器化 K8s CronJob大规模部署 ││ • Docker 镜像打包脚本 依赖 ││ • CronJob 每天触发 Pod 执行 ││ 优点: 弹性、可观测、易回滚 ││ 缺点: 基础设施要求高 │└─────────────────────────────────────────────────────┘推荐: 生产用方案A开发测试用方案B四、代码讲解面向对象设计4.1 类结构总览类名 职责 设计模式LoopHealthGrade 回路健康等级枚举 枚举类型ReportFormat 报告格式枚举 枚举类型LoopConfig 单回路配置dataclass 值对象AssessmentThresholds 评估阈值dataclass 值对象LoopMetrics 单回路指标结果dataclass 值对象ControllerDataLoader 数据加载与校验 单一职责MetricsCalculator 指标计算与评级 策略模式ReportGenerator 报告格式化输出 模板方法DailyReportScheduler 调度器聚合根 聚合根/外观模式4.2 完整源代码单文件约 670 行直接复制下方代码保存为pid_daily_report.py 即可运行仅需 Python 3.8 标准库无需安装任何第三方包。detailssummary/summary定时导出控制器运行数据 → 自动生成每日控制效果评估文档参考: 哈尔滨工程大学《工业过程控制》第5章PID控制、第6章单回路控制系统设计作者: 工业控制与上位机开发工程师适用: CSDN技术博客完整发布版(单文件, ~670行)import csvimport osimport mathimport timeimport datetimeimport loggingfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuplefrom enum import Enum, autofrom statistics import mean, stdev# # 1. 基础数据结构(枚举 值对象)# class LoopHealthGrade(Enum):回路健康等级(参考工业标准)EXCELLENT (优秀, ✅)GOOD (良好, ☑️)FAIR (一般, ⚠️)POOR (较差, ❌)def __init__(self, label: str, icon: str):self.label labelself.icon iconclass ReportFormat(Enum):报告输出格式CSV csvTXT txtBOTH bothdataclassclass LoopConfig:单个控制回路的配置 —— 值对象loop_id: str # 回路编号, 如 TIC-101loop_name: str # 回路名称, 如 反应器温度控制pv_column: str # 过程值(PV)在CSV中的列名sp_column: str # 设定值(SP)在CSV中的列名op_column: str # 操作值(OP)在CSV中的列名setpoint_nominal: float # 标称设定值(用于归一化)deadband: float 1.0 # 稳态误差允许带(绝对值)sample_interval_sec: float 1.0 # 采样间隔(秒)dataclassclass AssessmentThresholds:评估阈值配置 —— 值对象(参考哈工程教材性能指标)overshoot_good: float 5.0 # 超调5% → 降为良好overshoot_fair: float 15.0 # 超调15% → 降为一般/较差steady_error_good: float 1.0 # 稳态误差1% → 降为良好steady_error_fair: float 3.0 # 稳态误差3% → 降为一般/较差oscillation_min_cycles: int 3 # 至少N个周期才算振荡dataclassclass LoopMetrics:单个回路的计算指标 —— 值对象loop_id: strgrade: LoopHealthGradegrade_reason: str# 哈工程教材四大积分指标ise: float # 平方误差积分 ∫e²(t)dtiae: float # 绝对误差积分 ∫|e(t)|dtitae: float # 时间加权绝对误差 ∫t|e(t)|dtovershoot_pct: floatsteady_error_pct: floatsettling_time_sec: Optional[float]oscillation_cycles: intpv_mean: floatpv_std: floatop_mean: floatop_std: floatdata_points: intdef to_row(self) - Dict[str, str]:转为CSV行return {回路编号: self.loop_id,健康等级: f{self.grade.icon} {self.grade.label},评级原因: self.grade_reason,ISE: f{self.ise:.4f},IAE: f{self.iae:.4f},ITAE: f{self.itae:.4f},超调量(%): f{self.overshoot_pct:.2f},稳态误差(%): f{self.steady_error_pct:.4f},调节时间(s): f{self.settling_time_sec:.1f} if self.settling_time_sec else N/A,振荡周期数: str(self.oscillation_cycles),PV均值: f{self.pv_mean:.3f},PV标准差: f{self.pv_std:.4f},OP均值: f{self.op_mean:.3f},OP标准差: f{self.op_std:.4f},数据点数: str(self.data_points),}# # 2. 数据加载器(单一职责)# class ControllerDataLoader:加载并校验DCS/SCADA导出的控制器运行数据期望CSV格式(首行表头):timestamp,pv,sp,op2025-01-15 08:00:00,70.1,70.0,50.2def __init__(self, file_path: str, loops: List[LoopConfig]):self.file_path file_pathself.loops {lp.loop_id: lp for lp in loops}logging.info(f[DataLoader] 初始化, 数据文件: {file_path})def load_all(self) - Dict[str, List[Dict[str, float]]]:加载所有回路数据, 返回 {loop_id: [rows]}if not os.path.exists(self.file_path):raise FileNotFoundError(f数据文件不存在: {self.file_path})raw self._read_csv()result: Dict[str, List[Dict[str, float]]] {}for loop in self.loops.values():result[loop.loop_id] self._extract_loop(raw, loop)logging.info(f[DataLoader] 回路 {loop.loop_id} 加载 f{len(result[loop.loop_id])} 条记录)return resultdef _read_csv(self) - List[Dict[str, str]]:with open(self.file_path, r, encodingutf-8-sig) as f:reader csv.DictReader(f)return list(reader)def _extract_loop(self, raw: List[Dict[str, str]],loop: LoopConfig) - List[Dict[str, float]]:提取单个回路的 PV/SP/OP 时间序列, 做基础校验out: List[Dict[str, float]] []for row in raw:try:pv float(row.get(loop.pv_column, ).strip())sp float(row.get(loop.sp_column, ).strip())op float(row.get(loop.op_column, ).strip())out.append({pv: pv, sp: sp, op: op})except (ValueError, AttributeError):# 跳过坏行(通讯错误整条剔除)continueif len(out) 10:logging.warning(f[DataLoader] 回路 {loop.loop_id} 有效数据过少f({len(out)}条), 评估结果可能不可靠)return out# # 3. 指标计算器(单一职责 策略模式)# class MetricsCalculator:计算单回路控制性能指标核心公式(参考哈工程《工业过程控制》第5-6章):ISE Σ e(t)² · ΔtIAE Σ |e(t)| · ΔtITAE Σ t·|e(t)| · Δtσ% (PV_max - SP) / SP × 100%def __init__(self, thresholds: AssessmentThresholds AssessmentThresholds()):self.th thresholdslogging.info([Metrics] 指标计算器初始化)def evaluate(self, loop: LoopConfig,data: List[Dict[str, float]]) - LoopMetrics:对单个回路执行完整评估n len(data)dt loop.sample_interval_sec# 1. 误差序列 e SP - PVerrors [d[sp] - d[pv] for d in data]pvs [d[pv] for d in data]ops [d[op] for d in data]times [i * dt for i in range(n)]# 2. 三大积分指标ise sum(e * e * dt for e in errors)iae sum(abs(e) * dt for e in errors)itae sum(t * abs(e) * dt for t, e in zip(times, errors))# 3. 超调量(基于归一化设定值)sp_nominal loop.setpoint_nominalif sp_nominal ! 0:pv_max max(pvs)overshoot (pv_max - sp_nominal) / sp_nominal * 100.0overshoot max(0.0, overshoot)else:overshoot 0.0# 4. 稳态误差(取最后20%数据的平均偏差)tail_n max(10, n // 5)tail_errors errors[-tail_n:]steady_err (mean(tail_errors) / sp_nominal * 100.0) if sp_nominal else 0.0steady_err abs(steady_err)# 5. 调节时间(进入±2%带不再跑出)settling_time self._calc_settling_time(errors, sp_nominal, dt, tol_pct2.0)# 6. 振荡检测(过零次数 → 半周期数)zero_crossings sum(1 for i in range(1, n)if errors[i - 1] * errors[i] 0)osc_cycles zero_crossings // 2# 7. 综合评级grade, reason self._grade(overshoot, steady_err, osc_cycles, errors, sp_nominal)return LoopMetrics(loop_idloop.loop_id,gradegrade,grade_reasonreason,iseise, iaeiae, itaeitae,overshoot_pctovershoot,steady_error_pctsteady_err,settling_time_secsettling_time,oscillation_cyclesosc_cycles,pv_meanmean(pvs),pv_stdstdev(pvs) if n 1 else 0.0,op_meanmean(ops),op_stdstdev(ops) if n 1 else 0.0,data_pointsn,)# ---------- 内部方法 ----------def _calc_settling_time(self, errors: List[float],sp: float, dt: float,tol_pct: float 2.0) - Optional[float]:计算调节时间: 首次进入±tol%带后不再越出if sp 0:tol tol_pct / 100.0else:tol abs(sp) * tol_pct / 100.0entered Falsefor i, e in enumerate(errors):if abs(e) tol:if not entered:entered Truefirst_in ielse:entered False # 跑出容差带, 重置if entered:return first_in * dtreturn None # 未进入稳态def _grade(self, overshoot: float, steady_err: float,osc_cycles: int, errors: List[float],sp: float) - Tuple[LoopHealthGrade, str]:综合评级reasons: List[str] []grade LoopHealthGrade.EXCELLENT# 超调降级if overshoot self.th.overshoot_fair:grade LoopHealthGrade.POORreasons.append(f超调过大({overshoot:.1f}%{self.th.overshoot_fair}%))elif overshoot self.th.overshoot_good:if grade ! LoopHealthGrade.POOR:grade LoopHealthGrade.FAIRreasons.append(f超调偏高({overshoot:.1f}%{self.th.overshoot_good}%))# 稳态误差降级if steady_err self.th.steady_error_fair:grade LoopHealthGrade.POORreasons.append(f稳态误差大({steady_err:.2f}%{self.th.steady_error_fair}%))elif steady_err self.th.steady_error_good:if grade LoopHealthGrade.EXCELLENT:grade LoopHealthGrade.GOODreasons.append(f稳态误差偏高({steady_err:.2f}%{self.th.steady_error_good}%))# 振荡降级if osc_cycles self.th.oscillation_min_cycles:if grade LoopHealthGrade.EXCELLENT:grade LoopHealthGrade.GOODreasons.append(f存在持续振荡({osc_cycles}周期))# 积分饱和检测(误差长期不归零)n len(errors)tail errors[-max(10, n // 4):]if sp ! 0 and mean([abs(e) for e in tail]) / sp 0.05:if grade LoopHealthGrade.EXCELLENT:grade LoopHealthGrade.GOODreasons.append(误差长期未归零(疑似积分饱和))if not reasons:reasons.append(各指标均在优秀范围内)return grade, ; .join(reasons)# # 4. 报告生成器(模板方法)# class ReportGenerator:将评估结果格式化为每日控制效果评估文档支持 CSV(便于归档/Excel打开) 与 TXT(便于邮件正文/值班记录)def __init__(self, output_dir: str output,fmt: ReportFormat ReportFormat.BOTH):self.output_dir output_dirself.fmt fmtos.makedirs(output_dir, exist_okTrue)logging.info(f[Report] 报告生成器初始化, 输出目录: {output_dir})def generate(self, metrics_list: List[LoopMetrics],generation_time: datetime.datetime) - Dict[str, str]:生成报告, 返回生成的文件路径字典written: Dict[str, str] {}if self.fmt in (ReportFormat.CSV, ReportFormat.BOTH):csv_path self._write_csv(metrics_list, generation_time)written[csv] csv_pathif self.fmt in (ReportFormat.TXT, ReportFormat.BOTH):txt_path self._write_txt(metrics_list, generation_time)written[txt] txt_pathreturn written# ---------- 内部方法 ----------def _write_csv(self, metrics: List[LoopMetrics],gen_time: datetime.datetime) - str:path os.path.join(self.output_dir,fpid_daily_report_{gen_time.strftime(%Y%m%d)}.csv)# 写入UTF-8-BOM, 让Excel直接打开中文不乱码with open(path, w, encodingutf-8-sig, newline) as f:writer csv.DictWriter(f, fieldnameslist(metrics[0].to_row().keys()))writer.writeheader()for m in metrics:writer.writerow(m.to_row())logging.info(f[Report] CSV报告已生成: {path})return pathdef _write_txt(self, metrics: List[LoopMetrics],gen_time: datetime.datetime) - str:path os.path.join(self.output_dir,fpid_daily_report_{gen_time.strftime(%Y%m%d)}.txt)lines: List[str] []lines.append( * 68)lines.append( 控制器每日运行效果评估文档)lines.append(f 生成时间: {gen_time.strftime(%Y-%m-%d %H:%M:%S)})lines.append(f 评估回路数: {len(metrics)})lines.append( * 68)lines.append()# 汇总表lines.append(【汇总】)header (f{回路:10}{等级:8}{超调%:8}f{稳差%:10}{ISE:14}{ITAE:14})lines.append(header)lines.append(- * len(header))for m in metrics:lines.append(f{m.loop_id:10}{m.grade.icon}{m.grade.label:6}f{m.overshoot_pct:8.2f}{m.steady_error_pct:10.4f}f{m.ise:14.4f}{m.itae:14.4f})lines.append()# 各回路明细for m in metrics:lines.append(- * 68)lines.append(f回路: {m.loop_id} | 等级: {m.grade.icon} {m.grade.label})lines.append(f 评级原因: {m.grade_reason})lines.append(f ISE{m.ise:.4f} IAE{m.iae:.4f} ITAE利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛