# -*- 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