feat: 完成 issue #20 [M1] 搭建缺省通用点位集（演示用，不依赖客户数据）

2026-08-04 22:44:20 +08:00
parent 91bb0c5912
commit 366311abdc
3 changed files with 135 additions and 0 deletions
@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+"""iAOP-Template-Ti 一期 · 缺省通用点位集 sanity 检查（离线基本验证）。
+
+检查项：
+1. 点位字典 CSV：表头与内核 `core/edge-gateway/config/point_dict.example.csv`
+   对齐（9 列），且非空、无空行、每行列数一致；
+2. protocol 列为合法采集协议（对齐内核 schema.VALID_PROTOCOLS）；
+3. sampleRate 为正整数。
+
+用法：python _sanity_check.py
+"""
+import csv
+import os
+import sys
+
+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（缺省点位集全部使用 simulator 演示驱动）
+VALID_PROTOCOLS = {
+    "opcua", "s7", "modbus", "weighing", "energy", "simulator",
+}
+
+
+def main() -> int:
+    failures: list[str] = []
+
+    # 1) 点位字典 CSV
+    csv_path = os.path.join(HERE, "point_dict.default.csv")
+    with open(csv_path, "r", encoding="utf-8") as fh:
+        rows = list(csv.reader(fh))
+    if not rows:
+        failures.append("point_dict.default.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 not row or all(not c for c in row):
+                failures.append(f"CSV 第 {i} 行为空行")
+            elif len(row) != len(EXPECTED_COLUMNS):
+                failures.append(f"CSV 第 {i} 行列数异常：{len(row)}")
+            else:
+                # 2) protocol 合法
+                proto = row[8].strip().lower()
+                if proto not in VALID_PROTOCOLS:
+                    failures.append(
+                        f"CSV 第 {i} 行非法协议：'{proto}'（合法值见内核 schema.VALID_PROTOCOLS）")
+                # 3) sampleRate 正整数
+                rate = row[5].strip()
+                if not rate.isdigit() or int(rate) <= 0:
+                    failures.append(f"CSV 第 {i} 行 sampleRate 必须为正整数：'{rate}'")
+
+    if failures:
+        print("FAIL")
+        for f in failures:
+            print(" -", f)
+        return 1
+    print(f"OK: 点位 {len(rows) - 1} 条，表头/协议/采样率校验通过")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())