Python模拟串口通信的示例详解

import serial
import serial.tools.list_ports
 
# 获取所有串口设备实例。

port_list = list(serial.tools.list_ports.comports())
def show_all_com():
    # 如果找到串口设备，则依次输出每个设备对应的串口号和描述信息。
    port_list_name = []
    if len(port_list) <= 0:
        print("the serial port can't find!")
    else:
        for itms in port_list:
            print(list(itms)[0], list(itms)[1])
            port_list_name.append(itms.device)

    return port_list_name

import serial
 
# 打开 COM3，将波特率配置为115200，数据位为7，停止位为2，无校验位，读超时时间为0.5秒。
ser = serial.Serial(port="COM3",
                    baudrate=115200,
                    bytesize=serial.SEVENBITS,
                    parity=serial.PARITY_NONE,
                    stopbits=serial.STOPBITS_TWO,
             php       timeout=0.5) 

import serial
from time import sleep
import queue
import time
from datetime import datetime

q = queue.Queue()

class SerialPort:
    def __init__(self, port, buand):
        self.port = serial.Serial(port, buand)
        self.port.close()
        if not self.port.isOpen():
            self.port.open()

    def port_open(self):
        if not self.port.isOpen():
            self.port.open()

    def port_close(self):
        self.port.close()

    #发送数据
    def send_data(self):
        """
        #write() 方法只能发送 bytes 类型的数据，所以需要对字符串进行 encode 编码。
        # write() 方法执行完成后，会将发送的字节数作为返回值。
        """
        while True:   # 循环发送数据
            date = datetime.now().strftime('%H:%M:%S.%f')[:-3]
            print("send:", date)
            self.China编程port.write(date.encode())
            time.sleep(1)

    # 读取数据
    def read_data(self):
        """
        #  read()方法默认一次读取一个字节，可以通过传入参数指定每次读取的字节数；
        read() 方法会将读取的内容作为返回值，类型为 bytes。
        """
        while True:
            count = self.port.inWaiting()
            if count != 0:
                recv = self.port.read(count)
                self.port.flushInput()
                q.put(recv.decode())
                print("receive:",recv.decode())
            sleep(0.5)

    #解析保存数据
    def save_json(self):
        while True:
            if q.empty():
                pass
            else:
                print("队列长度",q.qsize())
                data = q.get()
                print(data)
            sleep(0.1)

from tools.LoRaSerial import SerialPort
import threading
from time import sleep

if __name__=='__main__':
    baunRate = 115200
    serialPort_w = "COM3"
    mSerial_w = SerialPort(serialPort_w, baunRate)
    t1 = threading.Thread(target=mSerial_w.send_data)
    t1.daemon = True
    t1.start()

    while True:
        sleep(10)

from tools.LoRaSerial import SerialPort
import threading
from time import sleep

if __name__=='__main__':
    threads=[]
    baunRate = 115200

    serialPort_r = "COM4"
    mSerial_r = SerialPort(serialPort_r, baunRate)

    t2 = threading.Thread(target=mSerial_r.read_data)
    threads.append(t2)

    t3 = threading.Thread(target=mSerial_r.save_json)
    threads.append(t3)

    for t in threads:
        t.start()

    while True:
        sleep(10)