From 1290a8bc540bc07f50aa1000375992e483bd9e80 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:28:20 +0800 Subject: [PATCH] =?UTF-8?q?feat(#55):=20Ti=20=E8=A1=8C=E4=B8=9A=E5=B8=83?= =?UTF-8?q?=E5=B1=80=E6=A8=A1=E6=9D=BF=EF=BC=88=E5=9B=9B=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E6=B5=81=E7=A8=8B=E8=A7=86=E5=9B=BE=EF=BC=89+=20layout=5Fvalid?= =?UTF-8?q?ator?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 纯标准库实现，对齐 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 全校验通过。 --- templates/ti-cl4/dashboard/README.md | 37 ++ templates/ti-cl4/dashboard/__init__.py | 27 + templates/ti-cl4/dashboard/_sanity_check.py | 33 ++ templates/ti-cl4/dashboard/cockpit.ti.yaml | 108 ++++ .../ti-cl4/dashboard/layout_validator.py | 500 ++++++++++++++++++ .../ti-cl4/dashboard/tests/_bootstrap.py | 13 + .../dashboard/tests/test_layout_validator.py | 307 +++++++++++ 7 files changed, 1025 insertions(+) create mode 100644 templates/ti-cl4/dashboard/README.md create mode 100644 templates/ti-cl4/dashboard/__init__.py create mode 100644 templates/ti-cl4/dashboard/_sanity_check.py create mode 100644 templates/ti-cl4/dashboard/cockpit.ti.yaml create mode 100644 templates/ti-cl4/dashboard/layout_validator.py create mode 100644 templates/ti-cl4/dashboard/tests/_bootstrap.py create mode 100644 templates/ti-cl4/dashboard/tests/test_layout_validator.py diff --git a/templates/ti-cl4/dashboard/README.md b/templates/ti-cl4/dashboard/README.md new file mode 100644 index 0000000..20e9be0 --- /dev/null +++ b/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 加载、空布局。 diff --git a/templates/ti-cl4/dashboard/__init__.py b/templates/ti-cl4/dashboard/__init__.py new file mode 100644 index 0000000..15ee51a --- /dev/null +++ b/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", +] diff --git a/templates/ti-cl4/dashboard/_sanity_check.py b/templates/ti-cl4/dashboard/_sanity_check.py new file mode 100644 index 0000000..a7bf846 --- /dev/null +++ b/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()) diff --git a/templates/ti-cl4/dashboard/cockpit.ti.yaml b/templates/ti-cl4/dashboard/cockpit.ti.yaml new file mode 100644 index 0000000..adcf963 --- /dev/null +++ b/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: 自然语言查询入口（工艺/质量/能耗问答） diff --git a/templates/ti-cl4/dashboard/layout_validator.py b/templates/ti-cl4/dashboard/layout_validator.py new file mode 100644 index 0000000..a48a86c --- /dev/null +++ b/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 diff --git a/templates/ti-cl4/dashboard/tests/_bootstrap.py b/templates/ti-cl4/dashboard/tests/_bootstrap.py new file mode 100644 index 0000000..8de360c --- /dev/null +++ b/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) diff --git a/templates/ti-cl4/dashboard/tests/test_layout_validator.py b/templates/ti-cl4/dashboard/tests/test_layout_validator.py new file mode 100644 index 0000000..920daca --- /dev/null +++ b/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()