From 0660a4b256ee9cf604f6754d356910750a94e525 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:31:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(#65):=20=E9=85=8D=E7=BD=AE=E9=A2=84?= =?UTF-8?q?=E8=A7=88=E6=B8=B2=E6=9F=93=E5=BC=95=E6=93=8E=EF=BC=88=E5=B8=83?= =?UTF-8?q?=E5=B1=80/=E5=91=8A=E8=AD=A6/NL=E6=9F=A5=E8=AF=A2=EF=BC=8C?= =?UTF-8?q?=E5=AF=B9=E9=BD=90=20cockpit-layout-v1=20widget=20=E7=B1=BB?= =?UTF-8?q?=E5=9E=8B=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/template-console/config_store.py | 11 +- core/template-console/preview.py | 329 ++++++++++++++++++++ core/template-console/tests/test_preview.py | 188 +++++++++++ 3 files changed, 523 insertions(+), 5 deletions(-) create mode 100644 core/template-console/preview.py create mode 100644 core/template-console/tests/test_preview.py diff --git a/core/template-console/config_store.py b/core/template-console/config_store.py index de513af..e00d3f5 100644 --- a/core/template-console/config_store.py +++ b/core/template-console/config_store.py @@ -114,7 +114,8 @@ def validate_item(kind: ConfigKind, key: str, value: Any) -> ValidationResult: 校验规则（配置台 upsert 前置门禁，防止坏数据进快照）： - 通用：key 非空、匹配 ``[a-z0-9_.-]+``； - - model_param：value 为标量（int/float/bool/str）或标量列表； + - model_param：value 为标量（int/float/bool/str）、标量列表，或结构化 dict + （如 optimizer 配置 / 告警规则等复合超参）； - rag_config：top_k 为 1~50 的正整数、similarity_threshold 为 0~1 浮点、 sources 为非空字符串列表； - layout：value 为 widget 列表，每个 widget 有合法 type 与 x/y/w/h。 @@ -126,11 +127,11 @@ def validate_item(kind: ConfigKind, key: str, value: Any) -> ValidationResult: errors.append(f"key '{key}' 仅允许小写字母/数字/._-") if kind == ConfigKind.MODEL_PARAM: - if not isinstance(value, (int, float, bool, str, list)): - errors.append("model_param 的 value 必须为标量或标量列表") + if not isinstance(value, (int, float, bool, str, list, dict)): + errors.append("model_param 的 value 必须为标量/标量列表/结构化对象") elif isinstance(value, list) and any( - not isinstance(v, (int, float, bool, str)) for v in value): - errors.append("model_param 列表 value 仅允许标量元素") + not isinstance(v, (int, float, bool, str, dict)) for v in value): + errors.append("model_param 列表 value 仅允许标量或对象元素") # 常见超参范围提示（软约束，仅对已知键） if key == "learning_rate" and isinstance(value, (int, float)): if not (0 < value < 1): diff --git a/core/template-console/preview.py b/core/template-console/preview.py new file mode 100644 index 0000000..e92df8c --- /dev/null +++ b/core/template-console/preview.py @@ -0,0 +1,329 @@ +# -*- 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) diff --git a/core/template-console/tests/test_preview.py b/core/template-console/tests/test_preview.py new file mode 100644 index 0000000..dd4f9a5 --- /dev/null +++ b/core/template-console/tests/test_preview.py @@ -0,0 +1,188 @@ +# -*- coding: utf-8 -*- +"""配置预览渲染引擎测试（issue #65）。 + +覆盖： +1. 布局预览（widget 卡片 + 网格占用率 + 越界/重叠检测）； +2. 告警预览（规则渲染 + 模拟触发评估）； +3. NL 查询预览（模板 + 示例问句生成）； +4. 从 ConfigStore 一键预览（含空配置的宽容处理）； +5. PreviewResult 的 ok/to_dict。 +""" +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from template_console.config_store import ConfigKind, ConfigStore # noqa: E402 +from template_console.preview import ( # noqa: E402 + AlarmRule, + GRID_COLUMNS, + NLQueryTemplate, + PreviewKind, + PreviewResult, + preview_from_store, + render_alarm_preview, + render_layout_preview, + render_nl_query_preview, +) + + +class LayoutPreviewTest(unittest.TestCase): + """布局预览。""" + + def test_basic_render(self): + widgets = [ + {"type": "process_view", "src": "x.svg", "x": 0, "y": 0, "w": 12, "h": 4, + "description": "工艺流程"}, + {"type": "trend", "bind": "R-801.TEMP", "x": 0, "y": 4, "w": 6, "h": 2}, + ] + r = render_layout_preview(widgets) + self.assertEqual(len(r.items), 2) + self.assertEqual(r.items[0]["type"], "process_view") + self.assertEqual(r.items[1]["bind"], "R-801.TEMP") + self.assertTrue(r.ok) # 无越界/重叠 + self.assertTrue(any("网格占用率" in n for n in r.notes)) + + def test_overflow_warning(self): + # x+w 超过 12 列 + widgets = [{"type": "trend", "bind": "p", "x": 8, "y": 0, "w": 6, "h": 2}] + r = render_layout_preview(widgets) + self.assertFalse(r.ok) + self.assertTrue(any("越界" in w for w in r.warnings)) + + def test_overlap_warning(self): + widgets = [ + {"type": "trend", "bind": "a", "x": 0, "y": 0, "w": 6, "h": 2}, + {"type": "kpi_card", "metric": "m", "label": "L", "x": 3, "y": 0, "w": 6, "h": 2}, + ] + r = render_layout_preview(widgets) + self.assertFalse(r.ok) + self.assertTrue(any("重叠" in w for w in r.warnings)) + + def test_bad_widget_type(self): + widgets = [{"type": "unknown", "x": 0, "y": 0, "w": 1, "h": 1}] + r = render_layout_preview(widgets) + self.assertFalse(r.ok) + self.assertTrue(any("非法类型" in w for w in r.warnings)) + + def test_kpi_card_carries_metric_and_label(self): + widgets = [{"type": "kpi_card", "metric": "yield", "label": "产率", + "x": 0, "y": 0, "w": 3, "h": 2}] + r = render_layout_preview(widgets) + self.assertEqual(r.items[0]["metric"], "yield") + self.assertEqual(r.items[0]["label"], "产率") + + def test_empty_widgets(self): + r = render_layout_preview([]) + self.assertEqual(r.items, []) + self.assertTrue(r.ok) + + +class AlarmPreviewTest(unittest.TestCase): + """告警预览。""" + + def test_render_rules(self): + rules = [AlarmRule("R-801.TEMP", "反应釜温度", ">", 120.0, "critical")] + r = render_alarm_preview(rules) + self.assertEqual(len(r.items), 1) + self.assertIn("critical", r.items[0]["text"]) + self.assertTrue(r.ok) + + def test_evaluate_triggered(self): + rules = [AlarmRule("R-801.TEMP", "温度", ">", 120.0, "critical")] + r = render_alarm_preview(rules, current_values={"R-801.TEMP": 130.0}) + self.assertTrue(r.items[0]["triggered"]) + self.assertEqual(r.items[0]["state"], "触发") + self.assertTrue(any("1/1" in n for n in r.notes)) + + def test_evaluate_not_triggered(self): + rules = [AlarmRule("P1", "温度", ">", 120.0, "warn")] + r = render_alarm_preview(rules, current_values={"P1": 100.0}) + self.assertFalse(r.items[0]["triggered"]) + self.assertEqual(r.items[0]["state"], "正常") + + def test_operators(self): + for op, val, thr in [(">=", 120, 120), ("<", 50, 100), ("<=", 100, 100), ("==", 5, 5)]: + rule = AlarmRule("P", "m", op, thr, "warn") + self.assertTrue(rule.evaluate(val), f"{op} {val} {thr} 应触发") + + def test_bad_level_and_operator(self): + rules = [AlarmRule("P", "m", "~", 1.0, level="boom")] + r = render_alarm_preview(rules) + self.assertFalse(r.ok) + self.assertTrue(any("非法级别" in w for w in r.warnings)) + self.assertTrue(any("非法运算符" in w for w in r.warnings)) + + +class NLQueryPreviewTest(unittest.TestCase): + """NL 查询预览。""" + + def test_render_with_examples(self): + t = NLQueryTemplate( + name="batch_query", + question_template="最近一批的{metric}是多少？", + slots={"metric": ["产率", "能耗"]}, + data_source="tdengine", + answer_hint="返回当批聚合值", + ) + r = render_nl_query_preview([t]) + self.assertEqual(len(r.items), 1) + self.assertEqual(len(r.items[0]["examples"]), 2) + self.assertIn("产率", r.items[0]["examples"][0]) + self.assertEqual(r.items[0]["data_source"], "tdengine") + + def test_empty_question_warns(self): + t = NLQueryTemplate(name="x", question_template="", slots={}) + r = render_nl_query_preview([t]) + self.assertFalse(r.ok) + + def test_no_slots_returns_template(self): + t = NLQueryTemplate(name="x", question_template="整体能耗？", slots={}) + self.assertEqual(t.render_examples(), ["整体能耗？"]) + + +class StorePreviewTest(unittest.TestCase): + """从 ConfigStore 一键预览。""" + + def setUp(self): + self._tmp = tempfile.mkdtemp() + self.store = ConfigStore(self._tmp) + + def tearDown(self): + import shutil + shutil.rmtree(self._tmp, ignore_errors=True) + + def test_layout_preview_from_store(self): + self.store.upsert(ConfigKind.LAYOUT, "dashboard", + [{"type": "trend", "bind": "p", "x": 0, "y": 0, "w": 6, "h": 2}]) + r = preview_from_store(self.store, PreviewKind.LAYOUT) + self.assertEqual(len(r.items), 1) + self.assertEqual(r.items[0]["bind"], "p") + + def test_empty_layout_is_graceful(self): + r = preview_from_store(self.store, PreviewKind.LAYOUT) + self.assertEqual(r.items, []) + self.assertTrue(any("未配置" in n for n in r.notes)) + + def test_alarm_preview_from_store(self): + self.store.upsert(ConfigKind.MODEL_PARAM, "alarm_temp", + {"point_id": "R-801.TEMP", "metric": "温度", + "operator": ">", "threshold": 120, "level": "critical"}) + r = preview_from_store(self.store, PreviewKind.ALARM, + current_values={"R-801.TEMP": 130}) + self.assertTrue(r.items[0]["triggered"]) + + def test_nl_query_preview_from_store(self): + self.store.upsert(ConfigKind.RAG_CONFIG, "nl_batch", + {"name": "批次查询", "question_template": "{m}多少？", + "slots": {"m": ["产率"]}, "data_source": "tdengine"}) + r = preview_from_store(self.store, PreviewKind.NL_QUERY) + self.assertEqual(len(r.items), 1) + self.assertEqual(r.items[0]["name"], "批次查询") + + +if __name__ == "__main__": + unittest.main()