DebugTool/source/dev_Lamina.py

import time
from pathlib import Path
from serial import Serial
from utl import trans_list_to_str
from func_frame import make_frame_dlt645, check_frame_dlt645

class LaminaAdapter:
    def __init__(self, com_name="COM16", addr_645=[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA], addr_modbus=0x01):
        # 初始化串口通信
        if com_name is not None:
            self.com = Serial(com_name, baudrate=115200, parity='N', timeout=2)
        else:
            self.com =None

        # 设置645协议地址
        self.addr_645 = addr_645
        # 设置Modbus地址
        self.addr_modbus = addr_modbus

    def frame_read(self, daddr=0x60, dlen=0x50):
        block_modbus = {
            'addr_dev' : self.addr_modbus,
            'data_addr' : daddr,
            'data_len' : dlen,
            'type' : 'read',
        }
        block_dlt645 = {
            'addr' : self.addr_645,
            'type' : 'modbus',
            'data' : block_modbus,
        }
        frame = make_frame_dlt645(block_dlt645)

        if self.com is None:
            print(trans_list_to_str(frame))
            return
        
        self.com.read_all()
        self.com.write(bytearray(frame))
        time.sleep(0.5)
        frame_recv = self.com.read_all()
        output_text = check_frame_dlt645(frame_recv, block=block_dlt645)
        print(output_text)

    def frame_write_one(self, daddr=0x85, dval=-900):
        block_modbus = {
            'addr_dev' : self.addr_modbus,
            'data_addr' : daddr,
            'data_val' : dval,
            'type' : 'write_one',
        }
        block_dlt645 = {
            'addr' : self.addr_645,
            'type' : 'modbus',
            'data' : block_modbus,
        }
        frame = make_frame_dlt645(block_dlt645)
        if self.com is None:
            print(trans_list_to_str(frame))
            return
        
        self.com.write(bytearray(frame))

    def frame_write_dual(self, daddr=0x91, dval=600):
        block_modbus = {
            'addr_dev' : self.addr_modbus,
            'data_addr' : daddr,
            'data_val' : dval,
            'type' : 'write_dual',
        }
        block_dlt645 = {
            'addr' : self.addr_645,
            'type' : 'modbus',
            'data' : block_modbus,
        }
        frame = make_frame_dlt645(block_dlt645)
        if self.com is None:
            print(trans_list_to_str(frame))
            return
        
        self.com.write(bytearray(frame))

    def frame_write_str(self, daddr=0x82, dval=[0x06, 0x05, 0x04, 0x03, 0x02, 0x01]):
        block_modbus = {
            'addr_dev' : self.addr_modbus,
            'data_addr' : daddr,
            'data_val' : dval,
            'type' : 'write_str',
        }
        block_dlt645 = {
            'addr' : self.addr_645,
            'type' : 'modbus',
            'data' : block_modbus,
        }
        frame = make_frame_dlt645(block_dlt645)
        if self.com is None:
            print(trans_list_to_str(frame))
            return
        
        self.com.write(bytearray(frame))

    def frame_update(self, path_bin):
        """ 程序升级
            注意: 在使用单板升级测试时, 需要关闭低电压检测功能, 否则无法启动升级流程;

        """
        block_modbus = {
            'addr_dev' : self.addr_modbus,
            'type' : 'update',
            'step' : 'start',
            'index' : 0,
            'file' : Path(path_bin).read_bytes(),
            'header_offset': 128,
        }
        block_dlt645 = {
            'addr' : self.addr_645,
            'type' : 'modbus',
            'data' : block_modbus,
        }
        # 启动帧
        frame_master = bytearray(make_frame_dlt645(block_dlt645))

        # 等待擦除完成返回
        try_times = 3000
        self.com.read_all()
        while try_times:
            time.sleep(0.4)
            self.com.write(frame_master)
            frame_slave = self.com.read_all()
            if not frame_slave:
                try_times -= 1
                continue
            
            block_dlt645["data"]['file_block_size'] = check_frame_dlt645(frame_slave, block_dlt645)
            break

        if block_dlt645["data"]['file_block_size'] == 0:
            raise Exception("Error slave response.")

        # 避免接收到延迟返回报文
        time.sleep(0.4)

        # 文件传输
        block_dlt645["data"]['step'] = 'trans'
        data_remain = len(block_dlt645["data"]['file']) - block_dlt645['data']['header_offset']
        while data_remain > 0:
            frame_master = bytearray(make_frame_dlt645(block_dlt645))

            self.com.read_all()
            self.com.write(frame_master)
            time.sleep(0.2)
            frame_slave = None
            while not frame_slave:
                frame_slave = self.com.read_all()
            check_frame_dlt645(frame_slave, block_dlt645)
            
            block_dlt645["data"]['index'] += 1
            data_remain -= block_dlt645["data"]['file_block_size']

        # 结束升级
        block_dlt645["data"]['step'] = 'end'
        frame_master = bytearray(make_frame_dlt645(block_dlt645))

        self.com.read_all()
        self.com.write(frame_master)
        time.sleep(0.1)
        frame_slave = None
        while not frame_slave:
            frame_slave = self.com.read_all()
        check_frame_dlt645(frame_slave[:18], block_dlt645)

if __name__=='__main__':
    dev_lamina = LaminaAdapter()

    dev_lamina.frame_read(0x0100, 0x20)