python3 通过serial读取和发送串口数据进行485口稳定性测试

公司的设备有4个485的串口，近期需要做在做高低温测试时，验证458口的稳定性。在网上查阅了一下资料，方案大体以下两类：1.单个485口，通过485转usb在windows系统上进行收发；2.有多个485口的情况下，对485进行短接，在本地进行收发；综合考虑后采样在本地进行短接的方式进行测试；
485口设备，通过shell命令进行读写，非常简单。在/dev/下面找到对应485口的设备文件，类似于ttyS1,ttyS2这样的文件。1.通过 cat /dev/ttyS1 读取数据；2. 发送数据，echo 1111 > /dev/ttyS2 ；可以写个简单是shell脚本进行测试，此处不行扩展；
我们的设备里面，有python3的环境，因此用python实现了一个对4个口进行测试脚本；我设备单次发送的数据在32个字节的样子，发送间隔1分钟。测试脚本的压力是，单次发送48个字节，发送间隔是100ms。当485口通信异常时，会打印到控制台。测试时通过nohup 命令将控制台输出重定向到文件进行记录。代码如下：

# -*- coding:utf-8 -*-
# @Time      :2020/6/1
# 485口检测工具
# 依赖 pyserial 模块 安装方式 pip install pyserial
# 1.将485口A对A,B对B 两两进行短接,本机进行测试
# 2 也可以外接设备进行测试485口进行收发测试

from sys import argv
import serial
import time
import _thread
from multiprocessing import Pool, freeze_support

# 串口编号和对应设备文件的字典

port_dict = {
    '1': "/dev/ttyS7",
    '2': "/dev/ttyS6",
    '3': "/dev/ttyS3",
    '4': "/dev/ttyS4",
}


def recv_data(serial_FD, port_name=''):
    """
    接收函数
    :param serial_FD:
    :param port_name:串口编号
    :return:
    """
    print(f'{time.strftime("%Y-%m-%d %H:%M;%S", time.localtime())},runing recv thread[{port_name}] ,waiting 0.1s')
    time.sleep(0.1)
    while True:
        try:
            data_count = serial_FD.inWaiting()
            if data_count != 0:
                recv = serial_FD.read(48)
                # print(time.strftime("%Y-%m-%d %H:%M;%S", time.localtime()), " ---  data_recv lenth  --> ",
                #       len(recv.decode()))
                pass
            else:
                print(
                    f'{time.strftime("%Y-%m-%d %H:%M;%S", time.localtime())} ---  data_recv[{port_name}]   --> recv is error')
        except Exception as e:
            print(str(e))
        time.sleep(0.1)


def send_data(serial_FD):
    """
    发送方法
    :param serial_FD:
    :return:
    """
    # print(f'runing recv thread {time.strftime("%Y-%m-%d %H:%M;%S", time.localtime())},waiting 30s')
    # time.sleep(30)

    while 1:
        try:
            serial_FD.write(b'1' * 48)
        except Exception as e:
            print(str(e))
        time.sleep(0.1)


def run(port=None):
    """
    程序入口
    :return:
    """
    if port is None:
        if len(argv) != 2:
            print('Please enter the port number! eg: ptyhon3 rs485_tool.py 1')
            exit(-1)
        port = argv[1]

    serial_port = port_dict.get(str(port))
    if serial_port is None:
        print('port not in [1,2,3,4]')
        exit(-1)

    data_ser = serial.Serial(serial_port, 9600, timeout=5)
    data_ser.flushInput()
    _thread.start_new_thread(recv_data, (data_ser, port))
    send_data(data_ser)


def multi_run(port_list=''):
    pool = Pool(len(port_list))
    for port in port_list:
        pool.apply_async(run, args=(port,))
    pool.close()
    pool.join()


if __name__ == '__main__':
    freeze_support()
    multi_run([1, 2, 3, 4])
    # run()