mirror of
https://gitee.com/js-yhsec/energy_storage.git
synced 2026-05-28 03:09:24 +08:00
修改HTTP接口测试问题
This commit is contained in:
@@ -1,77 +1,88 @@
|
||||
import re
|
||||
import json
|
||||
from openpyxl import load_workbook
|
||||
|
||||
def parse_datatype(text):
|
||||
datatypes = ["int16", "uint16", "int32", "uint32", "int64", "uint64"]
|
||||
for index, datatype in enumerate(datatypes):
|
||||
left, separator, right = text.partition(datatype)
|
||||
if len(separator) != 0:
|
||||
return datatype, left + right
|
||||
return "", text
|
||||
def read_cell(sheet, row, col):
|
||||
val = str(sheet.cell(row, col).value)
|
||||
if val == "None":
|
||||
val = ""
|
||||
return val.strip()
|
||||
|
||||
def read_sheet(wb, topic, sht_name):
|
||||
sheet = wb[sht_name]
|
||||
|
||||
text = ""
|
||||
for i in range(1, sheet.max_row):
|
||||
# print(str(sheet.cell(i, 1).value))
|
||||
addr = read_cell(sheet, i, 8)
|
||||
name = read_cell(sheet,i, 2)
|
||||
datatype = read_cell(sheet,i, 4)
|
||||
unit = read_cell(sheet, i, 5)
|
||||
remark = read_cell(sheet, i, 6)
|
||||
remark = name + remark
|
||||
|
||||
if (len(addr) == 6):
|
||||
if (len(unit)>0):
|
||||
remark += '(' + unit + ')'
|
||||
if (len(text)>0):
|
||||
text += ',\n'
|
||||
text += '\t\t\t{"key": "%s", "datatype": "%s", "remark": "%s"}' % (addr, datatype, remark.replace("\n", ""))
|
||||
|
||||
|
||||
def parse_from_file(filename):
|
||||
data = ""
|
||||
with open(filename, "r", encoding='utf-8') as f: # 打开文件
|
||||
linedata = ""
|
||||
for line in f: # 行遍历
|
||||
line = line.strip()
|
||||
#linedata = f.readline() # 读取文件的一行
|
||||
flag = bool(re.search(r'0x[0-9A-Fa-f]{4}', line))
|
||||
linedata += line.strip()
|
||||
if flag:
|
||||
# print(linedata)
|
||||
# left, separator, right = linedata.partition("0x")
|
||||
parts = re.split(r'(0x[0-9A-Fa-f]{4})', linedata)
|
||||
linedata = ""
|
||||
if (len(text)>0):
|
||||
text = "\n" + text + "\n\t\t"
|
||||
text = '\t"%s":{\n\t\t"addr":[%s]\n\t}' % (topic, text)
|
||||
return text
|
||||
|
||||
|
||||
key = parts[1]
|
||||
datatype, remark = parse_datatype(parts[0].strip())
|
||||
remark = remark.replace("\t", " ").replace("\"", "")
|
||||
|
||||
item = {}
|
||||
item["key"] = key
|
||||
item["datatype"] = datatype
|
||||
item["remark"] = remark
|
||||
|
||||
if len(data) > 0:
|
||||
data += ",\n"
|
||||
data += ("\t\t\t" + json.dumps(item, ensure_ascii=False))
|
||||
return data
|
||||
|
||||
|
||||
addritems = {}
|
||||
addritems["EMS_YT"] = ["EMS遥调.txt", 1]
|
||||
addritems["BCU_YX"] = ["BCU电池簇遥信.txt", 1]
|
||||
addritems["BCU_YC"] = ["BCU电池簇遥测.txt", 1]
|
||||
addritems["BMS_YC"] = ["BMS电池堆遥测.txt", 1]
|
||||
addritems["EMS_YX"] = ["EMS遥信.txt", 1]
|
||||
addritems["EMS_YC"] = ["EMS遥测.txt", 1]
|
||||
addritems["EMS_YT"] = ["EMS遥调.txt", 1]
|
||||
addritems["PCS_YX"] = ["PCS遥信.txt", 1]
|
||||
addritems["PCS_YC"] = ["PCS遥测.txt", 1]
|
||||
addritems["PCU_YX"] = ["PCU遥信.txt", 1]
|
||||
addritems["PCU_YC"] = ["PCU遥测.txt", 1]
|
||||
addritems["MEM_YC"] = ["多功能电表遥测.txt", 1]
|
||||
addritems["TH_YC"] = ["温湿度状态遥测.txt", 1]
|
||||
addritems["Fire40_YX"] = ["消防4.0遥信.txt", 1]
|
||||
addritems["Cooling_YX"] = ["冷机遥信.txt", 1]
|
||||
addritems["Cooling_YC"] = ["冷机遥测.txt", 1]
|
||||
wb = load_workbook('EMU对外通信点表最终修改1版_v9.xlsx', data_only=True)
|
||||
|
||||
text = ""
|
||||
text = read_sheet(wb, "EMS_YT", "EMS遥调")
|
||||
text += ',\n' + read_sheet(wb, "EMS_YX", "EMS遥信")
|
||||
text += ',\n' + read_sheet(wb, "PCU_YC", "PCU遥测")
|
||||
text += ',\n' + read_sheet(wb, "PCU_YX", "PCU遥信")
|
||||
text += ',\n' + read_sheet(wb, "PCS_YC", "PCS遥测")
|
||||
text += ',\n' + read_sheet(wb, "PCS_YX", "PCS遥信")
|
||||
text += ',\n' + read_sheet(wb, "BMS_YC", "BMS电池堆遥测")
|
||||
text += ',\n' + read_sheet(wb, "BCU_YC", "BCU电池簇遥测")
|
||||
text += ',\n' + read_sheet(wb, "BCU_YX", "BCU电池簇遥信")
|
||||
# text += ',\n' + read_sheet(wb, "AirC_YC", "空调遥测")
|
||||
# text += ',\n' + read_sheet(wb, "AirC_YX", "空调遥信")
|
||||
text += ',\n' + read_sheet(wb, "MEM_YC", "多功能电表遥测")
|
||||
text += ',\n' + read_sheet(wb, "TH_YC", "温湿度状态遥测")
|
||||
text += ',\n' + read_sheet(wb, "Fire40_YX", "消防4.0遥信")
|
||||
text += ',\n' + read_sheet(wb, "Cooling_YC", "冷机遥测")
|
||||
text += ',\n' + read_sheet(wb, "Cooling_YX", "冷机遥信")
|
||||
|
||||
with open('registeraddr.json', 'w', encoding='utf-8') as f:
|
||||
f.write("{")
|
||||
index = 0
|
||||
for key in addritems:
|
||||
filename = addritems[key][0]
|
||||
count = addritems[key][1]
|
||||
print("parse: ", key, filename)
|
||||
data = parse_from_file(filename)
|
||||
if len(data) > 0:
|
||||
data = "\n" + data + "\n\t"
|
||||
if index != 0:
|
||||
f.write(",")
|
||||
f.write("\n\t\"" + key + "\": {\n\t\t\"count\":" + str(count) + ",\n\t\t\"addr\":[" + data + "\t]\n\t}")
|
||||
index+=1
|
||||
f.write("\n}")
|
||||
f.write("{\n" + text + "\n}")
|
||||
|
||||
|
||||
def read_sheet_alarm(wb, device_type, sht_name):
|
||||
sheet = wb[sht_name]
|
||||
text = ""
|
||||
for i in range(1, sheet.max_row):
|
||||
addr = read_cell(sheet, i, 8)
|
||||
is_alarm = (read_cell(sheet, i, 7) == "告警")
|
||||
|
||||
if (len(addr) == 6 and is_alarm) :
|
||||
name = read_cell(sheet,i, 2)
|
||||
datatype = read_cell(sheet,i, 4)
|
||||
unit = read_cell(sheet, i, 5)
|
||||
remark = read_cell(sheet, i, 6)
|
||||
remark = name + remark
|
||||
|
||||
if (len(unit)>0):
|
||||
remark += '(' + unit + ')'
|
||||
if (len(text)>0):
|
||||
text += ',\n'
|
||||
text += '\t\t\t{"key": "%s", "datatype": "%s", "remark": "%s"}' % (addr, datatype, remark.replace("\n", ""))
|
||||
if (len(text)>0):
|
||||
text = "\n" + text + "\n\t\t"
|
||||
text = '\t"%s":{\n\t\t"addr":[%s]\n\t}' % (device_type, text)
|
||||
return text;
|
||||
|
||||
text_err = ""
|
||||
text_err = read_sheet_alarm(wb, 103, "PCU遥信")
|
||||
text_err += ',\n' + read_sheet_alarm(wb, 104, "PCS遥信")
|
||||
with open('registeraddrErr.json', 'w', encoding='utf-8') as f:
|
||||
f.write("{\n" + text_err + "\n}")
|
||||
|
||||
Reference in New Issue
Block a user