templates/ti-cl4/dashboard/README.md

@@ -0,0 +1,37 @@
+# 海绵钛驾驶舱布局资产 + 校验器（Issue #55 / PRD 5.5）
+
+> 父 Issue「⑤ Ti 行业布局模板（四状态流程视图）· 0.5d」
+
+把海绵钛车间驾驶舱布局落为**对齐 iAOP-cockpit-layout-v1 的资产 + 可校验的纯标准库
+校验器**（无 node/前端构建环境，零运行时依赖）。
+
+## 资产：`cockpit.ti.yaml`
+
+四状态工艺流程（PRD 4.2 海绵钛：**氯化 → 精制 → 还原 → 蒸馏**）：
+
+- `process_view` 主视图（首屏立即加载），`stages` 声明四状态覆盖；
+- `trend` 实时趋势（氯化炉温度 `CLF-01.TEMP` / 氯气流量 `CLF-01.CL2`）；
+- `kpi_card` KPI（TiCl₄纯度 `RF-01.PURITY` / 杂质 `RF-01.IMP` / 电耗 `E-01.KWH` / 蒸汽 `ST-01.STEAM`）；
+- `alarm_panel` 告警面板 + `nl_query` NL 查询入口。
+
+所有 `bind` 的 `point_id` 对齐 `templates/ti-cl4/point-dict/point_dict.default.csv`。
+
+## 校验器：`layout_validator.py`
+
+`LayoutValidator(layout_yaml, point_dict_csv).validate()` → `LayoutReport`，校验：
+
+1. **widget 类型合法**：在 `iAOP-cockpit-layout-v1` 允许集合内（process_view/trend/kpi_card/alarm_panel/nl_query）；
+2. **12 列网格不越界**：`0 ≤ x`、`x + w ≤ 12`、`y ≥ 0`、`w/h > 0`；
+3. **bind point_id 在点位字典内**：防模板漂移（trend/kpi_card 的 bind 必须命中点字典）；
+4. **四状态覆盖完整**：process_view 的 stages 必须覆盖氯化/精制/还原/蒸馏，order 单调递增、id 唯一。
+
+零依赖 YAML 子集解析（复制 impurity-forecast 的 `_parse_yaml_subset`，无 pyyaml）。
+
+## 测试
+
+```bash
+python -m unittest discover -s templates/ti-cl4/dashboard/tests -p "test_*.py" -v
+```
+
+覆盖正常 + 边界 + 错误（18 用例）：真实资产端到端通过、非法类型、网格越界/负坐标/零宽、
+bind 漂移、缺 process_view、缺必需状态、order 非单调、stage 重复、$schema 头、CSV 加载、空布局。

templates/ti-cl4/dashboard/__init__.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+"""海绵钛驾驶舱布局资产 + 校验器包（Issue #55）。
+
+对齐 PRD 5.5「⑤ 配置化驾驶舱」布局 JSON Schema（iAOP-cockpit-layout-v1）：
+四状态工艺流程（氯化 → 精制 → 还原 → 蒸馏）。
+"""
+from .layout_validator import (
+    LayoutError,
+    LayoutIssue,
+    LayoutReport,
+    LayoutValidator,
+    WidgetSpec,
+    ALLOWED_WIDGET_TYPES,
+    GRID_COLUMNS,
+    REQUIRED_STAGES,
+)
+
+__all__ = [
+    "LayoutError",
+    "LayoutIssue",
+    "LayoutReport",
+    "LayoutValidator",
+    "WidgetSpec",
+    "ALLOWED_WIDGET_TYPES",
+    "GRID_COLUMNS",
+    "REQUIRED_STAGES",
+]

templates/ti-cl4/dashboard/_sanity_check.py

@@ -0,0 +1,33 @@
+# -*- coding: utf-8 -*-
+"""海绵钛驾驶舱布局冒烟脚本（Issue #55）。
+
+直接运行 ``python _sanity_check.py`` 验证：cockpit.ti.yaml + point_dict.default.csv
+端到端校验通过（widget 类型/网格/bind/四状态全覆盖）。零第三方依赖。
+"""
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+
+from layout_validator import LayoutValidator  # noqa: E402
+
+LAYOUT_YAML = os.path.join(HERE, "cockpit.ti.yaml")
+POINT_DICT_CSV = os.path.join(
+    HERE, os.pardir, "point-dict", "point_dict.default.csv")
+
+
+def main() -> int:
+    report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
+    if not report.passed:
+        print("FAIL")
+        for issue in report.errors:
+            print(f" - [{issue.widget_id}] {issue.field}: {issue.reason}")
+        return 1
+    print(f"OK: {report.widget_count} widgets，"
+          f"类型/网格/bind/四状态校验通过")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

templates/ti-cl4/dashboard/cockpit.ti.yaml

