储罐联锁逻辑仿真 —— 液位超高/过低安全保护系统那年半夜A3罐液位计突然卡死在60%实际液位悄悄爬到98%进料阀还开着。幸亏SIS系统的联锁在99%硬切断否则就是一场漫溢事故。那一刻我才明白安全联锁不是万一的保险而是必须的底线。—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸一、实际应用场景描述在石化仓储、液化气站、精细化工等场景储罐液位失控是最致命的安全隐患之一┌──────────────────────────────────────────────┐│ 储罐联锁保护系统SIS ││ ││ ┌────────────────────────────┐ ││ │ 液位变送器 (LT) │ ││ │ • 雷达/伺服/差压式 │ ││ │ • 4-20mA 输出 │ ││ │ • 两选一/三选二 冗余 │ ││ └────────────┬───────────────┘ ││ │ PV (0~100%) ││ ▼ ││ ┌────────────────────────────┐ ││ │ 联锁逻辑处理器 (SIS) │ ││ │ │ ││ │ ┌───────────────────┐ │ ││ │ │ 超高联锁 (HH) │ │ ││ │ │ if PV ≥ 95%: │ │ ││ │ │ CLOSE FV-101 │ │ ││ │ │ ALARM HIHI │ │ ││ │ └───────────────────┘ │ ││ │ │ ││ │ ┌───────────────────┐ │ ││ │ │ 高低联锁 (LL) │ │ ││ │ │ if PV ≤ 10%: │ │ ││ │ │ STOP P-201 │ │ ││ │ │ ALARM LOLO │ │ ││ │ └───────────────────┘ │ ││ │ │ ││ │ ┌───────────────────┐ │ ││ │ │ 中间报警 (H/L) │ │ ││ │ │ 80% 预警, 20% 预警 │ │ ││ │ └───────────────────┘ │ ││ │ │ ││ │ • 表决逻辑 (2oo3) │ │ ││ │ • 去抖动/死区 │ │ ││ │ • 联锁旁路管理 │ │ ││ └────────────┬───────────────┘ ││ │ DO信号 ││ ▼ ││ ┌────────────────────────────┐ ││ │ 执行机构 │ ││ │ • FV-101 进料调节阀 │ ││ │ • P-201 出料泵 (MCC) │ ││ │ • XV-301 紧急切断阀 │ ││ └───────────────────────────┘ ││ ││ 核心: 独立保护层 表决冗余 失效安全 全行程测试 │└──────────────────────────────────────────────┘DCS 逻辑 vs SIS 独立联锁维度 DCS 软逻辑 SIS 独立联锁独立性 ❌ 与控制系统共用硬件 ✅ 独立硬件独立电源可靠性 ❌ MTBF 较低 ✅ SIL2/3 认证响应时间 ❌ 扫描周期 100ms ✅ ≤50ms 硬切断旁路管理 ❌ 容易误操作 ✅ 钥匙锁审计追踪测试验证 ❌ 无系统测试 ✅ 定期全行程测试故障模式 ❌ 可能输出危险值 ✅ Fail-Safe 设计二、引入痛点2.1 现场的真实困境场景 现场发生了什么 根因漫溢事故 A3罐满了阀还开着喷出原料 液位计卡死无独立联锁抽空事故 泵空转2小时机封烧毁 低液位未联锁停泵假液位 变送器故障显示60%实际98% 单点测量无冗余表决联锁旁路 检修后忘恢复联锁一直旁路 旁路管理不规范DCS死机 操作站蓝屏阀门全失控 控制与联锁未分离2.2 核心矛盾安全联锁的本质不是让生产更顺畅而是在一切失控时保证不伤人、不毁设备、不污染环境。DCS 是优化生产的工具SIS 是保命的底线——两者必须物理隔离、逻辑独立、各司其职。2.3 我们要解决什么用一段精简的 Python 程序构建一个 储罐联锁逻辑仿真系统实现1. 多级液位报警 —— LO/LOLO/HI/HIHI 四段式2. 双向联锁 —— 超高关进料阀过低停出料泵3. 冗余表决 —— 2oo3 三取二逻辑4. 失效安全 —— 信号丢失→安全侧动作5. 旁路管理 —— 受控旁路超时自动恢复6. 事件记录 —— 完整审计追踪三、核心逻辑讲解3.1 理论基础安全仪表系统SIS本工具基于哈工程《工业过程控制》第十四章安全仪表系统和第十五章系统可靠性① 保护层概念Layers of Protection┌─────────────────────────────────────┐│ 第1层: 工艺设计余量安全系数 │├─────────────────────────────────────┤│ 第2层: 基本过程控制系统BPCS/DCS │├─────────────────────────────────────┤│ 第3层: 安全仪表系统SIS⭐本工具 │├─────────────────────────────────────┤│ 第4层: 物理保护安全阀/爆破片 │├─────────────────────────────────────┤│ 第5层: 厂区应急围堰/消防 │└─────────────────────────────────────┘② 表决逻辑Voting Logic1oo1: 单点最不可靠2oo2: 两点中有任一点触发即动作高可用低安全性2oo3: 三取二兼顾可用性与安全性⭐推荐1oo2: 两点都触发才动作低可用高安全性③ SIL 等级与 PFDSIL等级 PFD范围 要求SIL1 10⁻² ~ 10⁻¹ 基本联锁SIL2 10⁻³ ~ 10⁻² 中等风险SIL3 10⁻⁴ ~ 10⁻³ 高风险SIL4 10⁻⁵ ~ 10⁻⁴ 极高风险核/航天④ 失效安全Fail-Safe原则信号丢失 → 阀门关闭弹簧复位电源丢失 → 泵停止运转通信中断 → 保持安全状态CPU故障 → 看门狗触发安全输出3.2 联锁逻辑架构┌─────────────┐│ 3个液位变送器 ││ LT-101A/B/C │└──────┬──────┘│ 4-20mA × 3┌─────────▼─────────┐│ 信号调理 合理性检查 ││ • 量程转换 ││ • 偏差检测(5%报警) ││ • 信号丢失检测 │└─────────┬─────────┘│ PV_A, PV_B, PV_C┌─────────▼─────────┐│ 2oo3 表决逻辑 ││ 多数表决 → PV_valid ││ 全故障 → 安全侧 │└─────────┬─────────┘│ PV_effective┌─────────▼─────────┐│ 联锁逻辑判断 ││ ┌─────────────┐ ││ │ PV≥95% → HH │ ││ │ PV≥80% → HI │ ││ │ PV≤20% → LO │ ││ │ PV≤10% → LL │ ││ └─────────────┘ ││ 旁路状态检查 ││ 去抖动延时 │└─────────┬─────────┘│ 联锁动作指令┌─────────▼─────────┐│ 输出处理 ││ • 超高→关进料阀 ││ • 过低→停出料泵 ││ • 双通道冗余输出 │└───────────────────┘四、代码讲解面向对象设计4.1 类结构总览类名 职责 设计模式TankConfig 储罐配置dataclass 值对象InterlockLevels 联锁阈值dataclass 值对象AnalogSignal 模拟量信号含故障注入 策略模式LevelTransmitter 液位变送器含冗余 装饰器模式VotingLogic 表决逻辑2oo3/1oo2 策略模式InterlockLogic 联锁逻辑处理器 模板方法FinalElement 最终执行元件 状态模式BypassManager 旁路管理器 单例模式EventLogger 事件记录器 观察者模式TankSIS SIS 聚合根 聚合根TankSimulator 储罐过程模型 领域模型SISVisualizer 可视化工具 封装4.2 核心代码完整可运行完整源码约 400 行包含 12 个类、2oo3 表决、多级联锁、旁路管理、事件记录。以下为精简核心版可直接复制运行。detailssummary 完整源码点击展开/折叠/summary储罐联锁逻辑仿真 —— 液位超高/过低安全保护参考哈尔滨工程大学《工业过程控制》第十四章安全仪表系统from dataclasses import dataclass, fieldfrom typing import List, Tuple, Optional, Dict, Deque, Callablefrom enum import Enum, autoimport numpy as npimport matplotlib.pyplot as pltfrom collections import dequeimport timeimport mathfrom datetime import datetimeimport hashlib# # 1. 基础数据结构值对象# dataclassclass InterlockLevels:联锁阈值配置 —— 值对象low_low: float 10.0 # LL - 停泵阈值 (%)low: float 20.0 # LO - 低液位预警 (%)high: float 80.0 # HI - 高液位预警 (%)high_high: float 95.0 # HH - 关阀阈值 (%)hysteresis: float 2.0 # 滞环宽度 (%)def validate(self):验证阈值合理性assert self.low_low self.low self.high self.high_highassert self.hysteresis 0dataclassclass TankConfig:储罐配置 —— 值对象tank_id: str T-101capacity: float 5000.0 # 容积 (m³)diameter: float 15.0 # 直径 (m)max_level: float 12.0 # 最高液位 (m)initial_level: float 5.0 # 初始液位 (m)inflow_nominal: float 50.0 # 额定进料流量 (m³/h)outflow_nominal: float 40.0 # 额定出料流量 (m³/h)cross_section: float field(initFalse)def __post_init__(self):self.cross_section math.pi * (self.diameter / 2) ** 2dataclassclass AlarmState:报警状态 —— 值对象hihi: bool False # 高高液位hi: bool False # 高液位lo: bool False # 低液位lolo: bool False # 低低液位sensor_fault: bool False # 传感器故障vote_fault: bool False # 表决故障# # 2. 信号模型策略模式# class AnalogSignal:模拟量信号 —— 支持故障注入def __init__(self, name: str, nominal_value: float 50.0):self.name nameself.nominal nominal_valueself.current_value nominal_valueself.fault_type: Optional[str] None # stuck, drift, lost, noiseself.fault_param: float 0.0self.noise_std: float 0.5 # 正常噪声 (%)self.random_gen np.random.RandomState(hash(name) % 2**32)def set_fault(self, fault_type: str, param: float 0.0):注入故障self.fault_type fault_typeself.fault_param paramdef clear_fault(self):清除故障self.fault_type Noneself.fault_param 0.0def get_value(self, true_value: float) - float:获取当前信号值if self.fault_type stuck:# 卡死在某个值return self.fault_paramelif self.fault_type drift:# 漂移self.current_value self.fault_param * 0.1return self.current_valueelif self.fault_type lost:# 信号丢失返回4mA对应值return -10.0 # 低于量程下限elif self.fault_type noise:# 异常噪声return true_value self.random_gen.normal(0, self.fault_param)else:# 正常状态 小噪声noise self.random_gen.normal(0, self.noise_std)return true_value noisedef is_signal_lost(self, value: float) - bool:判断信号是否丢失return value -5.0 # 对应3.2mA# # 3. 液位变送器装饰器模式# class LevelTransmitter:液位变送器 —— 含冗余和合理性检查def __init__(self, transmitter_id: str, signal: AnalogSignal):self.tid transmitter_idself.signal signalself.last_value: float 50.0self.deviation_alarm: bool Falsedef read(self, true_level_percent: float) - float:读取液位值raw self.signal.get_value(true_level_percent)self.last_value raw# 合理性检查与上次值偏差过大if abs(raw - self.last_value) 20.0: # 单次跳变20%self.deviation_alarm Trueelse:self.deviation_alarm Falsereturn rawdef is_healthy(self, value: float) - bool:检查变送器健康状态if self.signal.is_signal_lost(value):return Falseif self.deviation_alarm:return Falsereturn True# # 4. 表决逻辑策略模式# class VotingLogic:2oo3 表决逻辑staticmethoddef vote_2oo3(values: List[float], signal_lost_flags: List[bool]) - Tuple[float, bool]:三取二表决返回: (表决后的有效值, 是否有效)valid_values []valid_indices []for i, (val, lost) in enumerate(zip(values, signal_lost_flags)):if not lost and -10.0 val 110.0:valid_values.append(val)valid_indices.append(i)# 少于2个有效信号 → 安全侧if len(valid_values) 2:return 50.0, False # 返回中间值标记无效# 2oo3: 取中值最鲁棒if len(valid_values) 3:median_val sorted(valid_values)[1]return median_val, Trueelif len(valid_values) 2:# 2个有效取平均return np.mean(valid_values), Truereturn valid_values[0], Truestaticmethoddef detect_disagreement(values: List[float], threshold: float 5.0) - bool:检测变送器间偏差过大valid [v for v in values if -10.0 v 110.0]if len(valid) 2:return Falsereturn (max(valid) - min(valid)) threshold# # 5. 联锁逻辑处理器模板方法# class InterlockLogic:联锁逻辑处理器 —— 模板方法def __init__(self, levels: InterlockLevels, tank_id: str T-101):self.levels levelsself.tank_id tank_idself.alarm AlarmState()self.trip_history: Deque[Tuple[float, str, str]] deque(maxlen100)# 去抖动延时self.hh_dwell_time: float 0.0self.ll_dwell_time: float 0.0self.debounce_time: float 2.0 # 2秒去抖self.tripped_hh: bool Falseself.tripped_ll: bool False# 回调self.on_trip_hh: Optional[Callable] Noneself.on_trip_ll: Optional[Callable] Noneself.on_alarm: Optional[Callable] Nonedef evaluate(self, pv: float, dt: float, bypass_active: bool False) - Dict[str, bool]:评估联锁条件actions {close_inlet: False,stop_outlet_pump: False,alarm_hihi: False,alarm_hi: False,alarm_lo: False,alarm_lolo: False}if bypass_active:return actions # 旁路激活不执行联锁# ---- 高液位逻辑 ----# HI 报警带滞环if pv self.levels.high:actions[alarm_hi] Trueself.alarm.hi Trueelif pv self.levels.high - self.levels.hysteresis:self.alarm.hi False# HH 联锁去抖动if pv self.levels.high_high:self.hh_dwell_time dtif self.hh_dwell_time self.debounce_time:actions[close_inlet] Trueself.tripped_hh Trueself._record_trip(pv, HH_TRIP, 关闭进料阀)if self.on_trip_hh:self.on_trip_hh(pv)else:self.hh_dwell_time 0.0# 复位条件if pv self.levels.high - self.levels.hysteresis:self.tripped_hh False# ---- 低液位逻辑 ----# LO 报警if pv self.levels.low:actions[alarm_lo] Trueself.alarm.lo Trueelif pv self.levels.low self.levels.hysteresis:self.alarm.lo False# LL 联锁if pv self.levels.low_low:self.ll_dwell_time dtif self.ll_dwell_time self.debounce_time:actions[stop_outlet_pump] Trueself.tripped_ll Trueself._record_trip(pv, LL_TRIP, 停止出料泵)if self.on_trip_ll:self.on_trip_ll(pv)else:self.ll_dwell_time 0.0# 复位条件if pv self.levels.low self.levels.hysteresis:self.tripped_ll Falsereturn actionsdef _record_trip(self, pv: float, trip_type: str, action: str):记录联锁动作entry (time.time(), trip_type, fPV{pv:.1f}% - {action})self.trip_history.append(entry)def get_trip_summary(self) - str:获取联锁动作摘要if not self.trip_history:return 无联锁动作记录lines [f {self.tank_id} 联锁动作记录 ]for ts, trip_type, detail in self.trip_history:t_str datetime.fromtimestamp(ts).strftime(%Y-%m-%d %H:%M:%S)lines.append(f [{t_str}] {trip_type}: {detail})return \n.join(lines)# # 6. 最终执行元件状态模式# class FinalElement:最终执行元件 —— 状态模式def __init__(self, element_id: str, element_type: str VALVE):self.eid element_idself.etype element_type # VALVE / PUMPself.state: str NORMAL # NORMAL / TRIPPED / FAULTself.position: float 100.0 # 阀门开度 0~100%self.running: bool True # 泵运行状态self.fail_safe: str CLOSE if element_type VALVE else STOPdef close_valve(self) - str:关闭阀门安全动作self.state TRIPPEDself.position 0.0return f {self.eid}: 阀门关闭 (安全动作)def open_valve(self) - str:打开阀门正常动作if self.state NORMAL:self.position 100.0return f {self.eid}: 阀门打开return f ⚠️ {self.eid}: 阀门联锁锁定无法打开def stop_pump(self) - str:停止泵安全动作self.state TRIPPEDself.running Falsereturn f {self.eid}: 泵已停止 (安全动作)def start_pump(self) - str:启动泵if self.state NORMAL:self.running Truereturn f ▶️ {self.eid}: 泵已启动return f ⚠️ {self.eid}: 泵联锁锁定无法启动def reset(self) - str:复位到正常状态self.state NORMALif self.etype VALVE:self.position 100.0else:self.running Truereturn f {self.eid}: 已复位到正常状态# # 7. 旁路管理器单例模式# class BypassManager:旁路管理器 —— 单例模式_instance None_lock Truedef __new__(cls):if cls._instance is None:cls._instance super().__new__(cls)cls._instance._init()return cls._instancedef _init(self):self.bypasses: Dict[str, Dict] {} # {tank_id: {active, reason, operator, expire_time}}self.audit_log: Deque[Tuple[float, str, str]] deque(maxlen500)def activate(self, tank_id: str, reason: str, operator: str,duration_minutes: float 30.0) - str:激活旁路import timeexpire time.time() duration_minutes * 60self.bypasses[tank_id] {active: True,reason: reason,operator: operator,expire_time: expire,activated_at: time.time()}entry (time.time(), operator, f旁路激活[{tank_id}]: {reason} (有效期{duration_minutes}分钟))self.audit_log.append(entry)return f ⚠️ 旁路已激活: {tank_id} | 原因: {reason} | 操作人: {operator}def deactivate(self, tank_id: str, operator: str) - str:解除旁路if tank_id in self.bypasses:del self.bypasses[tank_id]entry (time.time(), operator, f旁路解除[{tank_id}] | 操作人: {operator})self.audit_log.append(entry)return f ✅ 旁路已解除: {tank_id} | 操作人: {operator}def is_bypassed(self, tank_id: str) - bool:检查是否旁路import timeif tank_id not in self.bypasses:return Falsebp self.bypasses[tank_id]if time.time() bp[expire_time]:# 自动过期del self.bypasses[tank_id]entry (time.time(), SYSTEM, f旁路自动过期[{tank_id}])self.audit_log.append(entry)return Falsereturn bp[active]def get_status(self) - str:获取旁路状态if not self.bypasses:return 无旁路激活lines [ 旁路状态 ]for tid, bp in self.bypasses.items():remaining (bp[expire_time] - time.time()) / 60lines.append(f {tid}: 激活 | 剩余{remaining:.0f}分钟 | {bp[reason]})return \n.join(lines)def get_audit_log(self, last_n: int 10) - List[str]:获取审计日志recent list(self.audit_log)[-last_n:]return [f [{datetime.fromtimestamp(ts).strftime(%H:%M:%S)}] {op}: {msg}for ts, op, msg in recent]# # 8. 事件记录器观察者模式# class EventLogger:事件记录器 —— 观察者模式def __init__(self, max_events: int 1000):self.events: Deque[Dict] deque(maxlenmax_events)self.subscribers: List[Callable] []def log(self, event_type: str, message: str, severity: str INFO,metadata: Dict None):记录事件event {timestamp: time.time(),type: event_type,message: message,severity: severity,metadata: metadata or {}}self.events.append(event)# 通知订阅者for sub in self.subscribers:sub(event)# 控制台输出ts datetime.fr利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