linghtiin/DebugTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复raise语句用法;

Browse Source 
添加float数据解析;
更新升级传输方案;
This commit is contained in:
何 泽隆
2024-05-21 20:20:23 +08:00
parent 2582721652
commit 5cf4246338
4 changed files with 80 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
source/dev_EnergyRouter.py
View File

@@ -58,12 +58,12 @@ class EnergyRouter:
self.tcp_socket.send(frame_master)
frame_slave = self.tcp_socket.recv(32)
if frame_slave == '':
raise("TCP closed.")
raise Exception("TCP closed.")
self.block['file_block_size'] = check_frame_modbus(frame_slave, self.block)
if self.block['file_block_size'] == 0:
raise("Error slave response.")
raise Exception("Error slave response.")
# 避免接收到延迟返回报文
time.sleep(0.4)
@@ -96,10 +96,10 @@ class EnergyRouter:
seq_frame_slave[i] = self.tcp_socket.recv(8)
# 接收到空数据, 对端已关闭连接
if seq_frame_slave[i] == '':
raise("TCP closed.")
raise Exception("TCP closed.")
result, seq_current, seq_hope = check_frame_modbus(seq_frame_slave[i], None)
if seq_current < seq_offset:
raise("Error.")
raise Exception("Error.")
elif result:
seq_window[seq_current - seq_offset] = 2
data_remain -= self.block['file_block_size']
@@ -129,7 +129,7 @@ class EnergyRouter:
self.tcp_socket.send(frame_master)
frame_slave = self.tcp_socket.recv(8)
if frame_slave == '':
raise("TCP closed.")
raise Exception("TCP closed.")
ret = check_frame_modbus(frame_slave[:18], self.block)

34
source/dev_Lamina.py
View File

