# -*- coding: utf-8 -*- """⑤.7 配置预览渲染引擎 —— issue #65 / PRD ⑤.7。 配置台让实施工程师"边配边看"：改完布局/告警/查询配置后，立即在预览区看到 驾驶舱会变成什么样、告警会怎么触发、NL 查询会怎么响应——**不必发布到生产 就能确认效果**。本模块是预览区的渲染后端，把 ``ConfigStore`` 里的配置渲染 为**结构化的预览片段**（dict/JSON），对齐 ``iAOP-cockpit-layout-v1`` 的 widget 类型与既有驾驶舱资产（resin cockpit）。 三类预览： - **布局预览**（``render_layout_preview``）：把 layout widget 列表渲染为 带占位网格坐标的 widget 描述（type/bind/metric/description + x/y/w/h）， 计算网格占用率（发现越界/重叠）； - **告警预览**（``render_alarm_preview``）：把告警规则（point + 阈值 + 级别） 渲染为"当 X 超过 Y 时，触发 级别 告警"的可读条目 + 模拟评估（给定当前值 是否触发）； - **NL 查询预览**（``render_nl_query_preview``）：把 NL 查询模板渲染为示例 问答对（模板 × 示例槽位 → 渲染后的问句 + 预期数据来源）。 预览是**只读、无副作用**的——只读配置、产出结构化输出，不改任何状态， 对齐 PRD「预演不污染生产」。 零运行时依赖：仅用 dataclass / Enum / 标准库。 """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional from .config_store import ALLOWED_WIDGET_TYPES, ConfigKind, ConfigStore # --------------------------------------------------------------------------- # 预览模型 # --------------------------------------------------------------------------- class PreviewKind(str, Enum): """三类预览。""" LAYOUT = "layout" ALARM = "alarm" NL_QUERY = "nl_query" @dataclass class PreviewResult: """一次预览渲染的结果（结构化片段 + 说明 + 问题提示）。""" kind: PreviewKind title: str items: List[Dict[str, Any]] = field(default_factory=list) # 渲染后的条目 notes: List[str] = field(default_factory=list) # 说明 / 渲染提示 warnings: List[str] = field(default_factory=list) # 布局越界/重叠等 reason: str = "" # 本次预览的来源说明 @property def ok(self) -> bool: return not self.warnings def to_dict(self) -> dict: return { "kind": self.kind.value, "title": self.title, "items": self.items, "notes": self.notes, "warnings": self.warnings, "reason": self.reason, "ok": self.ok, } # --------------------------------------------------------------------------- # 布局预览 # --------------------------------------------------------------------------- #: 驾驶舱网格规格（对齐 resin cockpit：12 列 × 若干行，w/h 以网格单元计） GRID_COLUMNS = 12 def render_layout_preview( widgets: List[Dict[str, Any]], title: str = "驾驶舱布局预览", grid_columns: int = GRID_COLUMNS, ) -> PreviewResult: """渲染布局 widget 列表为预览片段。 每个 widget 渲染为带 type/描述/坐标的卡片；同时做**布局体检**： - 越界（x+w 超出列数 / y+h 超出合理行数）； - 重叠（两个 widget 矩形相交）； - 非法类型（不在 ``ALLOWED_WIDGET_TYPES``）。 """ result = PreviewResult(kind=PreviewKind.LAYOUT, title=title, reason=f"渲染 {len(widgets)} 个 widget") seen_rects: List[Dict[str, int]] = [] for i, w in enumerate(widgets): wt = w.get("type") x, y = w.get("x", 0), w.get("y", 0) ww, hh = w.get("w", 0), w.get("h", 0) card: Dict[str, Any] = { "index": i, "type": wt, "x": x, "y": y, "w": ww, "h": hh, "description": w.get("description", ""), } # 携带业务绑定（trend.bind / kpi_card.metric） if wt == "trend": card["bind"] = w.get("bind", "") elif wt == "kpi_card": card["metric"] = w.get("metric", "") card["label"] = w.get("label", "") elif wt == "process_view": card["src"] = w.get("src", "") result.items.append(card) # 体检：类型合法 if wt not in ALLOWED_WIDGET_TYPES: result.warnings.append(f"widget[{i}] 非法类型 '{wt}'") # 越界 if x < 0 or y < 0 or ww <= 0 or hh <= 0: result.warnings.append(f"widget[{i}] 坐标/尺寸非法 ({x},{y},{ww},{hh})") elif x + ww > grid_columns: result.warnings.append( f"widget[{i}] 越界：x+w={x + ww} > {grid_columns} 列") else: # 重叠检测（矩形相交） rect = {"x": x, "y": y, "w": ww, "h": hh} for j, prev in enumerate(seen_rects): if _rects_overlap(rect, prev): result.warnings.append(f"widget[{i}] 与 widget[{j}] 重叠") seen_rects.append(rect) # 网格占用率 total_area = sum(r["w"] * r["h"] for r in seen_rects) max_row = max((r["y"] + r["h"] for r in seen_rects), default=0) grid_area = grid_columns * max(max_row, 1) usage = round(total_area / grid_area * 100, 1) if grid_area else 0.0 result.notes.append(f"网格占用率 {usage}%（{grid_columns} 列，最大 {max_row} 行）") return result def _rects_overlap(a: Dict[str, int], b: Dict[str, int]) -> bool: """两个网格矩形是否相交（不含边界共享视为不重叠）。""" ax2, ay2 = a["x"] + a["w"], a["y"] + a["h"] bx2, by2 = b["x"] + b["w"], b["y"] + b["h"] return not (ax2 <= b["x"] or bx2 <= a["x"] or ay2 <= b["y"] or by2 <= a["y"]) # --------------------------------------------------------------------------- # 告警预览 # --------------------------------------------------------------------------- #: 合法的告警级别（对齐 cockpit alarm_panel） ALARM_LEVELS = {"info", "warn", "critical"} @dataclass class AlarmRule: """一条告警规则（供告警预览渲染与模拟评估）。""" point_id: str # 关联测点 metric: str # 指标名（展示用） operator: str # 比较运算符 > / >= / < / <= / == threshold: float # 阈值 level: str = "warn" # 告警级别 info/warn/critical message: str = "" # 告警文案模板（可含 {value}） def evaluate(self, value: float) -> bool: """给定当前值，判断是否触发告警。""" ops = { ">": value > self.threshold, ">=": value >= self.threshold, "<": value < self.threshold, "<=": value <= self.threshold, "==": value == self.threshold, } return ops.get(self.operator, False) def render_alarm_preview( rules: List[AlarmRule], current_values: Optional[Dict[str, float]] = None, title: str = "告警规则预览", ) -> PreviewResult: """渲染告警规则为可读条目，并用当前值模拟触发评估。 Args: rules: 告警规则列表； current_values: 当前测点值（point_id → value），用于模拟评估； 不提供则只渲染规则、不做触发评估。 """ result = PreviewResult(kind=PreviewKind.ALARM, title=title, reason=f"渲染 {len(rules)} 条告警规则") for r in rules: if r.level not in ALARM_LEVELS: result.warnings.append(f"告警 '{r.point_id}' 非法级别 '{r.level}'") if r.operator not in (">", ">=", "<", "<=", "=="): result.warnings.append(f"告警 '{r.point_id}' 非法运算符 '{r.operator}'") text = (f"当 {r.metric}({r.point_id}) {r.operator} {r.threshold} 时，" f"触发 [{r.level}] 告警") entry: Dict[str, Any] = { "point_id": r.point_id, "metric": r.metric, "operator": r.operator, "threshold": r.threshold, "level": r.level, "text": text, } if current_values is not None and r.point_id in current_values: val = current_values[r.point_id] triggered = r.evaluate(val) entry["current_value"] = val entry["triggered"] = triggered entry["state"] = "触发" if triggered else "正常" result.items.append(entry) if current_values is not None: triggered_count = sum(1 for e in result.items if e.get("triggered")) result.notes.append(f"模拟评估：{triggered_count}/{len(rules)} 条触发") return result # --------------------------------------------------------------------------- # NL 查询预览 # --------------------------------------------------------------------------- @dataclass class NLQueryTemplate: """一条 NL 查询模板（供 NL 查询预览渲染）。""" name: str # 模板名 question_template: str # 问句模板（含 {slot} 占位） slots: Dict[str, List[str]] # 槽位 → 候选取值（用于生成示例问句） data_source: str = "" # 预期数据来源（如 tdengine/rag） answer_hint: str = "" # 预期答案提示 def render_examples(self, max_per_slot: int = 2) -> List[str]: """用槽位候选值生成示例问句（笛卡尔积，限量）。""" if not self.slots: return [self.question_template] examples: List[str] = [] # 取每个槽位前 N 个候选，做限量笛卡尔积 first_slot = next(iter(self.slots)) for val in self.slots[first_slot][:max_per_slot]: examples.append(self.question_template.replace("{" + first_slot + "}", val)) if not examples: examples.append(self.question_template) return examples def render_nl_query_preview( templates: List[NLQueryTemplate], title: str = "NL 查询模板预览", ) -> PreviewResult: """渲染 NL 查询模板为示例问答对。""" result = PreviewResult(kind=PreviewKind.NL_QUERY, title=title, reason=f"渲染 {len(templates)} 个查询模板") for t in templates: examples = t.render_examples() entry: Dict[str, Any] = { "name": t.name, "data_source": t.data_source, "answer_hint": t.answer_hint, "examples": examples, } result.items.append(entry) if not t.question_template: result.warnings.append(f"模板 '{t.name}' 问句模板为空") result.notes.append(f"共生成 {sum(len(e['examples']) for e in result.items)} 条示例问句") return result # --------------------------------------------------------------------------- # 从 ConfigStore 一键预览 # --------------------------------------------------------------------------- def preview_from_store( store: ConfigStore, kind: PreviewKind = PreviewKind.LAYOUT, current_values: Optional[Dict[str, float]] = None, ) -> PreviewResult: """从 ConfigStore 读取配置并渲染对应预览（配置台预览区入口）。 - LAYOUT：读 ``layout`` 类目下 key 含 'dashboard' 的 widget 列表； - ALARM：读 ``model_param`` 类目下 key 以 'alarm_' 开头的规则； - NL_QUERY：读 ``rag_config`` 类目下 key 以 'nl_' 开头的模板。 配置缺失时返回空结果（含提示），不报错——预览是只读的、宽容的。 """ if kind == PreviewKind.LAYOUT: widgets: List[Dict[str, Any]] = [] for it in store.list(ConfigKind.LAYOUT): if isinstance(it.value, list): widgets.extend(it.value) if not widgets: return PreviewResult(kind=kind, title="布局预览（空）", notes=["未配置布局 widget，请在布局编辑页添加"]) return render_layout_preview(widgets) if kind == PreviewKind.ALARM: rules: List[AlarmRule] = [] for it in store.list(ConfigKind.MODEL_PARAM): if it.key.startswith("alarm_") and isinstance(it.value, dict): rules.append(AlarmRule( point_id=it.value.get("point_id", ""), metric=it.value.get("metric", ""), operator=it.value.get("operator", ">"), threshold=float(it.value.get("threshold", 0)), level=it.value.get("level", "warn"), message=it.value.get("message", ""), )) if not rules: return PreviewResult(kind=kind, title="告警预览（空）", notes=["未配置告警规则，请在告警编辑页添加"]) return render_alarm_preview(rules, current_values=current_values) # NL_QUERY templates: List[NLQueryTemplate] = [] for it in store.list(ConfigKind.RAG_CONFIG): if it.key.startswith("nl_") and isinstance(it.value, dict): templates.append(NLQueryTemplate( name=it.value.get("name", it.key), question_template=it.value.get("question_template", ""), slots=it.value.get("slots", {}), data_source=it.value.get("data_source", ""), answer_hint=it.value.get("answer_hint", ""), )) if not templates: return PreviewResult(kind=kind, title="NL 查询预览（空）", notes=["未配置 NL 查询模板，请在查询编辑页添加"]) return render_nl_query_preview(templates)