feat(#55): Ti 行业布局模板（四状态流程视图）+ layout_validator

纯标准库实现，对齐 PRD 5.5「⑤ 配置化驾驶舱」iAOP-cockpit-layout-v1： - cockpit.ti.yaml：海绵钛四状态工艺流程（氯化→精制→还原→蒸馏）， process_view 主视图声明 stages 覆盖，trend/kpi_card bind 对齐 point_dict.default.csv 的 point_id（CLF-01/RF-01/E-01/ST-01）。 - layout_validator.py：LayoutValidator 校验 widget 类型合法、12 列网格不越界、bind point_id 在点位字典内、四状态覆盖完整（order 单调/id 唯一）；零依赖 YAML 子集解析（复制 impurity-forecast _parse_yaml_subset）。 - tests：18 用例覆盖真实资产端到端、非法类型、网格越界/负坐标/零宽、 bind 漂移、缺 process_view、缺必需状态、order 非单调、stage 重复、 $schema 头、CSV 加载、空布局。 - _sanity_check.py：冒烟验证 9 widgets 全校验通过。
2026-08-05 05:32:23 +08:00
parent 2afea16005
commit 1290a8bc54
7 changed files with 1025 additions and 0 deletions
@@ -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