@@ -10,6 +10,7 @@ modbus_map = {
# 3 - lnt32
# 4 - str
# 5 - addr
# 6 - float
0x0E: ["故障字1", 1],
0x0F: ["故障字2", 1],
0x10: ["MPPT工作状态", 1],
@@ -25,8 +26,8 @@ modbus_map = {
0x1A: ["输入功率", 3],
0x1C: ["设备温度", 2],
0x1D: ["开关机状态", 1],
0x1E: ["保留", 1],
0x1F: ["保留", 1],
0x1E: ["电池电压", 6],
0x20: ["并机功率限值", 6],
0x60: ["光伏通道使能", 1],
0x61: ["最小启动输入电压", 2],
@@ -100,6 +101,11 @@ modbus_map = {
0xAF: ["保留", 1],
0x100: ["版本", 4, 16],
0x110: ["型号", 4, 16],
0x120: ["硬件", 4, 16],
0x130: ["SN", 4, 16],
0x140: ["MES", 4, 16],
0x150: ["Datetime", 4, 16],
0x160: ["厂商", 4, 16],
}
@@ -184,7 +190,7 @@ class LaminaAdapter:
self.block['data']['step'] = 'start'
self.block['data']['index'] = 0
self.block['data']['file'] = Path(path_bin).read_bytes()
self.block['data']['header_offset'] = 128
self.block['data']['header_offset'] = 184
# 启动帧
frame_master = bytearray(make_frame_dlt645(self.block))
@@ -203,7 +209,7 @@ class LaminaAdapter:
break
if self.block["data"]['file_block_size'] == 0:
raise("Error slave response.")
raise Exception("Error slave response.")
# 避免接收到延迟返回报文
time.sleep(0.4)
@@ -235,10 +241,26 @@ class LaminaAdapter:
frame_slave = None
while not frame_slave:
frame_slave = self.com.read_all()
check_frame_dlt645(frame_slave[:18], self.block)
check_frame_dlt645(frame_slave[:20], self.block)
if __name__=='__main__':
dev_lamina = LaminaAdapter()
dev_lamina = LaminaAdapter(com_name="COM8")
dev_lamina.frame_read(0x0E, 0x14)
dev_lamina.frame_read(0x0100, 0x20)
if 0:
dev_lamina.frame_write_str(0x0130, list("SN20240107546"))
dev_lamina.frame_write_str(0x0140, list("MES20240107546"))
dev_lamina.frame_write_str(0x0150, list("D2024050101030"))
time.sleep(2)
dev_lamina.frame_read(0x0130, 0x30)
if not hasattr(__builtins__,"__IPYTHON__"):
# path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240517_1100_T1.10.bin")
path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\SLCP001_240520_0000_T1.11.bin")
# path_bin = Path(r"D:\WorkingProject\LightStackAdapter\software\lamina_adapter\tools\upgrade\DGAPD_240516_0000_V1.10.bin")
dev_lamina.frame_update(path_bin)

63
source/func_frame.py
View File

@@ -1,3 +1,4 @@
import struct
from crc import Calculator, Crc16
from utl import display_hex, trans_list_to_str
@@ -7,11 +8,13 @@ modbus_map = {
# 3 - lnt32
# 4 - str
# 5 - addr
# 6 - float
0x01: ["Hex示例", 1],
0x02: ["Int示例", 2],
0x03: ["Int32示例", 3],
0x04: ["str示例", 4, 16],
0x10: ["addr示例", 5, 6],
0x20: ["Int32示例", 6],
}
@@ -25,11 +28,13 @@ def make_frame_modbus(block:dict):
""" 升级系列报文 """
frame.append(0x07)
if len(block['file']) <= block["header_offset"]:
raise("Modbus Update error, file too small.")
raise Exception("Modbus Update error, file too small.")
if block['step'] == 'start':
frame.append(0x01)
frame.append(0x00)
frame.append(0x00)
frame.append(0x00)
frame.append(block['header_offset'] // 256)
frame.append(block['header_offset'] % 256)
frame += list(block['file'][:block['header_offset']])
@@ -46,8 +51,12 @@ def make_frame_modbus(block:dict):
elif block['step'] == 'end':
frame.append(0x03)
frame.append(0x00)
frame.append(block['index'] // 256)
frame.append(block['index'] % 256)
frame.append(0x00)
frame.append(0x00)
else:
raise("Modbus Update Frame Step Error.")
raise Exception("Modbus Update Frame Step Error.")
else:
""" 数据读取系列报文 """
data_addr = block['data_addr']
@@ -85,19 +94,26 @@ def make_frame_modbus(block:dict):
elif block['type'] == "write_str":
frame.append(0x10)
data_len = len(block['data_val'])
item_len = 2 * block['data_define'][data_addr][2]
data_val = block['data_val']
if data_len > 0x0100:
raise("Modbus data len oversize.")
if data_len > item_len:
raise Exception("Modbus data len oversize.")
elif data_len < item_len:
data_val += '\000' * (item_len - data_len)
data_len = len(data_val)
frame.append(data_addr // 256 % 256)
frame.append(data_addr % 256)
frame.append(0x00)
frame.append(data_len // 2)
frame.append(data_len)
if type(data_val[0]) == str:
data_val = list(map(lambda x: x.encode()[0], data_val))
for i in range(data_len//2):
frame.append(data_val[2*i + 1])
frame.append(data_val[2*i])
else:
raise("Modbus Frame Type Error.")
raise Exception("Modbus Frame Type Error.")
crc = calculator.checksum(bytearray(frame))
@@ -116,7 +132,7 @@ def make_frame_dlt645(block:dict):
data_frame = make_frame_modbus(block["data"])
len_data = len(data_frame)
else:
raise("Unknown dlt645 frame type.")
raise Exception("Unknown dlt645 frame type.")
frame.append(0x68)
frame += block['addr'][:6]
@@ -178,6 +194,11 @@ def display_data(modbus_map: dict, address: int, data: list):
data_len = current_map[2]
item = display_str(data, data_len)
item = trans_list_to_str(item)
elif current_map[1] == 6:
""" 浮点数值表示 """
temp = [data[2], data[3], data[0], data[1]]
item = struct.unpack('>f', bytes(temp))[0]
data_len = 2
if len_max < len(data_label):
len_max = len(data_label)
@@ -193,23 +214,23 @@ def check_frame_modbus(frame, block=None):
""" 校验modbus帧回复 """
calculator = Calculator(Crc16.MODBUS)
if calculator.checksum(frame[:-2]) != (frame[-1] * 0x100 + frame[-2]):
raise("Frame Crc check failed.")
raise Exception("Frame Crc check failed.")
elif frame[0] != 0x01:
raise("Frame device address error.")
raise Exception("Frame device address error.")
code_func = frame[1]
code_subfunc = frame[2]
if code_func == 0x07:
""" 升级回复帧 """
if code_subfunc == 0x01 and frame[3] == 0x00:
return frame[4] * 256 + frame[5]
return frame[6] * 256 + frame[7]
elif code_subfunc == 0x02:
""" 解析帧序号及返回值, 不做序列号校验 """
update_result = frame[3]
update_index = frame[4] * 256 + frame[5]
# 检查严重错误标志
if update_result == 0xFF:
raise("Error, Update abort.")
raise Exception("Error, Update abort.")
if update_result >= 0xA0 or update_result == 0x0F:
check_result = False
else:
@@ -225,7 +246,7 @@ def check_frame_modbus(frame, block=None):
return check_result, update_index, check_hope
elif code_subfunc == 0x03:
if frame[3] == 0xFF:
raise("Update verification failed.")
raise Exception("Update verification failed.")
elif frame[3] == 0x01:
print("Update done.")
elif frame[3] == 0x00:
@@ -234,7 +255,7 @@ def check_frame_modbus(frame, block=None):
print("Update unknown return value.")
return frame[3]
else:
raise("Func code or Return code error.")
raise Exception("Func code or Return code error.")
elif code_func == 0x03 or code_func == 0x04:
""" 数据读取帧 """
if frame[2] == len(frame[3:-2]):
@@ -244,27 +265,27 @@ def check_frame_modbus(frame, block=None):
else:
return list(frame[3:-2])
else:
raise("Frame read error.")
elif code_func == 0x06 and frame[2] == len(frame[3:-2]):
raise Exception("Frame read error.")
elif code_func == 0x06:
""" 单个数据写入帧 """
return frame[13:-4]
elif code_func == 0x10 and frame[2] == len(frame[3:-2]):
return frame[2:-2]
elif code_func == 0x10:
""" 多个数据写入帧 """
return frame[13:-4]
return frame[2:-2]
else:
raise(f"Frame Date error. func={code_func}, func_sub={code_subfunc}, len={len(frame)}")
raise Exception(f"Frame Date error. func={code_func}, func_sub={code_subfunc}, len={len(frame)}")
def check_frame_dlt645(frame, block=None):
""" 校验dlt645帧回复 """
if sum(frame[:-2]) % 0x100 != frame[-2]:
raise("DLT645 Frame Verify error.")
raise Exception("DLT645 Frame Verify error.")
elif len(frame[10:-2]) != frame[9]:
raise("DLT645 Frame len error.")
raise Exception("DLT645 Frame len error.")
code_func = frame[8]
if code_func == 0x9F:
block = block['data']
return check_frame_modbus(frame[10:-2], block)
else:
raise("DLT645 Frame type error.")
raise Exception("DLT645 Frame type error.")

6
source/utl_upgrade.py
View File

@@ -30,11 +30,11 @@ def file_IntelHex_to_Bin(file_data, base_address, len_max=0xC000, conv_end=False
max_address = 0
for line in lines:
if max_address >= len_max:
raise("Bin file Oversize.")
raise Exception("Bin file Oversize.")
if line[0] == ':':
checksum = sum([int(x, 16) * (15 * (~i & 0x01) + 1) for i, x in enumerate(line[1:])])
if (checksum & 0x00FF) != 0:
raise('Hex file checksum error!')
raise Exception('Hex file checksum error!')
if line[7:9] == '00':
len = int(line[1:3], 16)
@@ -170,7 +170,7 @@ def make_datafile2(file_path: str, config: dict, header_len=512):
data_header = build_header(config, header_len)
if data_header is None:
raise("Header tag oversize. ")
raise Exception("Header tag oversize. ")
data_encrypt = file_encryption(data_bin)
return data_header, data_bin, data_encrypt