330 lines
14 KiB
Python
330 lines
14 KiB
Python
# -*- 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)
|