@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+# 海绵钛（Ti）车间驾驶舱布局资产（iAOP-Template-Ti，EPIC #12）。
+#
+# 对齐 PRD 5.5「⑤ 配置化驾驶舱」布局 JSON Schema（iAOP-cockpit-layout-v1）：
+# 切换行业模板后，驾驶舱按本布局自动重排，无需改前端代码。
+# widget 类型：process_view（工艺流程视图）/ trend（实时趋势）/ kpi_card（KPI卡片）/
+#             alarm_panel（告警面板）/ nl_query（NL查询入口）。
+#
+# 四状态工艺流程（PRD 4.2 海绵钛：氯化 → 精制 → 还原 → 蒸馏）：
+# - 氯化  （CLF-01）：TiO₂ + Cl₂ + C → TiCl₄（沸腾氯化炉）
+# - 精制         ：粗 TiCl₄ → 精 TiCl₄（除钒/除硅，常压精馏）
+# - 还原  （RF-01）：TiCl₄ + Mg → 海绵钛（真空还原，Kroll 法）
+# - 蒸馏         ：海绵钛 + 残余 Mg/MgCl₂ 分离（真空蒸馏）
+# process_view 的 stages 字段声明四状态覆盖，layout_validator 校验完整性。
+#
+# bind 的 point_id 对齐 templates/ti-cl4/point-dict/point_dict.default.csv。
+$schema: iAOP-cockpit-layout-v1
+title: 海绵钛车间驾驶舱
+theme: dark
+widgets:
+  # ---- 四状态工艺流程主视图（首屏立即加载） ---------------------------
+  - type: process_view
+    src: ti_four_state.svg
+    x: 0
+    y: 0
+    w: 12
+    h: 4
+    description: 四状态工艺流程（氯化 → 精制 → 还原 → 蒸馏）
+    stages:
+      - id: chlorination
+        name: 氯化
+        device: CLF-01
+        order: 1
+      - id: purification
+        name: 精制
+        order: 2
+      - id: reduction
+        name: 还原
+        device: RF-01
+        order: 3
+      - id: distillation
+        name: 蒸馏
+        order: 4
+  # ---- 实时趋势：氯化炉温度/氯气流量（工艺核心监控） -------------------
+  - type: trend
+    bind: CLF-01.TEMP
+    x: 0
+    y: 4
+    w: 6
+    h: 2
+    description: 氯化炉温度实时趋势（沸腾氯化炉温 850±50℃）
+  - type: trend
+    bind: CLF-01.CL2
+    x: 6
+    y: 4
+    w: 6
+    h: 2
+    description: 氯气流量实时趋势（流态化监控）
+  # ---- KPI 卡片：还原质量/能耗（海绵钛核心指标） -----------------------
+  - type: kpi_card
+    metric: ticl4_purity
+    bind: RF-01.PURITY
+    label: TiCl₄纯度
+    x: 0
+    y: 6
+    w: 3
+    h: 2
+    description: 还原 TiCl₄ 纯度（%，工艺 ≥ 99.9%）
+  - type: kpi_card
+    metric: ticl4_impurity
+    bind: RF-01.IMP
+    label: 杂质含量
+    x: 3
+    y: 6
+    w: 3
+    h: 2
+    description: 还原杂质含量（%，越低越好）
+  - type: kpi_card
+    metric: energy_per_ton
+    bind: E-01.KWH
+    label: 累计电耗
+    x: 6
+    y: 6
+    w: 3
+    h: 2
+    description: 车间累计电耗（kWh，单吨海绵钛综合能耗输入）
+  - type: kpi_card
+    metric: steam_flow
+    bind: ST-01.STEAM
+    label: 蒸汽流量
+    x: 9
+    y: 6
+    w: 3
+    h: 2
+    description: 蒸汽流量（t/h，公用工程监控）
+  # ---- 告警面板 + NL 查询入口 ------------------------------------------
+  - type: alarm_panel
+    x: 0
+    y: 8
+    w: 9
+    h: 3
+    description: 告警面板（氯化炉温/还原真空度/纯度/杂质异常）
+  - type: nl_query
+    x: 9
+    y: 8
+    w: 3
+    h: 3
+    description: 自然语言查询入口（工艺/质量/能耗问答）

templates/ti-cl4/dashboard/layout_validator.py

