feat(#72): 预警规则与阈值设定(声明式AlertRule三级severity+SOP联动,PRD 5.3 ③ 场景A)

把行业知识——预警分级(P0/P1/P2)+阈值+处置SOP——外置为模板配置(PRD line 152/171
阈值外置,行业工程师配置台维护),规则引擎零改动。

- alert_rules.py:AlertSeverity(P0/P1/P2) + AlertCondition(6 运算) + AlertRule(AND 语义)
  + AlertRuleEngine(按 severity 降序 + primary_alert) + 零依赖 YAML 解析(flow map)。
- config/alert_rules.template.yaml:6 条规则,特征名对齐 #70 FeatureSpec.name,sop 引用
  异常处置 SOP(供 #11 LLM 报警解释 + 驾驶舱红色告警 + 值班长确认,PRD 场景A)。
- tests/test_alert_rules.py:16 项单测(运算/规则/引擎排序/配置/端到端P0)全通过。
- _sanity_check_rules.py:冒烟(正常无告警 + 急升温P0)。

PRD line 333:关键告警不直接联动执行机构,高利害输出人工确认 → severity 分级支撑。
This commit is contained in:
2026-08-05 02:13:18 +08:00
parent f5d2294ee4
commit 4dcaf47af6
6 changed files with 733 additions and 0 deletions
@@ -0,0 +1,59 @@
# -*- coding: utf-8 -*-
"""炉层杂质预警规则引擎冒烟脚本(Issue #72)。
直接运行验证:模板规则资产可加载、规则引擎对急升温特征向量产出 P0 告警、
无异常时不告警。零第三方依赖。
"""
import importlib.util
import os
import sys
_PKG_DIR = os.path.dirname(os.path.abspath(__file__))
def _load_pkg(name, path):
if name in sys.modules:
return
spec = importlib.util.spec_from_file_location(
name, os.path.join(path, "__init__.py"),
submodule_search_locations=[path])
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
_load_pkg("impurity_forecast", _PKG_DIR)
from impurity_forecast import AlertRuleEngine, AlertSeverity # noqa: E402
CONFIG = os.path.join(_PKG_DIR, "config", "alert_rules.template.yaml")
def main() -> int:
eng = AlertRuleEngine.from_template_config(CONFIG)
print(f"[OK] 规则配置加载: {len(eng.rules)} 条预警规则")
for r in eng.rules:
print(f" {r.describe()} (sop={r.sop or '-'})")
# 1) 正常工况不告警
normal = eng.evaluate(1, {"炉温_ema5": 850.0, "炉温_rate10": 0.01,
"氯气流量_std10": 3.0, "炉压_rate10": 0.01,
"炉层状态_mean10": 60.0})
assert not normal, "正常工况不应告警"
print("[OK] 正常工况:无告警")
# 2) 急升温 + 炉压急变 → P0 红色告警(PRD 场景A)
abnormal = eng.evaluate(2, {"炉温_ema5": 905.0, "炉温_rate10": 0.06,
"氯气流量_std10": 4.0, "炉压_rate10": 0.09,
"炉层状态_mean10": 60.0})
assert abnormal, "异常工况应触发告警"
primary = eng.primary_alert(abnormal)
assert primary.severity == AlertSeverity.P0, "主告警应为 P0"
print(f"[OK] 异常工况:主告警 {primary.severity.value}({primary.message})"
f" sop={primary.sop}")
print("炉层杂质预警规则引擎冒烟通过 ✅")
return 0
if __name__ == "__main__":
raise SystemExit(main())