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())
|