@@ -0,0 +1,500 @@
+# -*- coding: utf-8 -*-
+"""海绵钛驾驶舱布局校验器（Issue #55 / PRD 5.5「⑤ 配置化驾驶舱」）。
+
+PRD 5.5：切换行业模板后，驾驶舱按布局资产自动重排，无需改前端代码。
+本模块把布局资产（``cockpit.ti.yaml``）落为**可校验的纯标准库资产 + 校验器**——
+给定布局 YAML + 点位字典 CSV，校验：
+
+1. **widget 类型合法**：在 PRD 5.5 ``iAOP-cockpit-layout-v1`` 允许集合内
+   （process_view/trend/kpi_card/alarm_panel/nl_query）。
+2. **12 列网格不越界**：每个 widget ``0 ≤ x`` 且 ``x + w ≤ 12``，``y ≥ 0``、
+   ``h > 0``；坐标为非负整数，w/h 正整数（网格对齐）。
+3. **bind 的 point_id 在点位字典内**：trend/kpi_card 的 ``bind`` 必须命中
+   ``point_dict.default.csv`` 的 ``point_id`` 列（防模板漂移）。
+4. **四状态视图覆盖完整**：process_view 的 ``stages`` 必须覆盖工艺全流程
+   （氯化/精制/还原/蒸馏），且 order 单调递增、id 唯一。
+
+校验产出 :class:`LayoutReport`（PASS/FAIL + 逐条 :class:`LayoutIssue`，
+每条 issue 带 ``reason`` 可解释）。
+
+设计要点
+--------
+- **零依赖 YAML 子集解析**：复制 impurity-forecast features.py 的
+  ``_parse_yaml_subset``（无 pyyaml），支持 map/list/标量/行内 flow map。
+- **纯标准库**：CSV 用标准库 csv，无 numpy/pyyaml 依赖。
+- **换行业只改资产**：校验器对任何对齐 ``iAOP-cockpit-layout-v1`` 的布局都适用。
+
+用法::
+
+    report = LayoutValidator(layout_yaml, point_dict_csv).validate()
+    if not report.passed:
+        for issue in report.issues:
+            print(issue.severity, issue.widget_id, issue.reason)
+"""
+from __future__ import annotations
+
+import csv
+import os
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Dict, List, Optional, Tuple
+
+#: iAOP-cockpit-layout-v1 允许的 widget 类型集合（对齐 resin _sanity_check）。
+ALLOWED_WIDGET_TYPES = frozenset({
+    "process_view", "trend", "kpi_card", "alarm_panel", "nl_query",
+})
+
+#: 12 列网格（主流前端栅格标准，对齐 cockpit layout v1）。
+GRID_COLUMNS = 12
+
+#: 海绵钛四状态工艺流程（PRD 4.2：氯化 → 精制 → 还原 → 蒸馏）。
+#: process_view 的 stages 必须覆盖这四个 id。
+REQUIRED_STAGES = ("chlorination", "purification", "reduction", "distillation")
+
+
+class LayoutError(ValueError):
+    """布局资产解析/声明错误（YAML 格式错、表头缺字段等）。"""
+
+
+class Severity(str, Enum):
+    """问题严重度。"""
+
+    ERROR = "error"    # 阻断：布局不可用（类型非法/越界/bind 缺失/状态缺失）
+    WARN = "warn"      # 告警：可运行但不规范（重复/顺序乱）
+
+
+@dataclass
+class LayoutIssue:
+    """单条布局校验问题（含 reason 可解释）。"""
+
+    severity: Severity
+    reason: str
+    widget_id: str = ""        # 关联 widget（index 或 src/metric）
+    field: str = ""            # 关联字段（type/x/bind/stages ...）
+
+    @property
+    def is_error(self) -> bool:
+        return self.severity is Severity.ERROR
+
+
+@dataclass
+class LayoutReport:
+    """布局校验报告。"""
+
+    issues: List[LayoutIssue] = field(default_factory=list)
+    widget_count: int = 0
+
+    @property
+    def errors(self) -> List[LayoutIssue]:
+        return [i for i in self.issues if i.is_error]
+
+    @property
+    def passed(self) -> bool:
+        """通过 = 无 ERROR（WARN 不阻断）。"""
+        return not any(i.is_error for i in self.issues)
+
+    def to_dict(self) -> dict:
+        return {
+            "passed": self.passed,
+            "widget_count": self.widget_count,
+            "error_count": len(self.errors),
+            "warn_count": len(self.issues) - len(self.errors),
+            "issues": [
+                {"severity": i.severity.value, "widget_id": i.widget_id,
+                 "field": i.field, "reason": i.reason}
+                for i in self.issues
+            ],
+        }
+
+
+@dataclass
+class WidgetSpec:
+    """单个 widget 的内存模型（从 YAML 解析）。"""
+
+    index: int                       # 在 widgets 列表中的位置（0 起）
+    type: str
+    x: int = 0
+    y: int = 0
+    w: int = 1
+    h: int = 1
+    bind: str = ""                   # trend/kpi_card 绑定的 point_id
+    src: str = ""                    # process_view 的 SVG
+    metric: str = ""                 # kpi_card 的 metric
+    label: str = ""
+    description: str = ""
+    stages: List[Dict[str, object]] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# 零依赖 YAML 子集解析（复制自 impurity-forecast features.py，对齐 data-bus）
+# ---------------------------------------------------------------------------
+
+def _parse_scalar(text: str) -> str:
+    """去掉标量两侧引号与行内注释。"""
+    t = text.split(" #", 1)[0].strip()
+    if len(t) >= 2 and t[0] == t[-1] and t[0] in ("'", '"'):
+        return t[1:-1]
+    return t
+
+
+def _parse_flow_value(text: str):
+    """解析 ``key: value`` 右侧值，支持行内 flow map ``{k: v, k: v}``。"""
+    t = text.split(" #", 1)[0].strip()
+    if t.startswith("{") and t.endswith("}"):
+        inner = t[1:-1].strip()
+        out: Dict[str, object] = {}
+        if not inner:
+            return out
+        for part in inner.split(","):
+            if ":" not in part:
+                raise LayoutError(f"flow map 项不是键值对：{part!r}")
+            k, _, v = part.partition(":")
+            out[k.strip()] = _parse_scalar(v)
+        return out
+    return _parse_scalar(text)
+
+
+def _strip_comments(lines: List[str]) -> List[Tuple[str, int]]:
+    out: List[Tuple[str, int]] = []
+    for i, ln in enumerate(lines):
+        s = ln.strip()
+        if not s or s.startswith("#"):
+            continue
+        out.append((ln, i + 1))
+    return out
+
+
+def _parse_node(lines: List[Tuple[str, int]], i: int, indent: int):
+    """递归解析 YAML 节点（map / list / scalar）。返回 (value, next_i)。"""
+    text, _ = lines[i]
+    # ---- list 节点 ----
+    if text.lstrip(" ").startswith("- "):
+        items: List[object] = []
+        while i < len(lines):
+            t, no = lines[i]
+            stripped = t.lstrip(" ")
+            if not stripped.startswith("- "):
+                break
+            lead_j = len(t) - len(t.lstrip(" "))
+            if lead_j != indent:
+                break
+            item_text = stripped[2:].strip()
+            if not item_text:
+                raise LayoutError(f"cockpit.yaml 第 {no} 行：list 项为空")
+            if ":" in item_text:
+                map_indent = len(t) - len(t.lstrip(" ")) + 2
+                lines[i] = (" " * map_indent + item_text, no)
+                v, i = _parse_node(lines, i, map_indent)
+                items.append(v)
+            else:
+                items.append(_parse_flow_value(item_text))
+                i += 1
+        return items, i
+    # ---- map 节点 ----
+    result: Dict[str, object] = {}
+    while i < len(lines):
+        t, no = lines[i]
+        lead_j = len(t) - len(t.lstrip(" "))
+        if lead_j < indent or t.lstrip(" ").startswith("- "):
+            break
+        if lead_j > indent:
+            raise LayoutError(
+                f"cockpit.yaml 第 {no} 行缩进异常（期望 {indent}，实际 {lead_j}）")
+        if ":" not in t:
+            raise LayoutError(f"cockpit.yaml 第 {no} 行不是合法键值对：{t!r}")
+        key, _, rest = t.partition(":")
+        key = key.strip()
+        rest = rest.strip()
+        if rest:
+            result[key] = _parse_flow_value(rest)
+            i += 1
+            continue
+        if i + 1 >= len(lines):
+            raise LayoutError(f"cockpit.yaml 第 {no} 行 {key!r} 缺少值")
+        sub_indent = len(lines[i + 1][0]) - len(lines[i + 1][0].lstrip(" "))
+        if sub_indent <= indent:
+            raise LayoutError(f"cockpit.yaml 第 {no} 行 {key!r} 缺少值（无嵌套）")
+        v, i = _parse_node(lines, i + 1, sub_indent)
+        result[key] = v
+    return result, i
+
+
+def _load_yaml_text(text: str) -> Dict[str, object]:
+    """解析 YAML 文本为 dict（顶层必须是 map）。"""
+    lines = _strip_comments(text.splitlines())
+    if not lines:
+        return {}
+    top_indent = len(lines[0][0]) - len(lines[0][0].lstrip(" "))
+    value, next_i = _parse_node(lines, 0, top_indent)
+    if not isinstance(value, dict):
+        raise LayoutError("cockpit.yaml 顶层必须是 map")
+    if next_i < len(lines):
+        raise LayoutError(
+            f"cockpit.yaml 第 {lines[next_i][1]} 行：顶层存在多个节点")
+    return value
+
+
+# ---------------------------------------------------------------------------
+# 点位字典加载（CSV → point_id 集合）
+# ---------------------------------------------------------------------------
+
+def load_point_ids(csv_path: str) -> List[str]:
+    """从点位字典 CSV 加载全部 point_id（保序，对齐 CSV point_id 列）。
+
+    CSV 表头对齐 core/edge-gateway point_dict schema（第二列 point_id）。
+    """
+    if not os.path.isfile(csv_path):
+        raise LayoutError(f"点位字典 CSV 不存在：{csv_path}")
+    with open(csv_path, "r", encoding="utf-8") as fh:
+        rows = list(csv.reader(fh))
+    if not rows:
+        raise LayoutError(f"点位字典 CSV 为空：{csv_path}")
+    header = [c.strip() for c in rows[0]]
+    if "point_id" not in header:
+        raise LayoutError(
+            f"点位字典 CSV 表头缺 point_id 列：{header}")
+    col = header.index("point_id")
+    ids: List[str] = []
+    for i, row in enumerate(rows[1:], 2):
+        if len(row) <= col:
+            continue
+        pid = row[col].strip()
+        if pid:
+            ids.append(pid)
+    if not ids:
+        raise LayoutError(f"点位字典 CSV 无 point_id 数据行：{csv_path}")
+    return ids
+
+
+# ---------------------------------------------------------------------------
+# 校验器
+# ---------------------------------------------------------------------------
+
+class LayoutValidator:
+    """海绵钛驾驶舱布局校验器。
+
+    Args:
+        layout_yaml_path:  布局资产路径（cockpit.ti.yaml）。
+        point_dict_csv_path: 点位字典 CSV 路径（point_dict.default.csv）。
+        grid_columns:      网格列数（默认 12，对齐 cockpit layout v1）。
+        required_stages:   process_view 必须覆盖的 stage id（默认海绵钛四状态）。
+    """
+
+    def __init__(
+        self,
+        layout_yaml_path: str,
+        point_dict_csv_path: Optional[str] = None,
+        grid_columns: int = GRID_COLUMNS,
+        required_stages: Tuple[str, ...] = REQUIRED_STAGES,
+    ) -> None:
+        if grid_columns <= 0:
+            raise LayoutError(f"grid_columns 必须 > 0，实际 {grid_columns}")
+        self.layout_path = layout_yaml_path
+        self.point_dict_path = point_dict_csv_path
+        self.grid_columns = int(grid_columns)
+        self.required_stages = tuple(required_stages)
+
+    # ------------------------------------------------------------------
+    def validate(self) -> LayoutReport:
+        """执行全部校验，返回报告。"""
+        report = LayoutReport()
+        # 1) 解析布局 YAML
+        try:
+            with open(self.layout_path, "r", encoding="utf-8") as fh:
+                data = _load_yaml_text(fh.read())
+        except LayoutError:
+            raise
+        except OSError as exc:
+            raise LayoutError(f"布局 YAML 读取失败：{self.layout_path} ({exc})") from exc
+
+        # schema 头校验
+        schema = str(data.get("$schema", "")).strip()
+        if schema != "iAOP-cockpit-layout-v1":
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR,
+                field="$schema",
+                reason=f"$schema 应为 'iAOP-cockpit-layout-v1'，实际 {schema!r}",
+            ))
+
+        # 2) 解析 widgets
+        raw_widgets = data.get("widgets") or []
+        if not isinstance(raw_widgets, list):
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, field="widgets",
+                reason=f"widgets 必须是 list，实际 {type(raw_widgets).__name__}"))
+            return report
+        widgets = self._parse_widgets(raw_widgets, report)
+        report.widget_count = len(widgets)
+
+        if not widgets:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, field="widgets",
+                reason="布局无任何 widget"))
+            return report
+
+        # 3) 加载点位字典（bind 校验需要）
+        point_ids: Optional[set] = None
+        if self.point_dict_path:
+            try:
+                point_ids = set(load_point_ids(self.point_dict_path))
+            except LayoutError as exc:
+                report.issues.append(LayoutIssue(
+                    severity=Severity.ERROR, field="point_dict",
+                    reason=str(exc)))
+
+        # 4) 逐 widget 校验
+        for w in widgets:
+            self._check_widget(w, point_ids, report)
+
+        # 5) process_view 四状态覆盖
+        self._check_process_views(widgets, report)
+
+        return report
+
+    # ------------------------------------------------------------------
+    def _parse_widgets(self, raw_widgets: List[object],
+                       report: LayoutReport) -> List[WidgetSpec]:
+        widgets: List[WidgetSpec] = []
+        for idx, item in enumerate(raw_widgets):
+            if not isinstance(item, dict):
+                report.issues.append(LayoutIssue(
+                    severity=Severity.ERROR, widget_id=f"[{idx}]",
+                    field="widgets",
+                    reason=f"widgets[{idx}] 必须是 map，实际 {type(item).__name__}"))
+                continue
+            wtype = str(item.get("type", "")).strip()
+            widgets.append(WidgetSpec(
+                index=idx,
+                type=wtype,
+                x=_to_int(item.get("x"), 0),
+                y=_to_int(item.get("y"), 0),
+                w=_to_int(item.get("w"), 1),
+                h=_to_int(item.get("h"), 1),
+                bind=str(item.get("bind", "")).strip(),
+                src=str(item.get("src", "")).strip(),
+                metric=str(item.get("metric", "")).strip(),
+                label=str(item.get("label", "")).strip(),
+                description=str(item.get("description", "")).strip(),
+                stages=_as_list_of_dict(item.get("stages")),
+            ))
+        return widgets
+
+    # ------------------------------------------------------------------
+    def _check_widget(self, w: WidgetSpec, point_ids: Optional[set],
+                      report: LayoutReport) -> None:
+        wid = f"[{w.index}]({w.type})"
+        # 4a) widget 类型合法
+        if w.type not in ALLOWED_WIDGET_TYPES:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, widget_id=wid, field="type",
+                reason=f"非法 widget 类型 {w.type!r}（允许 {sorted(ALLOWED_WIDGET_TYPES)})"))
+
+        # 4b) 12 列网格不越界（坐标非负整数、x+w ≤ columns、h>0）
+        if w.x < 0 or w.y < 0:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, widget_id=wid, field="grid",
+                reason=f"坐标不能为负：x={w.x} y={w.y}"))
+        if w.w <= 0 or w.h <= 0:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, widget_id=wid, field="grid",
+                reason=f"w/h 必须为正整数：w={w.w} h={w.h}"))
+        if w.x + w.w > self.grid_columns:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, widget_id=wid, field="grid",
+                reason=f"越出 {self.grid_columns} 列网格：x={w.x}+w={w.w}"
+                       f"={w.x + w.w} > {self.grid_columns}"))
+
+        # 4c) bind 的 point_id 必须在点位字典内（trend/kpi_card）
+        if w.bind:
+            if point_ids is not None and w.bind not in point_ids:
+                report.issues.append(LayoutIssue(
+                    severity=Severity.ERROR, widget_id=wid, field="bind",
+                    reason=f"bind point_id {w.bind!r} 不在点位字典内"
+                           f"（防模板漂移，对齐 point_dict.default.csv）"))
+
+    # ------------------------------------------------------------------
+    def _check_process_views(self, widgets: List[WidgetSpec],
+                             report: LayoutReport) -> None:
+        """校验 process_view 的四状态覆盖完整。"""
+        pv = [w for w in widgets if w.type == "process_view"]
+        if not pv:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, field="process_view",
+                reason="布局缺少 process_view（四状态工艺流程主视图必需）"))
+            return
+
+        covered: Dict[str, WidgetSpec] = {}  # stage_id → widget
+        for w in pv:
+            wid = f"[{w.index}](process_view)"
+            if not w.stages:
+                report.issues.append(LayoutIssue(
+                    severity=Severity.ERROR, widget_id=wid, field="stages",
+                    reason="process_view 缺少 stages 声明（四状态覆盖必需）"))
+                continue
+
+            stage_ids: List[str] = []
+            orders: List[int] = []
+            seen: set = set()
+            for st in w.stages:
+                sid = str(st.get("id", "")).strip()
+                sname = str(st.get("name", "")).strip()
+                if not sid:
+                    report.issues.append(LayoutIssue(
+                        severity=Severity.ERROR, widget_id=wid, field="stages",
+                        reason=f"stage 缺少 id（name={sname!r}）"))
+                    continue
+                if sid in seen:
+                    report.issues.append(LayoutIssue(
+                        severity=Severity.WARN, widget_id=wid, field="stages",
+                        reason=f"stage id 重复：{sid!r}"))
+                    continue
+                seen.add(sid)
+                stage_ids.append(sid)
+                covered.setdefault(sid, w)
+                order = st.get("order")
+                if order is not None:
+                    try:
+                        orders.append(int(order))
+                    except (TypeError, ValueError):
+                        report.issues.append(LayoutIssue(
+                            severity=Severity.ERROR, widget_id=wid, field="stages",
+                            reason=f"stage {sid!r} order 非整数：{order!r}"))
+
+            # order 单调递增校验
+            if orders and len(orders) == len(stage_ids):
+                if orders != sorted(orders):
+                    report.issues.append(LayoutIssue(
+                        severity=Severity.ERROR, widget_id=wid, field="stages",
+                        reason=f"stage order 非单调递增：{orders}"))
+
+        # 必需四状态全覆盖
+        missing = [s for s in self.required_stages if s not in covered]
+        if missing:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, field="stages",
+                reason=f"process_view stages 未覆盖必需四状态：{missing}"
+                       f"（氯化/精制/还原/蒸馏）"))
+
+
+# ---------------------------------------------------------------------------
+# 辅助
+# ---------------------------------------------------------------------------
+
+def _to_int(value: object, default: int) -> int:
+    """把 YAML 解析出的值（可能是 str/int）转为 int；失败返回 default。"""
+    if value is None or value == "":
+        return default
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        raise LayoutError(f"坐标值不是整数：{value!r}")
+
+
+def _as_list_of_dict(value: object) -> List[Dict[str, object]]:
+    if not isinstance(value, list):
+        return []
+    out: List[Dict[str, object]] = []
+    for item in value:
+        if isinstance(item, dict):
+            out.append(item)
+    return out

