From 8e5ae968397febfd5a6a9161dfe621251a852291 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 02:57:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(#52):=20=E6=8A=A5=E8=AD=A6=E7=9C=8B?= =?UTF-8?q?=E6=9D=BF=E9=85=8D=E7=BD=AE=E5=8C=96=EF=BC=88=E9=98=88=E5=80=BC?= =?UTF-8?q?/=E8=A7=84=E5=88=99=E5=A4=96=E7=BD=AE=EF=BC=8CPRD=205.5=20?= =?UTF-8?q?=E5=91=8A=E8=AD=A6=E9=9D=A2=E6=9D=BF=E9=85=8D=E7=BD=AE=E5=8C=96?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把 alarm_panel 组件的展示语义外置为模板配置资产，使「切换行业模板零改前端代码」 （PRD 5.5 验收口径）覆盖到告警面板： - core/cockpit/alarm_config.py：AlarmPanelConfig 配置点 - severityColors（P0/P1/P2 三级 fg/bg/icon/border，覆盖主题默认告警色； P0 fg 即驾驶舱红色告警 PRD 5.3 ③ 场景A） - rulesSource / thresholdsSource（asset 资产路径 或 inline 内联规则， 阈值/规则外置 JSON，PRD line 152/171） - showSop（PRD 场景A LLM 报警解释 + 值班长确认是否展开 SOP） - requireAck / ackTimeoutS（高利害人工确认，PRD line 333；超时升级） - muteLower / groupBy / sortBy / maxItems（静默下限 + 大屏展示策略） - validate_alarm_config（聚合全部字段级错误，沿用 #50 风格） - load_alarm_config（校验失败抛 AlarmConfigError 携带错误清单） - render_alarm_panel_props（编译为 alarm_panel 组件 props，#51 渲染层合并即可） - filter_alarms_by_mute（验证 muteLower 配置点真实驱动展示策略） - core/cockpit/__init__.py：导出 #52 告警面板配置 API（#50/#51 合入后合并导出） - templates/ti-cl4/dashboard/alarm_panel.ti.yaml：氯化车间/海绵钛告警面板配置资产 - core/cockpit/scripts/verify_alarm_config.py：验收脚本（4 能力点，exit 0） - core/cockpit/tests/test_alarm_config.py：28 用例（校验/加载/渲染/过滤/负例） - core/cockpit/README.md：模块说明与用法 零运行时依赖（仅标准库）；与 #50/#51 解耦，不 import 未合入分支。 测试：python -m unittest discover -s tests -p test_alarm_config.py（28 OK） --- core/cockpit/README.md | 82 +++ core/cockpit/__init__.py | 58 ++ core/cockpit/alarm_config.py | 519 ++++++++++++++++++ core/cockpit/scripts/verify_alarm_config.py | 187 +++++++ core/cockpit/tests/test_alarm_config.py | 338 ++++++++++++ .../ti-cl4/dashboard/alarm_panel.ti.yaml | 50 ++ 6 files changed, 1234 insertions(+) create mode 100644 core/cockpit/README.md create mode 100644 core/cockpit/__init__.py create mode 100644 core/cockpit/alarm_config.py create mode 100644 core/cockpit/scripts/verify_alarm_config.py create mode 100644 core/cockpit/tests/test_alarm_config.py create mode 100644 templates/ti-cl4/dashboard/alarm_panel.ti.yaml diff --git a/core/cockpit/README.md b/core/cockpit/README.md new file mode 100644 index 0000000..edd2fff --- /dev/null +++ b/core/cockpit/README.md @@ -0,0 +1,82 @@ +# 配置化驾驶舱内核模块（iAOP-Core / cockpit） + +对齐 PRD 5.5「⑤ 配置化驾驶舱」：把驾驶舱抽象为模板化配置资产，按"布局/配置"渲染， +切换行业模板零改前端代码（PRD 5.5 验收口径）。 + +## 模块状态 + +| 能力 | Issue | 状态 | +| --- | --- | --- | +| 布局 JSON Schema 定义 + 校验/解析 | #50 | feature/issue-50（推进中） | +| 图表组件配置化渲染引擎（RenderPlan） | #51 | feature/issue-51（推进中） | +| **报警看板配置化（阈值/规则外置）** | **#52** | **本分支 feature/issue-52** | + +#52「报警看板配置化」把告警面板（`alarm_panel`）的展示语义外置为模板配置资产， +让"换行业只换配置、前端代码零改动"也覆盖到告警面板。配置点： + +- **严重度→颜色映射**（`severityColors`）：P0/P1/P2 三级各自的 `fg`/`bg`/可选 `icon`/`border`， + 覆盖主题默认告警色；驾驶舱红色告警（PRD 5.3 ③ 场景A）即由 `P0` 的 `fg` 决定。 +- **告警规则 / 阈值源绑定**（`rulesSource` / `thresholdsSource`）：告警面板订阅哪份 + 规则资产（`kind: asset` + `ref`）或内联规则（`kind: inline` + `data`）——把"看哪条规则" + 也变成配置项，避免把规则 id 写死在前端。 +- **SOP 联动**（`showSop`）：PRD 场景A「LLM 生成原因+处置建议 → 值班长确认」是否在面板展开。 +- **确认 / 静默行为**（`requireAck` / `ackTimeoutS` / `muteLower`）：关键告警是否强制人工确认、 + 超时升级、是否静默低于某 severity 的提示。 +- **分组 / 排序 / 最大条数**（`groupBy` / `sortBy` / `maxItems`）：大屏展示策略外置。 + +## 用法 + +```python +from cockpit import ( + validate_alarm_config, load_alarm_config, render_alarm_panel_props, +) + +# 1) 配置资产（通常来自模板 YAML；这里用 dict 示意） +config = { + "$schema": "iAOP-cockpit-alarm-panel-v1", + "severityColors": [ + {"severity": "P0", "fg": "#ff3b30", "bg": "rgba(255,59,48,0.12)", "icon": "alert-octagon"}, + {"severity": "P1", "fg": "#f5a623", "bg": "rgba(245,166,35,0.12)"}, + {"severity": "P2", "fg": "#3aa0ff", "bg": "rgba(58,160,255,0.10)"}, + ], + "rulesSource": {"kind": "asset", "ref": "ti-cl4/impurity-forecast/config/alert_rules.template.yaml"}, + "showSop": True, "requireAck": True, "ackTimeoutS": 300, "muteLower": "P2", + "groupBy": "severity", "sortBy": "severity", "maxItems": 50, +} + +# 2) 校验（聚合全部字段级错误，便于配置台「错误列表」展示） +result = validate_alarm_config(config) +assert result.ok, result.errors + +# 3) 加载为内存模型 +cfg = load_alarm_config(config) + +# 4) 编译为 alarm_panel 组件的 props（#51 渲染层在 _build_props 里合并即可） +props = render_alarm_panel_props(cfg) +# -> {'subscribe': 'alarm_stream', 'severityStyles': {...}, 'rulesSource': {...}, +# 'showSop': True, 'requireAck': True, 'ackTimeoutS': 300, ...} +``` + +切换行业模板（如树脂）= 换一份配置资产（不同配色 / 规则源），`render_alarm_panel_props` +随配置变化产出不同 props，**前端代码零改动**。 + +## 设计要点 + +- **零运行时依赖**：只用标准库，配置资产是可 `json.dumps` 的纯 dict，便于配置台发布与审计。 +- **声明式 + 强校验**：`validate_alarm_config` 收集全部字段级错误（沿用 #50 + `LayoutValidationResult` 风格）；`load_alarm_config` 校验失败抛 `AlarmConfigError` + 并携带错误清单。 +- **与 #50/#51 解耦**：本模块不 import `layout` / `renderer`（它们在各自 feature 分支上推进）， + 避免对未合并分支形成硬依赖；`render_alarm_panel_props` 仅产出 `alarm_panel` 组件的 props dict， + 由 #51 渲染层在 `_build_props` 里合并即可。#50/#51 合入 main 后，本模块的 `__init__` + 会合并它们的导出（`CockpitLayout` / `validate_layout` / `RenderPlan` 等）。 + +## 测试 / 验收 + +```bash +# 在 core/cockpit 目录下 +python -m unittest discover -s tests -p "test_alarm_config.py" -v # 28 用例 +python scripts/verify_alarm_config.py # 验收脚本 +``` + +参考资产：`templates/ti-cl4/dashboard/alarm_panel.ti.yaml`（氯化车间/海绵钛告警面板配置）。 diff --git a/core/cockpit/__init__.py b/core/cockpit/__init__.py new file mode 100644 index 0000000..f583997 --- /dev/null +++ b/core/cockpit/__init__.py @@ -0,0 +1,58 @@ +# -*- coding: utf-8 -*- +"""配置化驾驶舱内核模块（iAOP-Core / cockpit）—— 对齐 PRD 5.5「⑤ 配置化驾驶舱」。 + +当前对外暴露 issue #52「报警看板配置化（阈值/规则外置）」的能力：把告警面板的 +展示语义（严重度配色 / 规则源 / SOP / 确认 / 分组排序）外置为模板配置资产。 + +布局 JSON Schema（#50）与渲染层（#51）在其各自 feature 分支上推进；合入 main +后，本 ``__init__`` 会合并它们的导出（``CockpitLayout`` / ``validate_layout`` / +``RenderPlan`` 等）。当前只导出 #52 的告警面板配置 API，保证 #52 在 main 上 +可独立运行、可独立测试。 +""" +from __future__ import annotations + +from .alarm_config import ( # noqa: F401 + ALARM_CONFIG_SCHEMA_ID, + ALARM_CONFIG_SCHEMA_VERSION, + DEFAULT_ACK_TIMEOUT_S, + DEFAULT_MAX_ITEMS, + DEFAULT_MUTE_LOWER, + MAX_ALLOWED_ITEMS, + SEVERITY_RANK, + VALID_GROUP_BY, + VALID_SEVERITIES, + VALID_SORT_BY, + VALID_SOURCE_KINDS, + AlarmConfigError, + AlarmConfigValidationResult, + AlarmPanelConfig, + AlarmRulesSource, + SeverityColor, + filter_alarms_by_mute, + load_alarm_config, + render_alarm_panel_props, + validate_alarm_config, +) + +__all__ = [ + "ALARM_CONFIG_SCHEMA_ID", + "ALARM_CONFIG_SCHEMA_VERSION", + "DEFAULT_ACK_TIMEOUT_S", + "DEFAULT_MAX_ITEMS", + "DEFAULT_MUTE_LOWER", + "MAX_ALLOWED_ITEMS", + "SEVERITY_RANK", + "VALID_GROUP_BY", + "VALID_SEVERITIES", + "VALID_SORT_BY", + "VALID_SOURCE_KINDS", + "AlarmConfigError", + "AlarmConfigValidationResult", + "AlarmPanelConfig", + "AlarmRulesSource", + "SeverityColor", + "filter_alarms_by_mute", + "load_alarm_config", + "render_alarm_panel_props", + "validate_alarm_config", +] diff --git a/core/cockpit/alarm_config.py b/core/cockpit/alarm_config.py new file mode 100644 index 0000000..552c7a6 --- /dev/null +++ b/core/cockpit/alarm_config.py @@ -0,0 +1,519 @@ +# -*- coding: utf-8 -*- +"""报警看板配置化（issue #52 / PRD 5.5「⑤ 配置化驾驶舱」）。 + +PRD 5.5 的告警面板（``alarm_panel``）在 #50/#51 里被实现为「固定订阅 +``alarm_stream``、套用主题默认告警色」的硬编码组件。本模块把告警面板的 +**展示语义外置为模板配置**（PRD line 152/171：阈值/规则外置 JSON，行业工程师 +在配置台维护），让"换行业只换配置资产、前端代码零改动"这条验收口径也覆盖到 +告警面板。 + +外置的配置点（均由行业模板在配置台维护）： + +1. **严重度→颜色映射**（``severity_colors``）：P0/P1/P2 三级各自的前景色 / + 背景色 / 图标，覆盖主题默认告警色；驾驶舱红色告警（PRD 5.3 ③ 场景A） + 即由 ``P0`` 的 ``fg`` 决定，**换行业只改这张映射表**。 +2. **告警规则 / 阈值源绑定**（``rules_source`` / ``thresholds_source``）： + 告警面板订阅哪份规则资产（如 ``templates/ti-cl4/impurity-forecast/ + config/alert_rules.template.yaml``）与哪份阈值包——把"看哪条规则" + 也变成配置项，避免把规则 id 写死在前端。 +3. **SOP 联动开关**（``show_sop``）：PRD 场景A「LLM 生成原因+处置建议 → + 值班长确认」，是否在面板里展开处置 SOP（高利害人工确认，PRD line 333）。 +4. **确认 / 静默行为**（``require_ack`` / ``ack_timeout_s`` / ``mute_lower``）： + 关键告警是否强制人工确认、超时升级、是否静默低于某 severity 的提示。 +5. **分组 / 排序 / 最大条数**（``group_by`` / ``sort_by`` / ``max_items``）： + 大屏展示策略外置（与 #51 的 perf 策略互补：这里是"展示多少/怎么排"， + #51 是"怎么渲染得快"）。 + +设计要点 +-------- +- **零运行时依赖**：与 #50/#51、data-bus、rag-kb 一致，只用标准库； + 配置资产是可 ``json.dumps`` 的纯 dict，便于配置台发布与审计。 +- **声明式 + 强校验**：``validate_alarm_config`` 收集全部字段级错误 + （沿用 #50 ``LayoutValidationResult`` 风格），``load_alarm_config`` 在校验 + 不通过时抛 ``AlarmConfigError`` 并携带错误清单，便于配置台「错误列表」展示。 +- **与 #50/#51 解耦**：本模块不 import ``layout`` / ``renderer``（它们尚未合入 + main），避免对未合并分支形成硬依赖；``render_alarm_panel_props`` 仅产出一份 + ``alarm_panel`` 组件的 props dict，由 #51 渲染层在 ``_build_props`` 里合并即可。 +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + +# --------------------------------------------------------------------------- +# 版本标识（被告警面板配置资产的 $schema 引用） +# --------------------------------------------------------------------------- +ALARM_CONFIG_SCHEMA_ID: str = "iAOP-cockpit-alarm-panel-v1" +ALARM_CONFIG_SCHEMA_VERSION: int = 1 + +# --------------------------------------------------------------------------- +# 合法性集合 +# --------------------------------------------------------------------------- +# 严重度三级（与 templates/ti-cl4/impurity-forecast/alert_rules.py 的 AlertSeverity 对齐）。 +VALID_SEVERITIES: Tuple[str, ...] = ("P0", "P1", "P2") + +# severity 排序权重（越大越严重；用于 mute_lower / sort_by=severity 的排序基准）。 +SEVERITY_RANK: Dict[str, int] = {"P0": 3, "P1": 2, "P2": 1} + +# severity_colors 里每个 severity 必须声明的颜色键。 +_REQUIRED_COLOR_KEYS: Tuple[str, ...] = ("fg", "bg") +# severity_colors 里允许额外声明的可选颜色键（图标 / 边框）。 +_OPTIONAL_COLOR_KEYS: Tuple[str, ...] = ("icon", "border") + +# rules_source / thresholds_source 允许的 ``kind`` 取值： +# - asset : 指向模板仓库内的一份配置资产路径（如 alert_rules.template.yaml） +# - inline : 内联在配置里（rules_source.data 直接给出规则声明列表） +VALID_SOURCE_KINDS: Tuple[str, ...] = ("asset", "inline") + +# 告警面板允许的分组维度（PRD 5.5 能力：可按 severity / 工序 / 规则分组）。 +VALID_GROUP_BY: Tuple[str, ...] = ("severity", "rule", "process") + +# 告警面板允许的排序键（severity 按严重度；time 按触发时间倒序）。 +VALID_SORT_BY: Tuple[str, ...] = ("severity", "time") + +# 静默下限：低于该 severity 的告警不展示（默认 P2 = 提示级也展示，即不静默）。 +DEFAULT_MUTE_LOWER: str = "P2" + +# 单面板最大展示条数（PRD 5.5 大屏策略：超过则折叠 + 计数角标）。 +DEFAULT_MAX_ITEMS: int = 50 +MAX_ALLOWED_ITEMS: int = 500 + +# 确认超时下限（秒）：require_ack=True 时，超时未确认自动升级 severity。 +DEFAULT_ACK_TIMEOUT_S: int = 300 + + +# --------------------------------------------------------------------------- +# 异常 / 结果 +# --------------------------------------------------------------------------- +class AlarmConfigError(ValueError): + """告警面板配置校验失败。``load_alarm_config`` 在校验不通过时抛出。 + + ``errors`` 收集全部字段级错误，便于配置台「错误列表（行号+原因）」展示， + 沿用 #50 ``LayoutValidationError`` 的多错误聚合风格。 + """ + + def __init__(self, errors: List[str]): + super().__init__("; ".join(errors) if errors else "alarm panel config validation failed") + self.errors: List[str] = list(errors) + + +@dataclass +class AlarmConfigValidationResult: + """``validate_alarm_config`` 的返回值，区分「是否合法」与「全部错误清单」。""" + + ok: bool + errors: List[str] = field(default_factory=list) + # 校验通过后回填的规范化配置（补默认值后的 dict），便于直接发布/渲染。 + normalized: Optional[Dict[str, Any]] = None + + +# --------------------------------------------------------------------------- +# 内存模型（dataclass） +# --------------------------------------------------------------------------- +@dataclass +class SeverityColor: + """单个严重度的展示配色（覆盖主题默认告警色）。 + + Attributes: + severity: P0 / P1 / P2。 + fg: 前景色（告警文本 / 图标颜色；P0 的 fg 即驾驶舱「红色告警」）。 + bg: 背景色（告警条底色）。 + icon: 可选图标名（如 ``"alert-triangle"``）。 + border: 可选左边框色（用于告警条强调）。 + """ + + severity: str + fg: str + bg: str + icon: Optional[str] = None + border: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = {"severity": self.severity, "fg": self.fg, "bg": self.bg} + if self.icon is not None: + d["icon"] = self.icon + if self.border is not None: + d["border"] = self.border + return d + + +@dataclass +class AlarmRulesSource: + """告警规则 / 阈值的来源绑定（PRD「阈值外置 JSON」）。 + + Attributes: + kind: ``asset``（模板仓库内资产路径）或 ``inline``（内联声明）。 + ref: ``asset`` 时的资产路径（相对模板根，如 + ``ti-cl4/impurity-forecast/config/alert_rules.template.yaml``）。 + data: ``inline`` 时的规则声明列表（每条是 {id, severity, ...} dict）。 + """ + + kind: str + ref: Optional[str] = None + data: Optional[List[Dict[str, Any]]] = None + + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = {"kind": self.kind} + if self.ref is not None: + d["ref"] = self.ref + if self.data is not None: + d["data"] = list(self.data) + return d + + +@dataclass +class AlarmPanelConfig: + """一份告警面板配置的内存模型（对应一份模板级配置资产）。 + + 渲染层（Vue3 / 配置台）消费本对象即可驱动 ``AlarmPanel`` 组件的全部展示 + 行为：切换行业模板 = 加载另一份 ``AlarmPanelConfig``，**前端代码零改动**。 + """ + + severity_colors: List[SeverityColor] + rules_source: AlarmRulesSource + thresholds_source: Optional[AlarmRulesSource] = None + show_sop: bool = True + require_ack: bool = False + ack_timeout_s: int = DEFAULT_ACK_TIMEOUT_S + mute_lower: str = DEFAULT_MUTE_LOWER + group_by: str = "severity" + sort_by: str = "severity" + max_items: int = DEFAULT_MAX_ITEMS + schema: str = ALARM_CONFIG_SCHEMA_ID + + def to_dict(self) -> Dict[str, Any]: + """序列化为可发布的配置资产 dict（结构对齐校验器输入）。""" + d: Dict[str, Any] = { + "$schema": self.schema, + "severityColors": [c.to_dict() for c in self.severity_colors], + "rulesSource": self.rules_source.to_dict(), + "showSop": self.show_sop, + "requireAck": self.require_ack, + "muteLower": self.mute_lower, + "groupBy": self.group_by, + "sortBy": self.sort_by, + "maxItems": self.max_items, + } + if self.thresholds_source is not None: + d["thresholdsSource"] = self.thresholds_source.to_dict() + if self.require_ack: + d["ackTimeoutS"] = self.ack_timeout_s + return d + + +# --------------------------------------------------------------------------- +# 校验器 +# --------------------------------------------------------------------------- +def _is_str_nonempty(value: Any) -> bool: + return isinstance(value, str) and value.strip() != "" + + +def _validate_source( + source: Any, field_name: str, errors: List[str], required: bool +) -> Optional[Dict[str, Any]]: + """校验一个 rules_source / thresholds_source dict。 + + 返回规范化后的 dict（校验通过时），或 None（非法 / 缺失）。 + """ + ctx = field_name + if source is None: + if required: + errors.append(f"{ctx}: 缺失（告警面板必须绑定 rulesSource）") + return None + if not isinstance(source, dict): + errors.append(f"{ctx}: 必须是对象（dict），实际为 {type(source).__name__}") + return None + + kind = source.get("kind") + if kind not in VALID_SOURCE_KINDS: + errors.append( + f"{ctx}.kind: 非法 {kind!r}，合法值 {list(VALID_SOURCE_KINDS)}" + ) + return None + + normalized: Dict[str, Any] = {"kind": kind} + + if kind == "asset": + ref = source.get("ref") + if not _is_str_nonempty(ref): + errors.append(f"{ctx}.ref: kind=asset 时必须给出非空资产路径") + else: + normalized["ref"] = ref + else: # inline + data = source.get("data") + if not isinstance(data, list) or not data: + errors.append(f"{ctx}.data: kind=inline 时必须给出非空规则声明列表") + else: + # 每条内联规则至少要有 id（便于驾驶舱引用 / 审计） + bad = [ + str(i) + for i, item in enumerate(data) + if not isinstance(item, dict) or not _is_str_nonempty(item.get("id")) + ] + if bad: + errors.append( + f"{ctx}.data: 内联规则项 {','.join(bad)} 缺失 id 或非对象" + ) + else: + normalized["data"] = list(data) + + return normalized + + +def validate_alarm_config(config: Any) -> AlarmConfigValidationResult: + """对一份告警面板配置资产做结构 + 语义校验，返回校验结果。 + + 非法资产不会提前返回：尽量收集全部字段级错误，便于配置台一次性展示 + 「错误列表（字段 + 原因）」，与 #50 ``validate_layout`` 行为一致。 + """ + errors: List[str] = [] + + if not isinstance(config, dict): + return AlarmConfigValidationResult( + ok=False, errors=[f"配置必须是对象（dict），实际为 {type(config).__name__}"] + ) + + # $schema（选填，但若给出必须对齐版本标识） + schema = config.get("$schema") + if schema is not None and schema != ALARM_CONFIG_SCHEMA_ID: + errors.append( + f"$schema: 应为 {ALARM_CONFIG_SCHEMA_ID!r}，实际为 {schema!r}" + ) + + # severityColors：必填、至少覆盖 P0/P1/P2 三级、每级颜色键齐全 + raw_colors = config.get("severityColors", config.get("severity_colors")) + color_by_sev: Dict[str, Dict[str, Any]] = {} + if not isinstance(raw_colors, list) or not raw_colors: + errors.append("severityColors: 缺失或非列表（至少需要 P0/P1/P2 三级配色）") + else: + for i, item in enumerate(raw_colors): + ctx = f"severityColors[{i}]" + if not isinstance(item, dict): + errors.append(f"{ctx}: 必须是对象（dict）") + continue + sev = item.get("severity") + if sev not in VALID_SEVERITIES: + errors.append( + f"{ctx}.severity: 非法 {sev!r}，合法值 {list(VALID_SEVERITIES)}" + ) + continue + if sev in color_by_sev: + errors.append(f"{ctx}.severity: {sev!r} 重复声明") + continue + norm_color: Dict[str, Any] = {"severity": sev} + color_ok = True + for key in _REQUIRED_COLOR_KEYS: + v = item.get(key) + if not _is_str_nonempty(v): + errors.append(f"{ctx}.{key}: 缺失或非非空字符串") + color_ok = False + else: + norm_color[key] = v + for key in _OPTIONAL_COLOR_KEYS: + v = item.get(key) + if v is None: + continue + if not _is_str_nonempty(v): + errors.append(f"{ctx}.{key}: 给出则必须是非空字符串") + else: + norm_color[key] = v + if color_ok: + color_by_sev[sev] = norm_color + + for sev in VALID_SEVERITIES: + if sev not in color_by_sev: + errors.append(f"severityColors: 缺少 {sev} 级配色（必须覆盖 P0/P1/P2）") + + # rulesSource：必填 + raw_rules = config.get("rulesSource", config.get("rules_source")) + norm_rules = _validate_source(raw_rules, "rulesSource", errors, required=True) + + # thresholdsSource：选填（阈值可与规则同源，也可独立） + raw_thr = config.get("thresholdsSource", config.get("thresholds_source")) + norm_thr = _validate_source(raw_thr, "thresholdsSource", errors, required=False) + + # showSop：布尔 + show_sop = config.get("showSop", config.get("show_sop", True)) + if not isinstance(show_sop, bool): + errors.append(f"showSop: 必须是布尔，实际为 {type(show_sop).__name__}") + + # requireAck：布尔 + require_ack = config.get("requireAck", config.get("require_ack", False)) + if not isinstance(require_ack, bool): + errors.append(f"requireAck: 必须是布尔，实际为 {type(require_ack).__name__}") + + # ackTimeoutS：require_ack 时才生效，给出则必须是正整数 + ack_timeout = config.get("ackTimeoutS", config.get("ack_timeout_s", DEFAULT_ACK_TIMEOUT_S)) + if isinstance(ack_timeout, bool) or not isinstance(ack_timeout, int) or ack_timeout <= 0: + errors.append(f"ackTimeoutS: 必须是正整数（秒），实际为 {ack_timeout!r}") + + # muteLower：必须是合法 severity + mute_lower = config.get("muteLower", config.get("mute_lower", DEFAULT_MUTE_LOWER)) + if mute_lower not in VALID_SEVERITIES: + errors.append( + f"muteLower: 非法 {mute_lower!r}，合法值 {list(VALID_SEVERITIES)}" + ) + + # groupBy / sortBy：枚举 + group_by = config.get("groupBy", config.get("group_by", "severity")) + if group_by not in VALID_GROUP_BY: + errors.append(f"groupBy: 非法 {group_by!r}，合法值 {list(VALID_GROUP_BY)}") + sort_by = config.get("sortBy", config.get("sort_by", "severity")) + if sort_by not in VALID_SORT_BY: + errors.append(f"sortBy: 非法 {sort_by!r}，合法值 {list(VALID_SORT_BY)}") + + # maxItems：1..MAX_ALLOWED_ITEMS + max_items = config.get("maxItems", config.get("max_items", DEFAULT_MAX_ITEMS)) + if ( + isinstance(max_items, bool) + or not isinstance(max_items, int) + or not (1 <= max_items <= MAX_ALLOWED_ITEMS) + ): + errors.append( + f"maxItems: 必须是 1..{MAX_ALLOWED_ITEMS} 的整数，实际为 {max_items!r}" + ) + + if errors: + return AlarmConfigValidationResult(ok=False, errors=errors) + + # 规范化输出（统一字段名 + 补默认值），便于直接发布 / 渲染。 + normalized: Dict[str, Any] = { + "$schema": ALARM_CONFIG_SCHEMA_ID, + "severityColors": [color_by_sev[sev] for sev in VALID_SEVERITIES], + "rulesSource": norm_rules, + "showSop": show_sop, + "requireAck": require_ack, + "ackTimeoutS": ack_timeout if require_ack else DEFAULT_ACK_TIMEOUT_S, + "muteLower": mute_lower, + "groupBy": group_by, + "sortBy": sort_by, + "maxItems": max_items, + } + if norm_thr is not None: + normalized["thresholdsSource"] = norm_thr + + return AlarmConfigValidationResult(ok=True, errors=[], normalized=normalized) + + +def load_alarm_config(config: Any) -> AlarmPanelConfig: + """从已解析的 dict 构造 ``AlarmPanelConfig``（校验失败抛 ``AlarmConfigError``）。 + + 与 #50 ``load_layout`` 对称：先 ``validate_alarm_config``，再按规范化结果 + 构造内存模型；校验不通过时把全部错误聚合到异常里。 + """ + result = validate_alarm_config(config) + if not result.ok: + raise AlarmConfigError(result.errors) + assert result.normalized is not None # 校验通过一定回填 normalized + + norm = result.normalized + + severity_colors = [ + SeverityColor( + severity=c["severity"], + fg=c["fg"], + bg=c["bg"], + icon=c.get("icon"), + border=c.get("border"), + ) + for c in norm["severityColors"] + ] + + def _to_source(d: Dict[str, Any]) -> AlarmRulesSource: + return AlarmRulesSource( + kind=d["kind"], + ref=d.get("ref"), + data=d.get("data"), + ) + + thresholds_source = None + if norm.get("thresholdsSource") is not None: + thresholds_source = _to_source(norm["thresholdsSource"]) + + return AlarmPanelConfig( + severity_colors=severity_colors, + rules_source=_to_source(norm["rulesSource"]), + thresholds_source=thresholds_source, + show_sop=norm["showSop"], + require_ack=norm["requireAck"], + ack_timeout_s=norm["ackTimeoutS"], + mute_lower=norm["muteLower"], + group_by=norm["groupBy"], + sort_by=norm["sortBy"], + max_items=norm["maxItems"], + schema=ALARM_CONFIG_SCHEMA_ID, + ) + + +# --------------------------------------------------------------------------- +# 渲染辅助：把告警面板配置编译成 alarm_panel 组件的 props +# --------------------------------------------------------------------------- +def render_alarm_panel_props(config: AlarmPanelConfig) -> Dict[str, Any]: + """把一份 ``AlarmPanelConfig`` 编译成 ``alarm_panel`` 组件的 props dict。 + + #51 渲染层在 ``_build_props`` 里对 ``alarm_panel`` 原本只写死 + ``subscribe="alarm_stream"``；合入本模块后改为合并本函数的输出即可， + 前端 ``AlarmPanel`` 组件据此驱动「颜色 / 规则源 / SOP / 确认 / 分组排序」， + **切换行业模板只换配置资产，前端代码零改动**（PRD 5.5 验收口径）。 + + 返回的 props 与 #51 的 props 风格一致：扁平、可直接 ``json.dumps``、 + 键名用前端友好的 camelCase。 + """ + # severity → 展示描述（颜色 + 可选图标 / 边框），驱动告警条着色 + severity_styles: Dict[str, Dict[str, Any]] = {} + for c in config.severity_colors: + style: Dict[str, Any] = {"fg": c.fg, "bg": c.bg} + if c.icon is not None: + style["icon"] = c.icon + if c.border is not None: + style["border"] = c.border + severity_styles[c.severity] = style + + # 规则 / 阈值源：asset 透传 ref，inline 透传 data + def _source_props(src: AlarmRulesSource) -> Dict[str, Any]: + if src.kind == "asset": + return {"kind": "asset", "ref": src.ref or ""} + return {"kind": "inline", "data": list(src.data or [])} + + props: Dict[str, Any] = { + # 保留 #51 原有的订阅语义（向后兼容） + "subscribe": "alarm_stream", + "severityStyles": severity_styles, + "rulesSource": _source_props(config.rules_source), + "showSop": config.show_sop, + "requireAck": config.require_ack, + "muteLower": config.mute_lower, + "groupBy": config.group_by, + "sortBy": config.sort_by, + "maxItems": config.max_items, + } + if config.thresholds_source is not None: + props["thresholdsSource"] = _source_props(config.thresholds_source) + if config.require_ack: + props["ackTimeoutS"] = config.ack_timeout_s + + return props + + +def filter_alarms_by_mute( + alarms: List[Dict[str, Any]], + mute_lower: str = DEFAULT_MUTE_LOWER, +) -> List[Dict[str, Any]]: + """按 ``mute_lower`` 过滤告警列表（演示配置如何驱动展示策略）。 + + 给定一批告警 dict（每条含 ``severity``），返回严重度 ≥ ``mute_lower`` 的子集。 + 用于验证「静默下限」配置点真正生效——配置台改 ``muteLower`` 即可改变面板 + 展示内容，无需改前端代码。``sort_by=severity`` 时同时按严重度降序排序。 + """ + if mute_lower not in SEVERITY_RANK: + raise AlarmConfigError([f"mute_lower 非法: {mute_lower!r}"]) + floor = SEVERITY_RANK[mute_lower] + kept = [ + a for a in alarms + if isinstance(a, dict) and SEVERITY_RANK.get(a.get("severity"), 0) >= floor + ] + return sorted( + kept, + key=lambda a: SEVERITY_RANK.get(a.get("severity"), 0), + reverse=True, + ) diff --git a/core/cockpit/scripts/verify_alarm_config.py b/core/cockpit/scripts/verify_alarm_config.py new file mode 100644 index 0000000..d9e0306 --- /dev/null +++ b/core/cockpit/scripts/verify_alarm_config.py @@ -0,0 +1,187 @@ +# -*- coding: utf-8 -*- +"""报警看板配置化验收脚本（issue #52，PRD 5.5 验收口径）。 + +验证四个能力点（覆盖 PRD 5.5「切换模板零改码」对告警面板的覆盖）： +1. **Ti 行业模板配置资产合法**：``templates/ti-cl4/dashboard/alarm_panel.ti.yaml`` + 所描述的配置（等价 dict）通过 ``iAOP-cockpit-alarm-panel-v1`` 校验， + 并能渲染成 ``alarm_panel`` 组件 props（证明告警面板已配置化）； +2. **树脂模板同样可配置**：换一份配置资产（不同配色 / 规则源）即得到不同 props， + 前端代码零改动（PRD 5.5 验收口径「切换模板零改码」）； +3. **配置点真实生效**：``mute_lower`` 改变 → ``filter_alarms_by_mute`` 过滤结果 + 改变；``require_ack`` 关闭 → props 里不再出现 ``ackTimeoutS``； +4. **负例**：一份故意破坏的配置（缺 P0 配色 / 非法 severity / 缺 rulesSource） + 被正确拒绝并聚合多条错误（对齐 #50 校验器风格）。 + +用法（在 core/cockpit 目录下）： + python scripts/verify_alarm_config.py +退出码：0 = 全部通过；1 = 存在未达标项。 + +说明：Ti 模板资产本身是 YAML；本脚本不依赖 PyYAML（避免引入运行时依赖）， +而是用与该 YAML 文件内容等价的 dict 进行校验——字段语义完全一致， +若安装了 PyYAML，可直接解析原文件复现（见文末注释）。 +""" +from __future__ import annotations + +import os +import sys + +# 本脚本位于 core/cockpit/scripts/，需要把 core/ 加入 sys.path， +# 才能 `from cockpit import ...`（cockpit 包位于 core/cockpit） +sys.path.insert( + 0, + os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), +) + +from cockpit import ( # noqa: E402 + ALARM_CONFIG_SCHEMA_ID, + filter_alarms_by_mute, + load_alarm_config, + render_alarm_panel_props, + validate_alarm_config, +) + + +def _ti_config() -> dict: + """与 templates/ti-cl4/dashboard/alarm_panel.ti.yaml 等价的 dict。""" + return { + "$schema": ALARM_CONFIG_SCHEMA_ID, + "severityColors": [ + {"severity": "P0", "fg": "#ff3b30", "bg": "rgba(255,59,48,0.12)", + "icon": "alert-octagon", "border": "#ff3b30"}, + {"severity": "P1", "fg": "#f5a623", "bg": "rgba(245,166,35,0.12)", + "icon": "alert-triangle", "border": "#f5a623"}, + {"severity": "P2", "fg": "#3aa0ff", "bg": "rgba(58,160,255,0.10)", + "icon": "info"}, + ], + "rulesSource": { + "kind": "asset", + "ref": "ti-cl4/impurity-forecast/config/alert_rules.template.yaml", + }, + "thresholdsSource": { + "kind": "asset", + "ref": "ti-cl4/impurity-forecast/config/alert_rules.template.yaml", + }, + "showSop": True, + "requireAck": True, + "ackTimeoutS": 300, + "muteLower": "P2", + "groupBy": "severity", + "sortBy": "severity", + "maxItems": 50, + } + + +def _resin_config() -> dict: + """树脂行业模板的等价配置（不同配色 + 内联规则，证明换模板零改码）。""" + return { + "severityColors": [ + {"severity": "P0", "fg": "#e63329", "bg": "rgba(230,51,41,0.12)"}, + {"severity": "P1", "fg": "#ffb000", "bg": "rgba(255,176,0,0.12)"}, + {"severity": "P2", "fg": "#2bb673", "bg": "rgba(43,182,115,0.10)"}, + ], + "rulesSource": { + "kind": "inline", + "data": [ + {"id": "resin-temp-high", "severity": "P0"}, + {"id": "resin-crosslink-low", "severity": "P1"}, + ], + }, + "showSop": False, + "requireAck": False, + "muteLower": "P1", # 树脂模板只看 P0/P1，静默 P2 提示 + } + + +def _check(name: str, fn) -> bool: + try: + ok = fn() + except Exception as exc: # noqa: BLE001 + print(f"[FAIL] {name}: 抛异常 {exc!r}") + return False + status = "PASS" if ok else "FAIL" + print(f"[{status}] {name}") + return bool(ok) + + +def main() -> int: + print("== 报警看板配置化验收（issue #52 / PRD 5.5）==") + all_ok = True + + # 1) Ti 模板资产合法 + 可渲染为 props + def check_ti_valid() -> bool: + res = validate_alarm_config(_ti_config()) + if not res.ok: + for e in res.errors: + print(f" - {e}") + return False + cfg = load_alarm_config(_ti_config()) + props = render_alarm_panel_props(cfg) + # P0 红色告警配色进入 props；requireAck=True 携带 ackTimeoutS + return ( + props["severityStyles"]["P0"]["fg"] == "#ff3b30" + and props["requireAck"] is True + and props.get("ackTimeoutS") == 300 + and props["rulesSource"]["ref"].endswith("alert_rules.template.yaml") + ) + + all_ok &= _check("Ti 模板配置合法 + 渲染 props 正确", check_ti_valid) + + # 2) 树脂模板换配置 → props 不同，前端代码零改动 + def check_resin_differs() -> bool: + cfg = load_alarm_config(_resin_config()) + props = render_alarm_panel_props(cfg) + # 树脂 P0 用不同红色；内联规则透传 data；requireAck 关闭不带 ackTimeoutS + return ( + props["severityStyles"]["P0"]["fg"] == "#e63329" + and props["rulesSource"]["kind"] == "inline" + and len(props["rulesSource"]["data"]) == 2 + and "ackTimeoutS" not in props + and props["showSop"] is False + ) + + all_ok &= _check("树脂模板换配置 → props 随之变化（切换模板零改码）", check_resin_differs) + + # 3) mute_lower 配置点真实驱动展示策略 + def check_mute_lower() -> bool: + alarms = [ + {"severity": "P2", "id": "a"}, + {"severity": "P0", "id": "b"}, + {"severity": "P1", "id": "c"}, + ] + keep_all = filter_alarms_by_mute(alarms, "P2") + only_p0_p1 = filter_alarms_by_mute(alarms, "P1") + # mute=P2 全保留且按严重度降序；mute=P1 过滤掉 P2 + return ( + [a["id"] for a in keep_all] == ["b", "c", "a"] + and [a["id"] for a in only_p0_p1] == ["b", "c"] + ) + + all_ok &= _check("muteLower 配置点真实过滤 + 排序告警", check_mute_lower) + + # 4) 负例：破坏的配置被拒绝并聚合多条错误 + def check_negative() -> bool: + bad = { + "severityColors": [ + {"severity": "P0", "fg": "#f00", "bg": "#000"}, + # 缺 P1 / P2 + {"severity": "P0", "fg": "#f00", "bg": "#000"}, # 重复 P0 + {"severity": "P9", "fg": "#f00", "bg": "#000"}, # 非法 severity + ], + "rulesSource": {"kind": "asset"}, # 缺 ref + "ackTimeoutS": 0, # 非正整数 + "sortBy": "color", # 非法排序键 + "maxItems": 0, # 越界 + } + res = validate_alarm_config(bad) + # 应收集到多条错误（聚合而非首条即返） + return (not res.ok) and len(res.errors) >= 5 + + all_ok &= _check("负例：破坏的配置被拒绝并聚合多条错误", check_negative) + + print("=" * 48) + print("结果: " + ("全部通过 ✅" if all_ok else "存在未达标项 ❌")) + return 0 if all_ok else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/core/cockpit/tests/test_alarm_config.py b/core/cockpit/tests/test_alarm_config.py new file mode 100644 index 0000000..99b9398 --- /dev/null +++ b/core/cockpit/tests/test_alarm_config.py @@ -0,0 +1,338 @@ +# -*- coding: utf-8 -*- +"""报警看板配置化单元测试（issue #52 / PRD 5.5「⑤ 配置化驾驶舱」）。 + +覆盖： +1. 合法配置（Ti 模板）→ 校验通过、可解析为 ``AlarmPanelConfig``、序列化往返一致； +2. 默认值与字段别名（snake_case / camelCase 都接受）、thresholds_source 选填； +3. ``render_alarm_panel_props`` 正确编译 severity 样式 / 规则源 / 确认策略； +4. ``filter_alarms_by_mute`` 静默下限 + severity 降序排序； +5. 各类非法情况：非对象、错误 $schema、severityColors 缺级/重复/非法、 + rulesSource 缺失/asset 缺 ref/inline 缺 data、布尔/整数/枚举/区间非法； +6. ``load_alarm_config`` 校验失败抛 ``AlarmConfigError`` 并携带全部错误。 +""" +from __future__ import annotations + +import copy +import unittest + +from cockpit import ( # type: ignore[import-not-found] + ALARM_CONFIG_SCHEMA_ID, + DEFAULT_ACK_TIMEOUT_S, + DEFAULT_MAX_ITEMS, + AlarmConfigError, + AlarmPanelConfig, + filter_alarms_by_mute, + load_alarm_config, + render_alarm_panel_props, + validate_alarm_config, +) + + +def _valid_config() -> dict: + """返回一份合法的告警面板配置（Ti 模板风格）。""" + return { + "$schema": ALARM_CONFIG_SCHEMA_ID, + "severityColors": [ + {"severity": "P0", "fg": "#ff3b30", "bg": "rgba(255,59,48,0.12)", + "icon": "alert-octagon", "border": "#ff3b30"}, + {"severity": "P1", "fg": "#f5a623", "bg": "rgba(245,166,35,0.12)"}, + {"severity": "P2", "fg": "#3aa0ff", "bg": "rgba(58,160,255,0.10)"}, + ], + "rulesSource": {"kind": "asset", "ref": "ti-cl4/cfg/alert_rules.yaml"}, + "showSop": True, + "requireAck": True, + "ackTimeoutS": 300, + "muteLower": "P2", + "groupBy": "severity", + "sortBy": "severity", + "maxItems": 50, + } + + +class ValidateTest(unittest.TestCase): + def test_valid_config_passes(self): + res = validate_alarm_config(_valid_config()) + self.assertTrue(res.ok, msg=res.errors) + self.assertEqual(res.errors, []) + norm = res.normalized + assert norm is not None + # 规范化：severityColors 按 P0/P1/P2 顺序回填 + self.assertEqual([c["severity"] for c in norm["severityColors"]], ["P0", "P1", "P2"]) + # P0 的可选 icon/border 保留 + self.assertEqual(norm["severityColors"][0]["icon"], "alert-octagon") + + def test_non_dict_rejected(self): + res = validate_alarm_config(["not", "a", "dict"]) + self.assertFalse(res.ok) + self.assertIn("对象（dict）", res.errors[0]) + + def test_wrong_schema_rejected(self): + cfg = _valid_config() + cfg["$schema"] = "something-else" + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("$schema" in e for e in res.errors)) + + def test_snake_case_aliases_accepted(self): + # snake_case 字段名同样接受（向后兼容手写 YAML） + cfg = { + "severity_colors": [ + {"severity": "P0", "fg": "#f", "bg": "#0"}, + {"severity": "P1", "fg": "#f", "bg": "#0"}, + {"severity": "P2", "fg": "#f", "bg": "#0"}, + ], + "rules_source": {"kind": "asset", "ref": "a.yaml"}, + "show_sop": False, + "require_ack": False, + "mute_lower": "P2", + "group_by": "severity", + "sort_by": "severity", + "max_items": 10, + } + res = validate_alarm_config(cfg) + self.assertTrue(res.ok, msg=res.errors) + + def test_missing_severity_level_rejected(self): + cfg = _valid_config() + cfg["severityColors"] = cfg["severityColors"][:2] # 只留 P0/P1 + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("缺少 P2" in e for e in res.errors)) + + def test_duplicate_and_invalid_severity_rejected(self): + cfg = _valid_config() + cfg["severityColors"] = [ + {"severity": "P0", "fg": "#f", "bg": "#0"}, + {"severity": "P0", "fg": "#f", "bg": "#0"}, # 重复 + {"severity": "P9", "fg": "#f", "bg": "#0"}, # 非法 + ] + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + joined = "; ".join(res.errors) + self.assertIn("重复", joined) + self.assertIn("P9", joined) + + def test_missing_required_color_key_rejected(self): + cfg = _valid_config() + cfg["severityColors"][0] = {"severity": "P0", "fg": "#f"} # 缺 bg + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("bg" in e for e in res.errors)) + + def test_missing_rules_source_rejected(self): + cfg = _valid_config() + del cfg["rulesSource"] + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("rulesSource" in e for e in res.errors)) + + def test_asset_source_without_ref_rejected(self): + cfg = _valid_config() + cfg["rulesSource"] = {"kind": "asset"} # 缺 ref + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("ref" in e for e in res.errors)) + + def test_inline_source_without_data_rejected(self): + cfg = _valid_config() + cfg["rulesSource"] = {"kind": "inline"} # 缺 data + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("data" in e for e in res.errors)) + + def test_inline_source_item_without_id_rejected(self): + cfg = _valid_config() + cfg["rulesSource"] = {"kind": "inline", "data": [{"severity": "P0"}]} # 缺 id + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("id" in e for e in res.errors)) + + def test_invalid_source_kind_rejected(self): + cfg = _valid_config() + cfg["rulesSource"] = {"kind": "graphql", "ref": "x"} + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("kind" in e for e in res.errors)) + + def test_bool_fields_type_checked(self): + for field in ("showSop", "requireAck"): + cfg = _valid_config() + cfg[field] = "yes" + res = validate_alarm_config(cfg) + self.assertFalse(res.ok, msg=field) + self.assertTrue(any(field in e for e in res.errors), msg=field) + + def test_ack_timeout_must_be_positive_int(self): + cfg = _valid_config() + cfg["ackTimeoutS"] = 0 + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + self.assertTrue(any("ackTimeoutS" in e for e in res.errors)) + # bool 不应被当成 int 接受 + cfg["ackTimeoutS"] = True + res = validate_alarm_config(cfg) + self.assertFalse(res.ok) + + def test_enum_fields_validated(self): + for field, valid, invalid in [ + ("muteLower", ["P0", "P1", "P2"], "P9"), + ("groupBy", ["severity", "rule", "process"], "color"), + ("sortBy", ["severity", "time"], "color"), + ]: + for v in valid: + cfg = _valid_config() + cfg[field] = v + # muteLower=P0 合法但会过滤掉 P1/P2，不影响校验合法性 + self.assertTrue(validate_alarm_config(cfg).ok, msg=f"{field}={v}") + cfg = _valid_config() + cfg[field] = invalid + res = validate_alarm_config(cfg) + self.assertFalse(res.ok, msg=f"{field}={invalid}") + self.assertTrue(any(field in e for e in res.errors), msg=field) + + def test_max_items_range(self): + cfg = _valid_config() + cfg["maxItems"] = 0 + self.assertFalse(validate_alarm_config(cfg).ok) + cfg["maxItems"] = 501 + self.assertFalse(validate_alarm_config(cfg).ok) + cfg["maxItems"] = 500 + self.assertTrue(validate_alarm_config(cfg).ok) + cfg["maxItems"] = True # bool 不被当成 int + self.assertFalse(validate_alarm_config(cfg).ok) + + def test_defaults_applied_when_optional_omitted(self): + cfg = { + "severityColors": [ + {"severity": "P0", "fg": "#f", "bg": "#0"}, + {"severity": "P1", "fg": "#f", "bg": "#0"}, + {"severity": "P2", "fg": "#f", "bg": "#0"}, + ], + "rulesSource": {"kind": "asset", "ref": "a.yaml"}, + } + res = validate_alarm_config(cfg) + self.assertTrue(res.ok, msg=res.errors) + assert res.normalized is not None + self.assertEqual(res.normalized["showSop"], True) + self.assertEqual(res.normalized["requireAck"], False) + self.assertEqual(res.normalized["ackTimeoutS"], DEFAULT_ACK_TIMEOUT_S) + self.assertEqual(res.normalized["muteLower"], "P2") + self.assertEqual(res.normalized["groupBy"], "severity") + self.assertEqual(res.normalized["sortBy"], "severity") + self.assertEqual(res.normalized["maxItems"], DEFAULT_MAX_ITEMS) + # thresholds_source 选填，未给出则 normalized 不含它 + self.assertNotIn("thresholdsSource", res.normalized) + + def test_multiple_errors_aggregated(self): + # 一次性破坏多个字段，校验器应聚合多条错误而非首条即返 + bad = { + "severityColors": [{"severity": "P0", "fg": "#f", "bg": "#0"}], # 缺 P1/P2 + "rulesSource": {"kind": "asset"}, # 缺 ref + "ackTimeoutS": -1, + "sortBy": "color", + } + res = validate_alarm_config(bad) + self.assertFalse(res.ok) + self.assertGreaterEqual(len(res.errors), 4) + + +class LoadTest(unittest.TestCase): + def test_load_valid_returns_model(self): + cfg = load_alarm_config(_valid_config()) + self.assertIsInstance(cfg, AlarmPanelConfig) + self.assertEqual(cfg.schema, ALARM_CONFIG_SCHEMA_ID) + self.assertEqual(len(cfg.severity_colors), 3) + self.assertEqual(cfg.rules_source.kind, "asset") + self.assertEqual(cfg.rules_source.ref, "ti-cl4/cfg/alert_rules.yaml") + self.assertTrue(cfg.require_ack) + + def test_load_roundtrip(self): + original = _valid_config() + cfg = load_alarm_config(original) + # 内存模型序列化后再次校验仍合法（往返一致） + res = validate_alarm_config(cfg.to_dict()) + self.assertTrue(res.ok, msg=res.errors) + + def test_load_invalid_raises_with_errors(self): + bad = copy.deepcopy(_valid_config()) + del bad["rulesSource"] + with self.assertRaises(AlarmConfigError) as cm: + load_alarm_config(bad) + # 异常携带全部错误清单 + self.assertGreaterEqual(len(cm.exception.errors), 1) + self.assertTrue(any("rulesSource" in e for e in cm.exception.errors)) + + +class RenderPropsTest(unittest.TestCase): + def test_props_contain_severity_styles_and_sources(self): + cfg = load_alarm_config(_valid_config()) + props = render_alarm_panel_props(cfg) + # 保留 #51 的订阅语义（向后兼容） + self.assertEqual(props["subscribe"], "alarm_stream") + # severity 样式：P0 红色 + 可选 border + self.assertEqual(props["severityStyles"]["P0"]["fg"], "#ff3b30") + self.assertEqual(props["severityStyles"]["P0"]["border"], "#ff3b30") + # asset 规则源透传 ref + self.assertEqual(props["rulesSource"]["kind"], "asset") + self.assertTrue(props["rulesSource"]["ref"].endswith("alert_rules.yaml")) + # require_ack=True 携带 ackTimeoutS + self.assertEqual(props["ackTimeoutS"], 300) + self.assertTrue(props["showSop"]) + + def test_props_ack_omitted_when_not_required(self): + cfg = _valid_config() + cfg["requireAck"] = False + props = render_alarm_panel_props(load_alarm_config(cfg)) + self.assertNotIn("ackTimeoutS", props) + self.assertFalse(props["requireAck"]) + + def test_props_inline_source_data_passed_through(self): + cfg = _valid_config() + cfg["rulesSource"] = { + "kind": "inline", + "data": [{"id": "r1", "severity": "P0"}, {"id": "r2", "severity": "P1"}], + } + props = render_alarm_panel_props(load_alarm_config(cfg)) + self.assertEqual(props["rulesSource"]["kind"], "inline") + self.assertEqual(len(props["rulesSource"]["data"]), 2) + + def test_props_thresholds_source_optional(self): + # 未声明 thresholds_source → props 不含该键 + props = render_alarm_panel_props(load_alarm_config(_valid_config())) + self.assertNotIn("thresholdsSource", props) + # 声明后 → 透传 + cfg = _valid_config() + cfg["thresholdsSource"] = {"kind": "asset", "ref": "thr.yaml"} + props = render_alarm_panel_props(load_alarm_config(cfg)) + self.assertEqual(props["thresholdsSource"]["ref"], "thr.yaml") + + +class FilterMuteTest(unittest.TestCase): + def test_filter_keeps_above_floor_and_sorts_desc(self): + alarms = [ + {"severity": "P2", "id": "a"}, + {"severity": "P0", "id": "b"}, + {"severity": "P1", "id": "c"}, + ] + # mute=P2：全保留，按严重度降序 + kept = filter_alarms_by_mute(alarms, "P2") + self.assertEqual([a["id"] for a in kept], ["b", "c", "a"]) + # mute=P1：滤掉 P2 + kept = filter_alarms_by_mute(alarms, "P1") + self.assertEqual([a["id"] for a in kept], ["b", "c"]) + # mute=P0：只留 P0 + kept = filter_alarms_by_mute(alarms, "P0") + self.assertEqual([a["id"] for a in kept], ["b"]) + + def test_filter_ignores_unknown_severity(self): + alarms = [{"severity": "P9", "id": "x"}, {"severity": "P0", "id": "y"}] + kept = filter_alarms_by_mute(alarms, "P2") + self.assertEqual([a["id"] for a in kept], ["y"]) + + def test_filter_invalid_mute_raises(self): + with self.assertRaises(AlarmConfigError): + filter_alarms_by_mute([], "P9") + + +if __name__ == "__main__": + unittest.main() diff --git a/templates/ti-cl4/dashboard/alarm_panel.ti.yaml b/templates/ti-cl4/dashboard/alarm_panel.ti.yaml new file mode 100644 index 0000000..ea712ea --- /dev/null +++ b/templates/ti-cl4/dashboard/alarm_panel.ti.yaml @@ -0,0 +1,50 @@ +# -*- coding: utf-8 -*- +# 氯化车间/海绵钛驾驶舱 · 告警面板配置资产（iAOP-Template-Ti 一期）。 +# +# 对齐 PRD 5.5「⑤ 配置化驾驶舱」+ 5.3 ③ 场景A（异常检测 → 驾驶舱红色告警 + +# LLM 生成"原因+处置建议" → 值班长确认）：把告警面板的「严重度配色 / 规则源 / +# SOP 联动 / 确认与静默 / 分组排序」外置为模板配置（issue #52）。 +# +# 切换到其他行业模板（如树脂）= 换一份本文件，告警面板前端代码零改动 +# （PRD 5.5 验收口径「切换模板零改码」覆盖到告警面板）。 +# +# $schema 对应 core/cockpit/alarm_config.py 的 ALARM_CONFIG_SCHEMA_ID。 +$schema: iAOP-cockpit-alarm-panel-v1 +# severityColors：P0/P1/P2 三级展示配色（覆盖主题默认告警色）。 +# P0=红色（驾驶舱红色告警，PRD 场景A）、P1=黄色（加强监控）、P2=蓝色（提示）。 +severityColors: + - severity: P0 + fg: "#ff3b30" # 红色前景（告警文本 / 图标） + bg: "rgba(255,59,48,0.12)" + icon: alert-octagon + border: "#ff3b30" + - severity: P1 + fg: "#f5a623" # 黄色 + bg: "rgba(245,166,35,0.12)" + icon: alert-triangle + border: "#f5a623" + - severity: P2 + fg: "#3aa0ff" # 蓝色（提示） + bg: "rgba(58,160,255,0.10)" + icon: info +# rulesSource：告警面板订阅哪份规则资产（阈值/规则外置）。 +# 指向 issue #72 落地的炉层杂质预警规则模板（templates/ti-cl4/impurity-forecast/ +# config/alert_rules.template.yaml，在其 feature 分支上推进；此处先按规划路径绑定）。 +rulesSource: + kind: asset + ref: ti-cl4/impurity-forecast/config/alert_rules.template.yaml +# thresholdsSource：阈值可与规则同源，这里显式独立声明，便于配置台单独维护。 +thresholdsSource: + kind: asset + ref: ti-cl4/impurity-forecast/config/alert_rules.template.yaml +# SOP 联动：在面板里展开处置 SOP（PRD 场景A：LLM 报警解释 + 值班长确认）。 +showSop: true +# 关键告警强制人工确认（PRD line 333：高利害人工确认，不直接联动执行机构）。 +requireAck: true +ackTimeoutS: 300 # 5 分钟未确认自动升级 severity +# 静默下限：低于 P2 的不展示（Ti 一期 P0/P1/P2 全展示）。 +muteLower: P2 +# 分组 / 排序 / 最大条数：大屏展示策略外置。 +groupBy: severity +sortBy: severity +maxItems: 50 -- 2.54.0