357 lines
13 KiB
Python
357 lines
13 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""炉层杂质预警 · 预警规则与阈值设定引擎（Issue #72 / PRD 5.3 ③）。
|
|||
|
|
|
|||
|
|
PRD 5.3 ③ / 场景 A（line 80）：异常检测模型触发 → 驾驶舱红色告警 + LLM 生成
|
|||
|
|
"原因+处置建议" → 值班长确认。本模块把"行业知识"——**预警分级 + 阈值 + 处置 SOP**——
|
|||
|
|
外置为模板配置（PRD line 152/171：阈值外置 JSON，行业工程师在配置台维护），引擎
|
|||
|
|
按规则评估特征向量产出带 severity 的 Alert。
|
|||
|
|
|
|||
|
|
设计要点
|
|||
|
|
--------
|
|||
|
|
1. **声明式 AlertRule**：每条规则声明 ``id`` + ``severity``（P0/P1/P2）+ ``condition``
|
|||
|
|
（特征名 + 比较运算 + 阈值）+ ``sop``（处置 SOP 引用，供 LLM 报警解释/驾驶舱展示）。
|
|||
|
|
2. **severity 三级**（PRD line 333：关键告警不直接联动执行机构，高利害人工确认）：
|
|||
|
|
- ``P0``（critical）：红色告警，立即人工确认 + 紧急处置；
|
|||
|
|
- ``P1``（warning）：黄色告警，加强监控 + 预备处置；
|
|||
|
|
- ``P2``（info）：提示，记录跟踪。
|
|||
|
|
3. **规则引擎**：``AlertRuleEngine.evaluate`` 对一个特征向量评估全部规则，返回命中的
|
|||
|
|
Alert 列表（取最高 severity 为主告警）；可与 #70/#71 组合（特征向量/异常分数均可作为
|
|||
|
|
condition 输入）。
|
|||
|
|
4. **零依赖 YAML 子集解析**（对齐 data-bus / rag-kb / #70），阈值外置模板资产。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import math
|
|||
|
|
import os
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
from enum import Enum
|
|||
|
|
from typing import Callable, Dict, List, Optional, Sequence, Tuple
|
|||
|
|
|
|||
|
|
NAN = float("nan")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class AlertSeverity(str, Enum):
|
|||
|
|
"""预警严重度三级（PRD line 80 红色告警 / line 333 关键告警人工确认）。"""
|
|||
|
|
|
|||
|
|
P0 = "P0" # critical：红色，立即人工确认 + 紧急处置
|
|||
|
|
P1 = "P1" # warning：黄色，加强监控 + 预备处置
|
|||
|
|
P2 = "P2" # info：提示，记录跟踪
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def label(self) -> str:
|
|||
|
|
return {AlertSeverity.P0: "严重", AlertSeverity.P1: "警告",
|
|||
|
|
AlertSeverity.P2: "提示"}[self]
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def rank(self) -> int:
|
|||
|
|
"""排序权重，越大越严重（用于取主告警）。"""
|
|||
|
|
return {AlertSeverity.P0: 3, AlertSeverity.P1: 2, AlertSeverity.P2: 1}[self]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 比较运算符注册表（condition.op 取值）
|
|||
|
|
OPS: Dict[str, Callable[[float, float], bool]] = {
|
|||
|
|
">": lambda a, b: a > b,
|
|||
|
|
">=": lambda a, b: a >= b,
|
|||
|
|
"<": lambda a, b: a < b,
|
|||
|
|
"<=": lambda a, b: a <= b,
|
|||
|
|
"==": lambda a, b: a == b,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _is_num(x: object) -> bool:
|
|||
|
|
return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x))
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class AlertCondition:
|
|||
|
|
"""单条触发条件：特征名 + 比较运算 + 阈值。"""
|
|||
|
|
|
|||
|
|
feature: str
|
|||
|
|
op: str
|
|||
|
|
threshold: float
|
|||
|
|
|
|||
|
|
def __post_init__(self) -> None:
|
|||
|
|
if self.op not in OPS:
|
|||
|
|
raise ValueError(f"未知比较运算 {self.op!r}（应为 {sorted(OPS)}）")
|
|||
|
|
|
|||
|
|
def matches(self, values: Dict[str, float]) -> bool:
|
|||
|
|
v = values.get(self.feature)
|
|||
|
|
if not _is_num(v):
|
|||
|
|
return False
|
|||
|
|
return OPS[self.op](float(v), self.threshold)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class AlertRule:
|
|||
|
|
"""声明式预警规则（模板配置中的一行规则声明）。
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
id: 规则 id（稳定标识，供驾驶舱/审计引用）。
|
|||
|
|
severity: 严重度（P0/P1/P2）。
|
|||
|
|
conditions: 触发条件列表（AND 语义：全部满足才命中）。
|
|||
|
|
message: 告警文案（驾驶舱展示）。
|
|||
|
|
sop: 处置 SOP 引用（PRD 场景A：LLM 报警解释 + 值班长确认）。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
id: str
|
|||
|
|
severity: AlertSeverity
|
|||
|
|
conditions: List[AlertCondition]
|
|||
|
|
message: str = ""
|
|||
|
|
sop: str = ""
|
|||
|
|
|
|||
|
|
def __post_init__(self) -> None:
|
|||
|
|
if not self.id:
|
|||
|
|
raise ValueError("AlertRule.id 不能为空")
|
|||
|
|
if not self.conditions:
|
|||
|
|
raise ValueError(f"规则 {self.id!r} 至少需要 1 条 condition")
|
|||
|
|
|
|||
|
|
def matches(self, values: Dict[str, float]) -> bool:
|
|||
|
|
return all(c.matches(values) for c in self.conditions)
|
|||
|
|
|
|||
|
|
def describe(self) -> str:
|
|||
|
|
conds = " 且 ".join(f"{c.feature}{c.op}{c.threshold}" for c in self.conditions)
|
|||
|
|
return f"[{self.severity.value}] {self.id}: {conds}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class Alert:
|
|||
|
|
"""一次预警命中（规则 + 触发时的特征快照）。"""
|
|||
|
|
|
|||
|
|
rule_id: str
|
|||
|
|
severity: AlertSeverity
|
|||
|
|
message: str
|
|||
|
|
sop: str
|
|||
|
|
timestamp: float
|
|||
|
|
snapshot: Dict[str, float] = field(default_factory=dict)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class AlertRuleEngine:
|
|||
|
|
"""预警规则引擎：评估特征向量，产出带 severity 的 Alert 列表。
|
|||
|
|
|
|||
|
|
换行业只改模板配置（AlertRule 列表），引擎零改动（PRD line 152/171）。
|
|||
|
|
|
|||
|
|
用法::
|
|||
|
|
|
|||
|
|
engine = AlertRuleEngine(rules)
|
|||
|
|
alerts = engine.evaluate(timestamp=100, values={"炉温_ema5": 920.0})
|
|||
|
|
if alerts:
|
|||
|
|
primary = engine.primary_alert(alerts) # 取最高 severity
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(self, rules: Sequence[AlertRule]):
|
|||
|
|
if not rules:
|
|||
|
|
raise ValueError("AlertRuleEngine 至少需要 1 条规则")
|
|||
|
|
ids = set()
|
|||
|
|
for r in rules:
|
|||
|
|
if r.id in ids:
|
|||
|
|
raise ValueError(f"规则 id 重复：{r.id!r}")
|
|||
|
|
ids.add(r.id)
|
|||
|
|
self.rules: List[AlertRule] = list(rules)
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def from_template_config(cls, path: str) -> "AlertRuleEngine":
|
|||
|
|
return cls(load_alert_rules_config(path).rules)
|
|||
|
|
|
|||
|
|
def evaluate(self, timestamp: float,
|
|||
|
|
values: Dict[str, float]) -> List[Alert]:
|
|||
|
|
"""评估一个特征向量，返回全部命中规则的 Alert（按 severity 降序）。"""
|
|||
|
|
hits: List[Alert] = []
|
|||
|
|
for rule in self.rules:
|
|||
|
|
if rule.matches(values):
|
|||
|
|
hits.append(Alert(
|
|||
|
|
rule_id=rule.id, severity=rule.severity,
|
|||
|
|
message=rule.message, sop=rule.sop,
|
|||
|
|
timestamp=timestamp, snapshot=dict(values),
|
|||
|
|
))
|
|||
|
|
hits.sort(key=lambda a: a.severity.rank, reverse=True)
|
|||
|
|
return hits
|
|||
|
|
|
|||
|
|
def primary_alert(self, alerts: Sequence[Alert]) -> Optional[Alert]:
|
|||
|
|
"""取最高 severity 的主告警（驾驶舱红色告警）。无命中返回 None。"""
|
|||
|
|
return alerts[0] if alerts else None
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 模板配置（零依赖 YAML 子集解析，对齐 #70 / data-bus / rag-kb）
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class AlertRulesTemplateConfig:
|
|||
|
|
"""模板预警规则配置：模板元信息 + AlertRule 列表。"""
|
|||
|
|
|
|||
|
|
template: str
|
|||
|
|
version: str
|
|||
|
|
rules: List[AlertRule]
|
|||
|
|
description: str = ""
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _parse_scalar(text: str) -> str:
|
|||
|
|
t = text.split(" #", 1)[0].strip()
|
|||
|
|
if len(t) >= 2 and t[0] == t[-1] and t[0] in ("'", '"'):
|
|||
|
|
return t[1:-1]
|
|||
|
|
return t
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _parse_flow_value(text: str):
|
|||
|
|
"""解析 ``key: value`` 右侧的值，支持行内 flow map ``{k: v, k: v}``。
|
|||
|
|
|
|||
|
|
其余（标量 / 引号串）退化为 :func:`_parse_scalar`。flow map 用于
|
|||
|
|
``conditions: [{feature: x, op: ">", threshold: 900.0}]`` 这种紧凑声明。
|
|||
|
|
"""
|
|||
|
|
t = text.split(" #", 1)[0].strip()
|
|||
|
|
if t.startswith("{") and t.endswith("}"):
|
|||
|
|
inner = t[1:-1].strip()
|
|||
|
|
out: Dict[str, object] = {}
|
|||
|
|
if not inner:
|
|||
|
|
return out
|
|||
|
|
for part in inner.split(","):
|
|||
|
|
if ":" not in part:
|
|||
|
|
raise ValueError(f"flow map 项不是键值对：{part!r}")
|
|||
|
|
k, _, v = part.partition(":")
|
|||
|
|
out[k.strip()] = _parse_scalar(v)
|
|||
|
|
return out
|
|||
|
|
return _parse_scalar(text)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _strip_comments(lines: List[str]) -> List[Tuple[str, int]]:
|
|||
|
|
out = []
|
|||
|
|
for i, ln in enumerate(lines):
|
|||
|
|
s = ln.strip()
|
|||
|
|
if not s or s.startswith("#"):
|
|||
|
|
continue
|
|||
|
|
out.append((ln, i + 1))
|
|||
|
|
return out
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _parse_node(lines: List[Tuple[str, int]], i: int, indent: int):
|
|||
|
|
text, _ = lines[i]
|
|||
|
|
if text.lstrip(" ").startswith("- "):
|
|||
|
|
items: List[object] = []
|
|||
|
|
while i < len(lines):
|
|||
|
|
t, no = lines[i]
|
|||
|
|
stripped = t.lstrip(" ")
|
|||
|
|
if not stripped.startswith("- "):
|
|||
|
|
break
|
|||
|
|
lead_j = len(t) - len(t.lstrip(" "))
|
|||
|
|
if lead_j != indent:
|
|||
|
|
break
|
|||
|
|
item_text = stripped[2:].strip()
|
|||
|
|
if not item_text:
|
|||
|
|
raise ValueError(f"alerts.yaml 第 {no} 行：list 项为空")
|
|||
|
|
# 行内 flow map（{k: v, ...}）优先用 _parse_flow_value，避免被
|
|||
|
|
# 下方「含 : 即 map 项」分支误判（flow map 也含 :）。
|
|||
|
|
if item_text.startswith("{") and item_text.endswith("}"):
|
|||
|
|
items.append(_parse_flow_value(item_text))
|
|||
|
|
i += 1
|
|||
|
|
elif ":" in item_text:
|
|||
|
|
map_indent = len(t) - len(t.lstrip(" ")) + 2
|
|||
|
|
lines[i] = (" " * map_indent + item_text, no)
|
|||
|
|
v, i = _parse_node(lines, i, map_indent)
|
|||
|
|
items.append(v)
|
|||
|
|
else:
|
|||
|
|
items.append(_parse_flow_value(item_text))
|
|||
|
|
i += 1
|
|||
|
|
return items, i
|
|||
|
|
result: Dict[str, object] = {}
|
|||
|
|
while i < len(lines):
|
|||
|
|
t, no = lines[i]
|
|||
|
|
lead_j = len(t) - len(t.lstrip(" "))
|
|||
|
|
if lead_j < indent or t.lstrip(" ").startswith("- "):
|
|||
|
|
break
|
|||
|
|
if lead_j > indent:
|
|||
|
|
raise ValueError(f"alerts.yaml 第 {no} 行缩进异常")
|
|||
|
|
if ":" not in t:
|
|||
|
|
raise ValueError(f"alerts.yaml 第 {no} 行不是合法键值对：{t!r}")
|
|||
|
|
key, _, rest = t.partition(":")
|
|||
|
|
key = key.strip()
|
|||
|
|
rest = rest.strip()
|
|||
|
|
if rest:
|
|||
|
|
result[key] = _parse_flow_value(rest)
|
|||
|
|
i += 1
|
|||
|
|
continue
|
|||
|
|
if i + 1 >= len(lines):
|
|||
|
|
raise ValueError(f"alerts.yaml 第 {no} 行 {key!r} 缺少值")
|
|||
|
|
sub_indent = len(lines[i + 1][0]) - len(lines[i + 1][0].lstrip(" "))
|
|||
|
|
if sub_indent <= indent:
|
|||
|
|
raise ValueError(f"alerts.yaml 第 {no} 行 {key!r} 缺少值（无嵌套）")
|
|||
|
|
v, i = _parse_node(lines, i + 1, sub_indent)
|
|||
|
|
result[key] = v
|
|||
|
|
return result, i
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _load_yaml_text(text: str) -> Dict[str, object]:
|
|||
|
|
lines = _strip_comments(text.splitlines())
|
|||
|
|
if not lines:
|
|||
|
|
return {}
|
|||
|
|
top_indent = len(lines[0][0]) - len(lines[0][0].lstrip(" "))
|
|||
|
|
value, next_i = _parse_node(lines, 0, top_indent)
|
|||
|
|
if not isinstance(value, dict):
|
|||
|
|
raise ValueError("alerts.yaml 顶层必须是 map")
|
|||
|
|
if next_i < len(lines):
|
|||
|
|
raise ValueError(f"alerts.yaml 第 {lines[next_i][1]} 行：顶层存在多个节点")
|
|||
|
|
return value
|
|||
|
|
|
|||
|
|
|
|||
|
|
def load_alert_rules_config(path: str) -> AlertRulesTemplateConfig:
|
|||
|
|
"""从模板预警规则 YAML 资产加载配置。
|
|||
|
|
|
|||
|
|
期望结构（详见 ``config/alert_rules.template.yaml``）::
|
|||
|
|
|
|||
|
|
template: ti-cl4
|
|||
|
|
version: 1.0.0
|
|||
|
|
rules:
|
|||
|
|
- id: bed_temp_critical
|
|||
|
|
severity: P0
|
|||
|
|
message: 炉温超上限，立即降流减料
|
|||
|
|
sop: SOP-CL-001
|
|||
|
|
conditions:
|
|||
|
|
- {feature: 炉温_ema5, op: ">", threshold: 900.0}
|
|||
|
|
"""
|
|||
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|||
|
|
data = _load_yaml_text(fh.read())
|
|||
|
|
|
|||
|
|
template = str(data.get("template", "")).strip()
|
|||
|
|
if not template:
|
|||
|
|
raise ValueError("alerts.yaml 缺少 template 字段")
|
|||
|
|
version = str(data.get("version", "1.0.0")).strip() or "1.0.0"
|
|||
|
|
description = str(data.get("description", "")).strip()
|
|||
|
|
|
|||
|
|
raw_rules = data.get("rules") or []
|
|||
|
|
if not isinstance(raw_rules, list):
|
|||
|
|
raise ValueError("alerts.yaml rules 必须是 list")
|
|||
|
|
rules: List[AlertRule] = []
|
|||
|
|
for idx, item in enumerate(raw_rules):
|
|||
|
|
if not isinstance(item, dict):
|
|||
|
|
raise ValueError(f"alerts.yaml rules[{idx}] 必须是 map")
|
|||
|
|
rid = str(item.get("id", "")).strip()
|
|||
|
|
sev_name = str(item.get("severity", "")).strip().upper()
|
|||
|
|
sev_map = {s.value: s for s in AlertSeverity}
|
|||
|
|
if sev_name not in sev_map:
|
|||
|
|
raise ValueError(
|
|||
|
|
f"alerts.yaml rules[{idx}] 未知 severity {sev_name!r}"
|
|||
|
|
f"（应为 {sorted(sev_map)}）")
|
|||
|
|
message = str(item.get("message", "")).strip()
|
|||
|
|
sop = str(item.get("sop", "")).strip()
|
|||
|
|
raw_conds = item.get("conditions") or []
|
|||
|
|
if not isinstance(raw_conds, list):
|
|||
|
|
raise ValueError(f"alerts.yaml rules[{idx}] conditions 必须是 list")
|
|||
|
|
conds: List[AlertCondition] = []
|
|||
|
|
for ci, c in enumerate(raw_conds):
|
|||
|
|
if not isinstance(c, dict):
|
|||
|
|
raise ValueError(f"alerts.yaml rules[{idx}].conditions[{ci}] 必须是 map")
|
|||
|
|
feature = str(c.get("feature", "")).strip()
|
|||
|
|
op = str(c.get("op", "")).strip()
|
|||
|
|
if op not in OPS:
|
|||
|
|
raise ValueError(
|
|||
|
|
f"alerts.yaml rules[{idx}].conditions[{ci}] 未知 op {op!r}")
|
|||
|
|
try:
|
|||
|
|
threshold = float(c.get("threshold"))
|
|||
|
|
except (TypeError, ValueError) as exc:
|
|||
|
|
raise ValueError(
|
|||
|
|
f"alerts.yaml rules[{idx}].conditions[{ci}] threshold 不是数值") from exc
|
|||
|
|
conds.append(AlertCondition(feature=feature, op=op, threshold=threshold))
|
|||
|
|
rules.append(AlertRule(id=rid, severity=sev_map[sev_name],
|
|||
|
|
conditions=conds, message=message, sop=sop))
|
|||
|
|
return AlertRulesTemplateConfig(template=template, version=version,
|
|||
|
|
rules=rules, description=description)
|