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):

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)

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()