串口通信案例

wen java案例 2026-08-01 190

本文目录导读：

串口通信案例

Python 串口通信案例(PySerial)
C++ 串口通信案例(Win32 API)
单片机串口通信(STC89C52)
串口调试工具案例(C# WinForms)
串口通信协议示例(Modbus RTU)
跨平台串口库(使用python-pyserial)
实用调试技巧

我将为您提供几个常见的串口通信案例,覆盖不同编程语言和应用场景。

Python 串口通信案例(PySerial)

基础读取示例

import serial
import time
# 打开串口
ser = serial.Serial(
    port='COM3',        # Windows端口,Linux为'/dev/ttyUSB0'
    baudrate=9600,      # 波特率
    bytesize=serial.EIGHTBITS,  # 数据位
    parity=serial.PARITY_NONE,   # 校验位
    stopbits=serial.STOPBITS_ONE, # 停止位
    timeout=1           # 超时时间(秒)
)
# 检查串口是否打开
if ser.isOpen():
    print("串口打开成功")
    # 发送数据
    ser.write(b'Hello World')
    # 读取数据
    data = ser.read(10)
    print(f"接收数据: {data}")
    # 读取一行
    line = ser.readline()
    print(f"接收行: {line}")
# 关闭串口
ser.close()

数据收发循环示例

import serial
import threading
import time
class SerialCommunication:
    def __init__(self, port, baudrate=115200):
        self.ser = serial.Serial(port, baudrate, timeout=0.5)
        self.running = False
        self.receive_thread = None
    def start(self):
        """启动接收线程"""
        self.running = True
        self.receive_thread = threading.Thread(target=self._receive_data)
        self.receive_thread.daemon = True
        self.receive_thread.start()
    def _receive_data(self):
        """接收数据线程"""
        while self.running:
            if self.ser.in_waiting:
                data = self.ser.read(self.ser.in_waiting)
                print(f"收到: {data.hex(' ')}")
            time.sleep(0.01)
    def send_data(self, data):
        """发送数据"""
        if isinstance(data, str):
            data = data.encode('utf-8')
        self.ser.write(data)
        print(f"发送: {data.hex(' ')}")
    def stop(self):
        """停止通信"""
        self.running = False
        if self.receive_thread:
            self.receive_thread.join()
        self.ser.close()
# 使用示例
if __name__ == "__main__":
    comm = SerialCommunication('COM3', 9600)
    comm.start()
    try:
        while True:
            comm.send_data("Hello设备")
            time.sleep(2)
    except KeyboardInterrupt:
        comm.stop()

C++ 串口通信案例(Win32 API)

#include <windows.h>
#include <iostream>
#include <string>
class SerialPort {
private:
    HANDLE hSerial;
    DCB dcbSerialParams;
public:
    bool open(const std::string& portName, DWORD baudRate) {
        // 打开串口
        hSerial = CreateFile(
            portName.c_str(),
            GENERIC_READ | GENERIC_WRITE,
            0,
            NULL,
            OPEN_EXISTING,
            FILE_ATTRIBUTE_NORMAL,
            NULL
        );
        if (hSerial == INVALID_HANDLE_VALUE) {
            std::cout << "无法打开串口" << std::endl;
            return false;
        }
        // 获取当前参数
        if (!GetCommState(hSerial, &dcbSerialParams)) {
            std::cout << "获取串口参数失败" << std::endl;
            return false;
        }
        // 设置参数
        dcbSerialParams.BaudRate = baudRate;
        dcbSerialParams.ByteSize = 8;
        dcbSerialParams.Parity = NOPARITY;
        dcbSerialParams.StopBits = ONESTOPBIT;
        if (!SetCommState(hSerial, &dcbSerialParams)) {
            std::cout << "设置串口参数失败" << std::endl;
            return false;
        }
        // 设置超时
        COMMTIMEOUTS timeouts = {0};
        timeouts.ReadIntervalTimeout = 50;
        timeouts.ReadTotalTimeoutMultiplier = 10;
        timeouts.ReadTotalTimeoutConstant = 100;
        timeouts.WriteTotalTimeoutMultiplier = 10;
        timeouts.WriteTotalTimeoutConstant = 100;
        SetCommTimeouts(hSerial, &timeouts);
        return true;
    }
    bool sendData(const char* data, int length) {
        DWORD bytesWritten;
        return WriteFile(hSerial, data, length, &bytesWritten, NULL);
    }
    int receiveData(char* buffer, int bufferLen) {
        DWORD bytesRead;
        if (ReadFile(hSerial, buffer, bufferLen, &bytesRead, NULL)) {
            return bytesRead;
        }
        return -1;
    }
    void close() {
        CloseHandle(hSerial);
    }
};
int main() {
    SerialPort serial;
    if (serial.open("COM3", 9600)) {
        std::cout << "串口打开成功" << std::endl;
        // 发送数据
        const char* msg = "Hello from C++";
        serial.sendData(msg, strlen(msg));
        // 接收数据
        char buffer[256];
        int received = serial.receiveData(buffer, sizeof(buffer));
        if (received > 0) {
            buffer[received] = '\0';
            std::cout << "接收: " << buffer << std::endl;
        }
        serial.close();
    }
    return 0;
}

单片机串口通信(STC89C52)

// 51单片机串口通信示例
#include <reg52.h>
#define FOSC 11059200L
#define BAUD 9600
void UART_Init() {
    SCON = 0x50;           // 模式1,8位UART,允许接收
    TMOD = 0x20;           // 定时器1工作在模式2(8位自动重载)
    // 设置波特率
    TH1 = TL1 = 256 - FOSC / 12 / 32 / BAUD;
    TR1 = 1;               // 启动定时器1
    ES = 1;                // 使能串口中断
    EA = 1;                // 使能总中断
}
void UART_SendByte(unsigned char dat) {
    SBUF = dat;
    while (!TI);           // 等待发送完成
    TI = 0;
}
void UART_SendString(unsigned char *str) {
    while (*str) {
        UART_SendByte(*str++);
    }
}
// 串口中断服务函数
void UART_ISR() interrupt 4 {
    unsigned char recv;
    if (RI) {               // 接收中断
        RI = 0;             // 清除标志
        recv = SBUF;        // 读取数据
        // 回显数据
        UART_SendByte(recv);
        // 如果是特定指令则执行
        if (recv == 'A') {
            P1 = 0x00;      // 控制LED
        } else if (recv == 'B') {
            P1 = 0xFF;
        }
    }
}
void main() {
    UART_Init();
    UART_SendString("MCU Ready\r\n");
    while (1) {
        // 主循环
    }
}

串口调试工具案例(C# WinForms)

using System;
using System.IO.Ports;
using System.Windows.Forms;
public partial class SerialForm : Form
{
    SerialPort serialPort = new SerialPort();
    Timer timer = new Timer();
    public SerialForm()
    {
        InitializeComponent();
        LoadPorts();
    }
    private void LoadPorts()
    {
        // 获取可用串口
        string[] ports = SerialPort.GetPortNames();
        comboBoxPort.Items.AddRange(ports);
        comboBoxPort.SelectedIndex = 0;
        // 波特率选项
        comboBoxBaud.Items.AddRange(new object[] { 
            9600, 19200, 38400, 57600, 115200 });
        comboBoxBaud.SelectedIndex = 0;
    }
    private void btnOpen_Click(object sender, EventArgs e)
    {
        try 
        {
            serialPort.PortName = comboBoxPort.Text;
            serialPort.BaudRate = int.Parse(comboBoxBaud.Text);
            serialPort.DataBits = 8;
            serialPort.Parity = Parity.None;
            serialPort.StopBits = StopBits.One;
            serialPort.Open();
            serialPort.DataReceived += new SerialDataReceivedEventHandler(DataReceived);
            btnOpen.Enabled = false;
            btnClose.Enabled = true;
            txtLog.AppendText("串口已打开\r\n");
        }
        catch (Exception ex)
        {
            MessageBox.Show("打开串口失败: " + ex.Message);
        }
    }
    private void DataReceived(object sender, SerialDataReceivedEventArgs e)
    {
        // 在UI线程中更新
        this.Invoke(new Action(() => {
            string data = serialPort.ReadExisting();
            txtLog.AppendText("接收: " + data + "\r\n");
        }));
    }
    private void btnSend_Click(object sender, EventArgs e)
    {
        if (serialPort.IsOpen)
        {
            serialPort.Write(txtSend.Text);
            txtLog.AppendText("发送: " + txtSend.Text + "\r\n");
        }
    }
}

串口通信协议示例(Modbus RTU)

import serial
import struct
import time
class ModbusRTU:
    def __init__(self, port, baudrate=9600):
        self.ser = serial.Serial(port, baudrate, timeout=1)
    def calculate_crc(self, data):
        """计算Modbus CRC16"""
        crc = 0xFFFF
        for byte in data:
            crc ^= byte
            for _ in range(8):
                if crc & 0x0001:
                    crc = (crc >> 1) ^ 0xA001
                else:
                    crc >>= 1
        return crc
    def read_holding_registers(self, slave_id, start_reg, count):
        """读取保持寄存器"""
        # 构建请求: 设备地址 + 功能码(03) + 起始地址 + 寄存器数量 + CRC
        request = struct.pack('>B B H H', slave_id, 0x03, start_reg, count)
        crc = self.calculate_crc(request)
        request += struct.pack('<H', crc)
        # 发送请求
        self.ser.write(request)
        # 等待响应
        time.sleep(0.05)
        # 读取响应
        response = self.ser.read(5 + count * 2 + 2)
        if len(response) < 5:
            return None
        # 验证CRC
        response_crc = struct.unpack('<H', response[-2:])[0]
        calc_crc = self.calculate_crc(response[:-2])
        if response_crc != calc_crc:
            print("CRC校验失败")
            return None
        # 解析数据
        data = []
        for i in range(count):
            register_value = struct.unpack('>H', response[3 + i*2:5 + i*2])[0]
            data.append(register_value)
        return data
    def write_single_register(self, slave_id, reg_address, value):
        """写单个寄存器"""
        # 构建请求: 地址 + 功能码(06) + 寄存器地址 + 值 + CRC
        request = struct.pack('>B B H H', slave_id, 0x06, reg_address, value)
        crc = self.calculate_crc(request)
        request += struct.pack('<H', crc)
        self.ser.write(request)
    def close(self):
        self.ser.close()
# 使用示例
if __name__ == "__main__":
    modbus = ModbusRTU('COM3', 9600)
    try:
        # 读取从站1,从寄存器0读取5个寄存器
        data = modbus.read_holding_registers(1, 0, 5)
        print(f"读取结果: {data}")
        # 写寄存器
        modbus.write_single_register(1, 10, 100)
    finally:
        modbus.close()

跨平台串口库(使用python-pyserial)

import serial
import serial.tools.list_ports
def list_available_ports():
    """列出所有可用的串口"""
    ports = serial.tools.list_ports.comports()
    for port in ports:
        print(f"端口: {port.device}")
        print(f"描述: {port.description}")
        print(f"硬件ID: {port.hwid}")
        print("-" * 40)
    return ports
def find_arduino():
    """查找Arduino设备"""
    ports = serial.tools.list_ports.comports()
    for port in ports:
        if "Arduino" in port.description:
            return port.device
    return None
# 字符设备配置示例
import termios
import tty
import sys
def configure_raw_mode(serial_port):
    """配置Linux下的原始模式"""
    fd = serial_port.fileno()
    old_settings = termios.tcgetattr(fd)
    try:
        tty.setraw(fd)
        # 配置串口
        serial_port.write(b'AT\r\n')
        response = serial_port.readline()
        print(f"响应: {response}")
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)

实用调试技巧

# 串口调试辅助函数
import serial
import time
import binascii
class SerialDebugger:
    def __init__(self, port, baudrate):
        self.ser = serial.Serial(port, baudrate)
        self.log_file = open('serial_log.txt', 'a')
    def hex_dump(self, data):
        """十六进制格式化输出"""
        return binascii.hexlify(data).decode('ascii')
    def log_data(self, direction, data):
        """记录收发数据"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        hex_data = self.hex_dump(data)
        ascii_data = data.decode('ascii', errors='replace')
        log_line = f"[{timestamp}] {direction}: HEX={hex_data} ASCII={ascii_data}\n"
        print(log_line.strip())
        self.log_file.write(log_line)
    def send_and_receive(self, data, timeout=2):
        """发送并接收"""
        self.log_data("TX", data)
        self.ser.write(data)
        start_time = time.time()
        response = b''
        while time.time() - start_time < timeout:
            if self.ser.in_waiting:
                byte = self.ser.read()
                response += byte
        self.log_data("RX", response)
        return response
# 使用示例
debugger = SerialDebugger('COM3', 9600)
response = debugger.send_and_receive(b'\x01\x03\x00\x00\x00\x02\xC4\x0B')

这些案例覆盖了串口通信的主要应用场景：