简介pyTEP 是面向过程控制、化工仿真与机器学习研究者的开源 Python 仿真 API用于田纳西伊士曼Tennessee Eastman流程的交互式模拟。它通过高级接口简化复杂工况的设置用户无需深入理解底层模拟器原理即可调整操作条件、运行仿真并保存数据适合控制算法验证、故障诊断与数据驱动建模等场景。资源包共 76 个文件约 754KB包含 16 个 Python 源码文件、23 个 MATLAB 的 m 文件、8 个 pkl 数据文件以及 mdl、mat、rst、mexw64/mexa64 等混合语言与平台组件另有 README、LICENSE、Makefile、pyproject.toml 等工程配置便于二次编译与跨语言调用。目前已有 532 人学习下载。借助该包读者可快速搭建 TEP 仿真环境复现标准测试工况获取仿真输出数据并参考 MATLAB 桥接与引擎工具代码理解接口设计为控制策略对比和模型训练提供可复用的实验基础。1. 从一次控制回路调参翻车说起pyTEP 到底解决了什么如果你做过过程控制、故障诊断或者强化学习方向的课题大概率绕不开 Tennessee Eastman 流程。这个由美国化工企业提出的标准仿真模型包含反应器、冷凝器、汽提塔、压缩机等单元几十个变量互相耦合还自带 20 多种预设故障工况。问题在于很多公开实现要么是编译好的二进制黑匣子要么是散落在论文附录里的 Fortran 片段想改一个控制器参数、加一个自定义扰动往往要花几天时间啃源码。pyTEP 这个开源仿真 API 的价值就在这里它把 Tennessee Eastman 流程封装成 Python 可调用的接口用几行代码就能完成场景配置、闭环仿真和数据采集。我第一次接触它是在一个故障诊断课题里当时需要批量生成不同故障模式下的传感器时序数据用传统方式跑一次仿真要手动改配置文件、重新编译而 pyTEP 让我在 Jupyter 里直接循环调用效率提升非常明显。这篇文章会从环境搭建、最小仿真、参数配置、避坑经验到进阶用法把 pyTEP 的落地路径完整走一遍适合做过程控制、异常检测、强化学习环境搭建的从业者参考。2. 环境搭建与最小仿真把 pyTEP 跑起来的第一公里2.1 安装方式与依赖版本选择pyTEP 的安装本身不复杂但依赖版本是第一个容易翻车的地方。常见做法是通过 pip 从源码仓库安装或者直接克隆仓库后以开发模式安装。我一般会先建一个干净的虚拟环境避免和系统里已有的科学计算库冲突。# 创建并激活虚拟环境以 conda 为例 conda create -n pytep_env python3.9 -y conda activate pytep_env # 安装核心依赖版本尽量锁定避免 API 变动 pip install numpy1.24.3 scipy1.10.1 matplotlib3.7.1 # 从源码安装 pyTEP假设已克隆到本地目录 cd pytep_repo pip install -e .这里有几个参数值得说明。Python 版本建议 3.8 到 3.10 之间太新的版本可能遇到部分科学计算库尚未适配的问题。numpy 和 scipy 是 pyTEP 内部数值计算的基础版本差异可能导致仿真步长或积分器行为不一致所以建议在项目初期就固定版本并写入 requirements.txt。pip install -e .是以可编辑模式安装好处是你随时可以修改 pyTEP 源码里的模型参数而不需要重新安装。安装完成后用一行命令验证是否成功import pytep print(pytep.__version__)如果输出版本号且没有报错说明基础环境已经就绪。如果报ModuleNotFoundError优先检查虚拟环境是否激活、安装路径是否在当前 Python 的 sys.path 中。2.2 跑通第一个闭环仿真代码逐行拆解pyTEP 的核心使用模式是创建仿真对象、设置运行时长和采样周期、指定故障类型、启动仿真并采集数据。下面是一个最小可运行示例模拟正常工况下运行 10 小时每 0.1 小时采样一次。import numpy as np from pytep.tep import TEPSimulator # 初始化仿真器指定随机种子保证结果可复现 sim TEPSimulator(seed42) # 配置仿真参数 sim.set_simulation_time(10.0) # 总仿真时长单位小时 sim.set_sample_time(0.1) # 采样周期单位小时 sim.set_fault_type(0) # 0 表示无故障正常工况 # 启动仿真并获取数据 data sim.run() # data 是一个字典包含状态变量、测量变量和控制变量 Xmeas data[Xmeas] # 测量变量形状 (样本数, 41) Xmv data[Xmv] # 控制变量形状 (样本数, 12) time data[time] # 时间戳 print(f仿真完成共采集 {Xmeas.shape[0]} 个样本) print(f测量变量维度{Xmeas.shape[1]}控制变量维度{Xmv.shape[1]})这段代码的逻辑很直接TEPSimulator是入口类seed参数控制随机数生成器影响初始状态和噪声。set_simulation_time和set_sample_time共同决定样本数量比如 10 小时除以 0.1 小时等于 100 个采样点。set_fault_type(0)表示正常运行pyTEP 通常支持 1 到 20 多种预设故障具体编号需要查阅对应版本的文档或源码中的故障定义表。run()方法返回的字典里Xmeas是 41 维测量变量Xmv是 12 维控制变量这是 Tennessee Eastman 流程的标准输出格式。注意不同版本的 pyTEP 在变量命名和返回结构上可能有差异如果data[Xmeas]报 KeyError先打印data.keys()看看实际字段名。2.3 故障工况切换与数据采集策略做故障诊断或异常检测时需要批量生成不同故障类型的数据。pyTEP 的优势在这里体现得很明显你可以在一个循环里切换故障类型而不需要重启进程或重新编译。import pandas as pd all_data [] fault_list [0, 1, 2, 4, 8] # 选取部分故障编号0 为正常 for fault_id in fault_list: sim TEPSimulator(seed42) sim.set_simulation_time(20.0) sim.set_sample_time(0.1) sim.set_fault_type(fault_id) result sim.run() df pd.DataFrame(result[Xmeas]) df[fault_type] fault_id df[time] result[time] all_data.append(df) # 合并所有工况数据 final_df pd.concat(all_data, ignore_indexTrue) final_df.to_csv(tep_fault_dataset.csv, indexFalse) print(f数据集保存完成总样本数{len(final_df)})这段代码的关键参数是fault_list你需要根据具体研究目标选择故障类型。常见做法是先用正常工况数据训练基线模型再用故障数据做测试。set_simulation_time(20.0)比上一个例子更长是为了让故障效应充分显现因为 Tennessee Eastman 流程本身有较大的惯性故障引入后需要一段时间才能从测量变量上观察到明显偏移。保存为 CSV 时把fault_type和time一起写入方便后续做时序切分和标签对齐。如果要做强化学习环境可以把run()替换成逐步调用step()方法每步返回当前状态、奖励和终止标志。pyTEP 的交互式仿真选项就是为这类场景准备的具体接口名称需要参考源码中的step或advance方法定义。3. 参数配置与场景定制让仿真贴近你的课题需求3.1 采样周期与仿真步长的匹配关系采样周期和内部积分步长是两个容易混淆的参数。采样周期决定你多久记录一次数据而内部积分步长决定求解器多久推进一步。如果采样周期远大于积分步长数据会显得很平滑如果两者接近可能会观察到数值噪声。我一般会把采样周期设为积分步长的整数倍避免出现非均匀采样。sim TEPSimulator(seed42) sim.set_simulation_time(5.0) sim.set_sample_time(0.05) # 采样周期 0.05 小时 sim.set_integration_step(0.01) # 内部积分步长 0.01 小时 sim.set_fault_type(1) data sim.run()这里set_integration_step不是所有版本都暴露的接口如果你的版本没有这个方法说明内部步长是固定的此时只需要关注采样周期即可。采样周期过小会导致数据量膨胀比如 20 小时仿真、0.01 小时采样就是 2000 个样本41 个测量变量加 12 个控制变量单次仿真就是 10 万多个浮点数批量生成时要注意内存和磁盘占用。3.2 自定义扰动与控制器参数调整pyTEP 预设的故障类型覆盖了常见的阶跃、随机变化和阀门粘滞等场景但课题需求往往更特殊。比如你想模拟某个进料温度缓慢漂移或者让某个控制阀的响应变慢就需要修改模型参数或注入自定义扰动。常见做法是找到 pyTEP 源码中定义故障的函数照着它的模式写一个自定义扰动函数。下面是一个概念性示例展示如何通过修改仿真器内部状态来注入扰动import numpy as np from pytep.tep import TEPSimulator sim TEPSimulator(seed42) sim.set_simulation_time(10.0) sim.set_sample_time(0.1) # 假设仿真器提供了获取和设置内部状态的接口 # 具体方法名以实际源码为准 state sim.get_state() state[reactor_temperature] 5.0 # 反应器温度偏移 5 度 sim.set_state(state) data sim.run()这段代码的重点不是具体方法名而是思路先获取当前状态字典修改目标变量再写回去。不同版本的 pyTEP 在状态表示上差异较大有的用数组有的用字典有的直接暴露属性。你需要先阅读源码中get_state和set_state的实现确认键名和单位。如果找不到这些接口退而求其次的做法是修改故障定义表在预设故障的基础上调整幅值参数。控制器参数调整也是类似逻辑。Tennessee Eastman 流程默认使用分散 PID 控制如果你要测试模型预测控制或强化学习控制器需要先禁用默认控制器再接入自己的控制逻辑。pyTEP 的交互式模式允许你在每个采样时刻传入控制量具体做法是调用step(action)而不是run()其中action是 12 维控制变量数组。3.3 批量实验的参数扫描与结果管理做对比实验时往往需要扫描多个参数组合比如不同故障幅值、不同控制器增益、不同噪声水平。手动改代码效率太低我一般会写一个参数网格用循环批量跑并把每次实验的配置和结果一起存档。import itertools import json import pandas as pd from pytep.tep import TEPSimulator # 定义参数网格 fault_types [0, 1, 2] noise_levels [0.0, 0.01, 0.05] seeds [42, 123, 456] results [] for ft, nl, sd in itertools.product(fault_types, noise_levels, seeds): sim TEPSimulator(seedsd) sim.set_simulation_time(15.0) sim.set_sample_time(0.1) sim.set_fault_type(ft) # 假设有设置噪声水平的方法 if hasattr(sim, set_noise_level): sim.set_noise_level(nl) data sim.run() df pd.DataFrame(data[Xmeas]) df[fault_type] ft df[noise_level] nl df[seed] sd results.append(df) final pd.concat(results, ignore_indexTrue) final.to_csv(tep_grid_experiment.csv, indexFalse) # 保存实验配置方便复现 config { fault_types: fault_types, noise_levels: noise_levels, seeds: seeds, simulation_time: 15.0, sample_time: 0.1 } with open(experiment_config.json, w) as f: json.dump(config, f, indent2)这段代码的价值在于可复现性。itertools.product生成笛卡尔积确保所有组合都被覆盖。每次实验的seed、fault_type、noise_level都写入数据框后续分析时可以直接按这些列分组。配置文件单独保存半年后回头看也能知道当时跑了什么。如果实验规模很大建议把每个组合的结果单独存一个文件避免单个 CSV 过大导致读取缓慢。4. 避坑与排查pyTEP 使用中的五个血泪教训4.1 仿真结果每次都不一样随机种子没生效现象同样的代码跑两次测量变量曲线明显不同故障检测模型的评估指标波动很大。原因pyTEP 内部可能使用了全局随机数生成器或者seed参数没有传递到所有涉及随机过程的模块比如测量噪声、初始状态扰动。解决先确认TEPSimulator(seed42)中的 seed 是否被正确传递。如果无效尝试在导入 pyTEP 之前设置 numpy 的全局种子np.random.seed(42)并在每次创建仿真器之前重新设置。更稳妥的做法是检查源码中随机数的调用位置看是否使用了独立的RandomState实例。如果确实无法控制可以在论文或报告中说明随机性来源并增加重复实验次数取平均。4.2 故障工况数据没有明显异常模型学不到东西现象用故障类型 1 的数据训练异常检测模型验证集 AUC 只有 0.55几乎等于随机猜测。原因Tennessee Eastman 流程的某些故障本身就很微弱或者故障引入时间太晚仿真时长不够让异常传播到测量变量。另外如果采样周期太大可能把瞬态过程平滑掉了。解决先画图看看故障工况和正常工况的测量变量均值有没有差异。如果没有尝试延长仿真时间、减小采样周期、增大故障幅值。pyTEP 的故障定义表中不同故障的严重程度不同选故障类型时优先选那些在文献中被验证为可检测的编号。如果还是不行考虑用残差或变化率作为特征而不是原始测量值。4.3 内存溢出批量仿真跑到一半崩溃现象循环生成 50 组仿真数据时程序占用内存持续上升最终被系统杀掉。原因每次sim.run()返回的数据都追加到列表里没有及时释放。如果单次仿真样本数很大累积起来很容易超过内存上限。解决把结果直接写入磁盘而不是全部留在内存。可以用df.to_csv(part_{}.csv.format(i))分文件保存最后再合并。或者用生成器模式每次只处理一个工况的数据训练模型时用tf.data或torch.utils.data.DataLoader流式读取。另外del sim和gc.collect()在循环末尾调用也有帮助但效果不如直接落盘。4.4 版本升级后 API 变了旧代码跑不通现象之前能跑的脚本换了环境或升级 pyTEP 后报AttributeError或TypeError。原因开源项目在迭代过程中可能重命名方法、调整参数顺序、改变返回结构。pyTEP 作为研究型项目API 稳定性不如商业软件。解决在项目初期就锁定版本把pytepx.y.z写入 requirements.txt。如果必须升级先阅读变更日志或对比源码 diff。更保险的做法是写一层薄封装把 pyTEP 的调用集中在一个模块里升级时只需要改这一个文件。我一般会在封装层里做参数校验和默认值填充这样上层实验代码不受底层 API 变动影响。4.5 仿真速度太慢参数扫描跑不完现象单次 20 小时仿真耗时超过 30 秒扫描 100 组参数需要近一小时。原因Tennessee Eastman 流程本身包含刚性微分方程求解器步长受限。如果采样周期很小输出数据量大I/O 也会成为瓶颈。解决先确认是否真的需要那么长的仿真时间和那么小的采样周期。很多故障在 5 小时内就能充分显现采样周期 0.05 小时足够。如果必须长时间仿真可以尝试调整求解器容差但要注意数值精度。另外把数据采集和仿真计算分离仿真时只记录关键变量后续再插值或重采样。如果课题允许用并行计算跑参数扫描每个进程独立跑一组参数能大幅缩短总时间。5. 进阶用法把 pyTEP 接入强化学习与在线监测流程5.1 用 step 接口搭建 Gym 风格环境pyTEP 的交互式仿真选项最适合做强化学习环境。核心思路是把run()替换成逐步调用每个时间步传入控制量返回状态、奖励和终止信号。下面是一个简化的环境封装示例import numpy as np from pytep.tep import TEPSimulator class TEPEnv: def __init__(self, fault_type0, max_steps200): self.sim TEPSimulator(seed42) self.sim.set_simulation_time(20.0) self.sim.set_sample_time(0.1) self.sim.set_fault_type(fault_type) self.max_steps max_steps self.current_step 0 self.state None def reset(self): self.sim.reset() self.current_step 0 self.state self.sim.get_state() return self._extract_observation() def step(self, action): # action 是 12 维控制变量 self.sim.set_control(action) self.sim.advance() self.current_step 1 next_state self.sim.get_state() reward self._compute_reward(next_state) done self.current_step self.max_steps self.state next_state return self._extract_observation(), reward, done, {} def _extract_observation(self): # 从状态中提取测量变量作为观测 return np.array(self.state[Xmeas]) def _compute_reward(self, state): # 示例奖励偏离设定值越远惩罚越大 setpoint np.zeros(41) deviation np.linalg.norm(state[Xmeas] - setpoint) return -deviation这段代码的关键在于reset()、step()和_extract_observation()三个方法。reset()把仿真器恢复到初始状态step()接收动作、推进一步、计算奖励。奖励函数需要根据具体任务设计比如跟踪设定值、抑制故障影响、最小化控制量变化。_compute_reward里的setpoint应该替换成你实际关注的变量目标值。如果 pyTEP 没有reset()方法可以重新创建一个TEPSimulator实例来达到同样效果。5.2 在线监测中的滑动窗口与故障报警延迟把 pyTEP 用于在线监测算法验证时需要模拟真实场景中的滑动窗口和报警逻辑。常见做法是维护一个固定长度的缓冲区每次新样本到达后更新缓冲区然后用监测模型计算统计量超过阈值就报警。from collections import deque import numpy as np class OnlineMonitor: def __init__(self, window_size50, threshold3.0): self.window deque(maxlenwindow_size) self.threshold threshold self.baseline_mean None self.baseline_std None def fit_baseline(self, normal_data): self.baseline_mean np.mean(normal_data, axis0) self.baseline_std np.std(normal_data, axis0) 1e-8 def update(self, sample): self.window.append(sample) if len(self.window) self.window.maxlen: return False window_array np.array(self.window) # 计算窗口内均值与基线的偏离程度 window_mean np.mean(window_array, axis0) z_score np.abs((window_mean - self.baseline_mean) / self.baseline_std) alarm np.any(z_score self.threshold) return alarm这个监测器的参数需要根据实际数据调整。window_size决定报警延迟和平滑程度窗口越大报警越稳定但延迟越高。threshold控制灵敏度3.0 是常见的经验值但最好用正常工况数据做交叉验证来确定。fit_baseline用正常工况数据计算均值和标准差后续每个样本的窗口均值与基线比较超过阈值就触发报警。这套逻辑可以直接套用到 pyTEP 生成的故障数据上评估不同故障类型的检测延迟。5.3 从仿真到半实物把 pyTEP 数据接入外部控制器一个容易被忽略的进阶用法是把 pyTEP 当作被控对象外部控制器通过接口读写变量。比如你想测试自己写的模型预测控制算法可以让 pyTEP 每个采样周期暂停等待控制器计算完控制量后再继续。这种模式在 pyTEP 中通常通过回调函数或事件机制实现具体取决于版本。如果 pyTEP 不支持暂停/恢复退而求其次的做法是用step()手动推进把控制器的计算逻辑放在循环里。这样虽然不能严格模拟实时性但足以验证控制算法的稳态性能和抗扰能力。我一般会在循环里加一个计时器记录每次控制器计算耗时如果耗时超过采样周期就在报告中说明实时性约束不满足。5.4 一个值得养成的习惯先画图再建模最后分享一个我踩过多次坑之后养成的习惯拿到任何一组 pyTEP 仿真数据先画三张图——正常工况的测量变量时序图、故障工况的测量变量时序图、两者的均值差异热力图。这三张图能在五分钟内告诉你数据质量如何、故障是否可检测、哪些变量最敏感。很多建模失败的原因不是算法不行而是数据本身就没有区分度。先画图再建模能省下大量调参时间。希望帮到你。本文还有配套的精品资源点击获取