73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
# -*- 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())
|