iAOP/templates/resin/_sanity_check.py

# -*- coding: utf-8 -*-
"""iAOP-Template-Resin 模板资产 sanity 检查（无构建环境下的离线基本验证）。

检查项：
1. 点位字典 CSV：表头与内核 `core/edge-gateway/config/point_dict.example.csv`
   对齐（9 列），且非空、无空行；
2. 所有 YAML 资产（RAG 配置 / 驾驶舱布局 / 版本清单）可被 yaml 解析；
3. version.yaml 声明的 assets 清单文件全部存在；
4. 驾驶舱布局 widget 类型在 PRD 5.5 schema 允许集合内。

用法：python _sanity_check.py
"""
import csv
import os
import sys

import yaml  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))

# 对齐 core/edge-gateway/config/point_dict.example.csv 表头
EXPECTED_COLUMNS = [
    "device_id", "point_id", "name", "unit", "dataType",
    "sampleRate", "qualityCode", "opcNode", "protocol",
]

# 对齐内核 schema.VALID_PROTOCOLS / VALID_UNITS（含树脂模板扩展量纲 rpm / mmol/g）
VALID_PROTOCOLS = {"opcua", "s7", "modbus", "weighing", "energy", "simulator"}
VALID_UNITS = {
    "℃", "kPa", "MPa", "m³/h", "m3/h", "%", "kg", "t", "t/h", "m³", "m3",
    "A", "V", "Hz", "kW", "kWh", "Pa", "bar", "mm", "L", "L/min", "m/s",
    "rpm", "mmol/g",
}

# PRD 5.5 iAOP-cockpit-layout-v1 允许的 widget 类型
ALLOWED_WIDGETS = {
    "process_view", "trend", "kpi_card", "alarm_panel", "nl_query",
}


def main() -> int:
    failures: list[str] = []

    # 1) 点位字典 CSV
    csv_path = os.path.join(HERE, "point-dict", "point_dict.resin.csv")
    with open(csv_path, "r", encoding="utf-8") as fh:
        rows = list(csv.reader(fh))
    if not rows:
        failures.append("point_dict.resin.csv 为空")
    else:
        header = [c.strip() for c in rows[0]]
        if header != EXPECTED_COLUMNS:
            failures.append(
                f"CSV 表头与内核 schema 不一致：{header} != {EXPECTED_COLUMNS}")
        if len(rows) < 2:
            failures.append("CSV 缺少数据行")
        for i, row in enumerate(rows[1:], 2):
            if len(row) != len(EXPECTED_COLUMNS):
                failures.append(f"CSV 第 {i} 行列数异常：{len(row)}")
                continue
            unit, protocol = row[3].strip(), row[8].strip()
            if unit not in VALID_UNITS:
                failures.append(f"CSV 第 {i} 行非法量纲: '{unit}'（对齐内核 schema.VALID_UNITS）")
            if protocol and protocol not in VALID_PROTOCOLS:
                failures.append(f"CSV 第 {i} 行非法协议: '{protocol}'（对齐内核 schema.VALID_PROTOCOLS）")

    # 2) YAML 资产可解析
    for rel in ("rag-kb/kb.resin.template.yaml",
                "dashboard/cockpit.resin.yaml",
                "version.yaml"):
        path = os.path.join(HERE, rel)
        with open(path, "r", encoding="utf-8") as fh:
            yaml.safe_load(fh)

    # 3) version.yaml assets 清单存在
    with open(os.path.join(HERE, "version.yaml"), "r", encoding="utf-8") as fh:
        version = yaml.safe_load(fh)
    for rel in version.get("assets", []):
        if not os.path.isfile(os.path.join(HERE, rel)):
            failures.append(f"version.yaml 声明资产缺失：{rel}")

    # 4) 驾驶舱布局 widget 类型合法
    with open(os.path.join(HERE, "dashboard", "cockpit.resin.yaml"),
              "r", encoding="utf-8") as fh:
        cockpit = yaml.safe_load(fh)
    if cockpit.get("$schema") != "iAOP-cockpit-layout-v1":
        failures.append("驾驶舱布局缺少/错误 $schema")
    for w in cockpit.get("widgets", []):
        if w.get("type") not in ALLOWED_WIDGETS:
            failures.append(f"非法 widget 类型：{w.get('type')}")

    if failures:
        print("FAIL")
        for f in failures:
            print(" -", f)
        return 1
    print(f"OK: 点位 {len(rows) - 1} 条，YAML/清单/widget 校验通过")
    return 0


if __name__ == "__main__":
    sys.exit(main())