templates/ti-cl4/dashboard/tests/_bootstrap.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+"""测试引导：把 dashboard 测试根目录加入 sys.path，使 layout_validator 可导入。
+
+dashboard 目录名是合法 Python 标识符，直接作为包导入；本引导把父目录
+（templates/ti-cl4/dashboard）挂到 sys.path，使 ``from layout_validator import ...``
+在 unittest 发现机制下可解析（与 core 模块测试引导同款）。
+"""
+import os
+import sys
+
+PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if PKG_DIR not in sys.path:
+    sys.path.insert(0, PKG_DIR)

templates/ti-cl4/dashboard/tests/test_layout_validator.py

@@ -0,0 +1,307 @@
+# -*- coding: utf-8 -*-
+"""海绵钛驾驶舱布局校验器测试（Issue #55）。
+
+覆盖：
+1. 真实 cockpit.ti.yaml + point_dict.default.csv 全部通过（端到端）；
+2. widget 类型合法集合（非法类型 → ERROR）；
+3. 12 列网格校验（越界/负坐标/非正 w/h）；
+4. bind point_id 在点位字典内（漂移 → ERROR）；
+5. process_view 四状态覆盖（缺 stage id / order 非单调 / 缺必需状态）；
+6. $schema 头校验；
+7. YAML 解析（flow map / 嵌套）；
+8. 点位字典 CSV 加载（缺表头/空文件）；
+9. 空布局 / 边界。
+"""
+import os
+import unittest
+
+import _bootstrap  # noqa: F401  (sys.path 挂载)
+
+from layout_validator import (
+    GRID_COLUMNS,
+    LayoutError,
+    LayoutValidator,
+    REQUIRED_STAGES,
+    Severity,
+    load_point_ids,
+)
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+DASHBOARD_DIR = os.path.dirname(HERE)
+LAYOUT_YAML = os.path.join(DASHBOARD_DIR, "cockpit.ti.yaml")
+POINT_DICT_CSV = os.path.join(
+    DASHBOARD_DIR, os.pardir, "point-dict", "point_dict.default.csv")
+
+
+def _write_layout(tmp_path: str, content: str) -> str:
+    """把布局内容写到临时文件，返回路径。"""
+    path = os.path.join(tmp_path, "cockpit.test.yaml")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(content)
+    return path
+
+
+def _write_point_dict(tmp_path: str, ids: list) -> str:
+    """写一个最小点位字典 CSV（仅 point_id 列）。"""
+    path = os.path.join(tmp_path, "points.csv")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\n")
+        for i, pid in enumerate(ids):
+            fh.write(f"D{i},{pid},n,u,float,1000,true,n,simulator\n")
+    return path
+
+
+# 最小合法布局模板（便于构造各类变形）
+_VALID_LAYOUT = """\
+$schema: iAOP-cockpit-layout-v1
+title: 测试驾驶舱
+theme: dark
+widgets:
+  - type: process_view
+    src: ti_four_state.svg
+    x: 0
+    y: 0
+    w: 12
+    h: 4
+    description: 四状态工艺流程
+    stages:
+      - id: chlorination
+        name: 氯化
+        order: 1
+      - id: purification
+        name: 精制
+        order: 2
+      - id: reduction
+        name: 还原
+        order: 3
+      - id: distillation
+        name: 蒸馏
+        order: 4
+  - type: trend
+    bind: CLF-01.TEMP
+    x: 0
+    y: 4
+    w: 6
+    h: 2
+    description: 氯化炉温度
+"""
+
+
+class TestEndToEndRealAssets(unittest.TestCase):
+    """真实 cockpit.ti.yaml + point_dict.default.csv 端到端校验。"""
+
+    def test_real_layout_passes(self):
+        report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
+        if not report.passed:
+            for issue in report.errors:
+                print("ERROR:", issue.widget_id, issue.field, issue.reason)
+        self.assertTrue(report.passed, "真实布局应通过全部校验")
+        self.assertGreater(report.widget_count, 0)
+
+    def test_real_layout_has_process_view_with_four_stages(self):
+        report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
+        # 无 stages 相关 ERROR
+        stage_errors = [i for i in report.errors if i.field == "stages"]
+        self.assertEqual(stage_errors, [])
+
+
+class TestWidgetType(unittest.TestCase):
+    """widget 类型合法性。"""
+
+    def test_invalid_widget_type_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = _VALID_LAYOUT.replace("type: trend", "type: radar_chart")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            errors = [i for i in report.errors if i.field == "type"]
+            self.assertEqual(len(errors), 1)
+            self.assertIn("非法 widget 类型", errors[0].reason)
+
+
+class TestGridBounds(unittest.TestCase):
+    """12 列网格校验。"""
+
+    def test_x_plus_w_exceeds_columns(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # trend x=10 w=6 → 16 > 12
+            layout = _VALID_LAYOUT.replace(
+                "    bind: CLF-01.TEMP\n    x: 0\n    y: 4\n    w: 6\n    h: 2",
+                "    bind: CLF-01.TEMP\n    x: 10\n    y: 4\n    w: 6\n    h: 2")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            grid_errors = [i for i in report.errors if i.field == "grid"
+                           and "越出" in i.reason]
+            self.assertEqual(len(grid_errors), 1)
+
+    def test_negative_x_rejected(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = _VALID_LAYOUT.replace("    x: 0\n    y: 4\n    w: 6\n    h: 2",
+                                           "    x: -1\n    y: 4\n    w: 6\n    h: 2")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            neg = [i for i in report.errors if "坐标不能为负" in i.reason]
+            self.assertEqual(len(neg), 1)
+
+    def test_zero_width_rejected(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = _VALID_LAYOUT.replace("    x: 0\n    y: 4\n    w: 6\n    h: 2",
+                                           "    x: 0\n    y: 4\n    w: 0\n    h: 2")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            wh = [i for i in report.errors if "w/h 必须为正整数" in i.reason]
+            self.assertEqual(len(wh), 1)
+
+
+class TestBindPointId(unittest.TestCase):
+    """bind point_id 在点位字典内。"""
+
+    def test_bind_not_in_dict_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            path = _write_layout(td, _VALID_LAYOUT)
+            # 点位字典不含 CLF-01.TEMP
+            csv_path = _write_point_dict(td, ["OTHER-01.X"])
+            report = LayoutValidator(path, csv_path).validate()
+            bind_err = [i for i in report.errors if i.field == "bind"]
+            self.assertEqual(len(bind_err), 1)
+            self.assertIn("不在点位字典内", bind_err[0].reason)
+
+    def test_bind_in_dict_passes(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            path = _write_layout(td, _VALID_LAYOUT)
+            csv_path = _write_point_dict(td, ["CLF-01.TEMP"])
+            report = LayoutValidator(path, csv_path).validate()
+            bind_err = [i for i in report.errors if i.field == "bind"]
+            self.assertEqual(bind_err, [])
+
+
+class TestProcessViewStages(unittest.TestCase):
+    """process_view 四状态覆盖。"""
+
+    def test_missing_process_view_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # 删除 process_view 块（保留 trend）
+            layout = """\
+$schema: iAOP-cockpit-layout-v1
+title: t
+widgets:
+  - type: trend
+    bind: CLF-01.TEMP
+    x: 0
+    y: 0
+    w: 6
+    h: 2
+"""
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            pv_err = [i for i in report.errors if i.field == "process_view"]
+            self.assertEqual(len(pv_err), 1)
+
+    def test_missing_required_stage_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # 删除 distillation stage
+            layout = _VALID_LAYOUT.replace(
+                "      - id: distillation\n        name: 蒸馏\n        order: 4\n", "")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            missing = [i for i in report.errors if "未覆盖必需四状态" in i.reason]
+            self.assertEqual(len(missing), 1)
+            self.assertIn("distillation", missing[0].reason)
+
+    def test_non_monotonic_order_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # 把 reduction order 改为 5（> distillation 的 4）→ 非单调
+            layout = _VALID_LAYOUT.replace("      - id: reduction\n        name: 还原\n        order: 3",
+                                           "      - id: reduction\n        name: 还原\n        order: 5")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            order_err = [i for i in report.errors if "order 非单调递增" in i.reason]
+            self.assertEqual(len(order_err), 1)
+
+    def test_duplicate_stage_id_warn(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # 重复 chlorination（覆盖必需状态校验仍过，但 WARN 重复）
+            layout = _VALID_LAYOUT + """\
+"""
+            # 构造一个有重复 stage 的 process_view（替换 stages 块）
+            dup_layout = _VALID_LAYOUT.replace(
+                "      - id: distillation\n        name: 蒸馏\n        order: 4",
+                "      - id: distillation\n        name: 蒸馏\n        order: 4\n"
+                "      - id: chlorination\n        name: 氯化2\n        order: 5")
+            path = _write_layout(td, dup_layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            dup = [i for i in report.issues if "stage id 重复" in i.reason]
+            self.assertEqual(len(dup), 1)
+
+
+class TestSchemaHeader(unittest.TestCase):
+    """$schema 头校验。"""
+
+    def test_wrong_schema_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = _VALID_LAYOUT.replace("iAOP-cockpit-layout-v1", "some-other-schema")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            schema_err = [i for i in report.errors if i.field == "$schema"]
+            self.assertEqual(len(schema_err), 1)
+
+
+class TestPointDictLoader(unittest.TestCase):
+    """点位字典 CSV 加载。"""
+
+    def test_load_point_ids(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            csv_path = _write_point_dict(td, ["A.X", "B.Y"])
+            ids = load_point_ids(csv_path)
+            self.assertEqual(ids, ["A.X", "B.Y"])
+
+    def test_missing_csv_raises(self):
+        with self.assertRaises(LayoutError):
+            load_point_ids("/nonexistent/points.csv")
+
+    def test_csv_missing_point_id_column_raises(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            path = os.path.join(td, "bad.csv")
+            with open(path, "w", encoding="utf-8") as fh:
+                fh.write("device_id,name\nD1,n\n")
+            with self.assertRaises(LayoutError):
+                load_point_ids(path)
+
+
+class TestReportExport(unittest.TestCase):
+    """报告序列化 + 边界。"""
+
+    def test_empty_layout_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = """\
+$schema: iAOP-cockpit-layout-v1
+title: t
+widgets: []
+"""
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            self.assertFalse(report.passed)
+
+    def test_report_to_dict(self):
+        report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
+        d = report.to_dict()
+        self.assertEqual(d["passed"], True)
+        self.assertIn("widget_count", d)
+        self.assertEqual(d["error_count"], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()