Python 串口通信库介绍与使用指南
·
本文档系统介绍 Python 环境下常用的串口通信库，包含安装方法、API 说明及完整示例代码。
目录
1. 概述
串口通信（Serial Communication）是嵌入式开发、物联网、工业控制等领域中最基础的数据传输方式之一。Python 拥有成熟且易用的串口通信生态，其中最核心、最常用的库是 pySerial。
主要库对比
| 库名 | 特点 | 适用场景 |
|---|---|---|
pySerial |
最主流，跨平台，同步阻塞/非阻塞 | 常规串口读写 |
pySerial-asyncio |
基于 asyncio 的异步串口 | 高并发、异步架构 |
pyusb |
直接操作 USB 设备 | USB 转串口底层控制 |
2. pySerial 库
2.1 安装
pip install pyserial
2.2 基本 API
打开串口
import serial
# 方式一：直接初始化
ser = serial.Serial(
port='/dev/ttyUSB0', # Windows 用 'COM3'
baudrate=9600, # 波特率
bytesize=serial.EIGHTBITS, # 数据位
parity=serial.PARITY_NONE, # 校验位
stopbits=serial.STOPBITS_ONE, # 停止位
timeout=1 # 读取超时（秒）
)
# 方式二：先创建对象，再打开
ser = serial.Serial()
ser.port = 'COM3'
ser.baudrate = 115200
ser.open()
常用属性与方法
| 方法/属性 | 说明 |
|---|---|
ser.is_open |
串口是否打开 |
ser.write(data) |
发送数据（bytes） |
ser.read(size) |
读取指定字节数 |
ser.readline() |
读取一行（以 \n 结尾） |
ser.readlines() |
读取所有行 |
ser.in_waiting |
输入缓冲区字节数 |
ser.out_waiting |
输出缓冲区字节数 |
ser.flush() |
刷新输出缓冲区 |
ser.reset_input_buffer() |
清空输入缓冲区 |
ser.reset_output_buffer() |
清空输出缓冲区 |
ser.close() |
关闭串口 |
2.3 完整示例：基础读写
import serial
import time
def basic_serial_example():
"""
基础串口通信示例：打开串口、发送数据、接收数据
"""
try:
# 打开串口（根据系统修改端口名）
ser = serial.Serial(
port='/dev/ttyUSB0', # Linux/Mac
# port='COM3', # Windows
baudrate=9600,
timeout=1
)
print(f"串口已打开: {ser.port}")
print(f"波特率: {ser.baudrate}")
# 发送数据
message = b"Hello, Serial!\n"
bytes_written = ser.write(message)
print(f"已发送 {bytes_written} 字节: {message}")
# 等待设备响应
time.sleep(0.5)
# 读取响应
if ser.in_waiting > 0:
response = ser.read(ser.in_waiting)
print(f"收到响应: {response}")
# 关闭串口
ser.close()
print("串口已关闭")
except serial.SerialException as e:
print(f"串口错误: {e}")
except Exception as e:
print(f"其他错误: {e}")
if __name__ == '__main__':
basic_serial_example()
2.4 完整示例：循环读取
import serial
import time
def continuous_read_example():
"""
持续监听串口数据，直到收到退出指令
"""
ser = serial.Serial(
port='/dev/ttyUSB0',
baudrate=115200,
timeout=1
)
print("开始监听串口数据，输入 'exit' 退出...")
try:
while True:
if ser.in_waiting > 0:
# 读取一行数据
line = ser.readline().decode('utf-8').strip()
print(f"[{time.strftime('%H:%M:%S')}] 收到: {line}")
# 处理特定指令
if line.lower() == 'exit':
print("收到退出指令")
break
time.sleep(0.1) # 避免 CPU 占用过高
except KeyboardInterrupt:
print("\n用户中断")
finally:
ser.close()
print("串口已关闭")
if __name__ == '__main__':
continuous_read_example()
2.5 完整示例：十六进制通信
import serial
def hex_communication_example():
"""
十六进制数据通信示例（常用于 Modbus、传感器等协议）
"""
ser = serial.Serial(
port='/dev/ttyUSB0',
baudrate=9600,
timeout=1
)
# 构造十六进制指令（例如：读取传感器数据）
# 示例：Modbus RTU 读取保持寄存器
command = bytes([0x01, 0x03, 0x00, 0x00, 0x00, 0x02, 0xC4, 0x0B])
print(f"发送指令: {command.hex()}")
ser.write(command)
# 等待响应
import time
time.sleep(0.1)
# 读取响应
if ser.in_waiting > 0:
response = ser.read(ser.in_waiting)
print(f"收到响应: {response.hex()}")
print(f"响应长度: {len(response)} 字节")
ser.close()
if __name__ == '__main__':
hex_communication_example()
3. pySerial-asyncio 库
3.1 安装
pip install pyserial-asyncio
3.2 完整示例：异步串口通信
import asyncio
import serial_asyncio
class SerialProtocol(asyncio.Protocol):
"""自定义串口协议处理器"""
def __init__(self):
self.transport = None
self.buffer = b''
def connection_made(self, transport):
self.transport = transport
print(f"串口已连接: {transport.serial.port}")
def data_received(self, data):
self.buffer += data
print(f"收到数据: {data.decode('utf-8', errors='ignore').strip()}")
def connection_lost(self, exc):
print("串口连接已断开")
async def async_serial_example():
"""
异步串口通信示例
"""
loop = asyncio.get_event_loop()
# 创建串口连接
transport, protocol = await serial_asyncio.create_serial_connection(
loop,
SerialProtocol,
'/dev/ttyUSB0',
baudrate=9600
)
# 发送数据
transport.serial.write(b"Hello Async Serial!\n")
# 保持运行一段时间
await asyncio.sleep(10)
# 关闭连接
transport.close()
if __name__ == '__main__':
asyncio.run(async_serial_example())
4. serial.tools.list_ports 端口枚举
4.1 列出可用串口
import serial.tools.list_ports
def list_serial_ports():
"""
列出系统中所有可用的串口设备
"""
ports = serial.tools.list_ports.comports()
if not ports:
print("未找到可用串口")
return []
print(f"发现 {len(ports)} 个串口设备:")
print("-" * 50)
available_ports = []
for port in ports:
print(f"端口: {port.device}")
print(f" 描述: {port.description}")
print(f" 硬件ID: {port.hwid}")
print(f" 制造商: {port.manufacturer or '未知'}")
print(f" 序列号: {port.serial_number or '未知'}")
print()
available_ports.append(port.device)
return available_ports
if __name__ == '__main__':
list_serial_ports()
4.2 自动选择串口
import serial.tools.list_ports
def auto_select_port(keyword='USB'):
"""
根据关键词自动选择串口
"""
ports = serial.tools.list_ports.comports()
for port in ports:
if keyword.lower() in port.description.lower():
print(f"自动选择端口: {port.device} ({port.description})")
return port.device
print(f"未找到包含 '{keyword}' 的串口")
return None
if __name__ == '__main__':
port = auto_select_port('USB')
if port:
print(f"使用端口: {port}")
5. 其他串口库
5.1 minimalmodbus（Modbus RTU）
pip install minimalmodbus
import minimalmodbus
def modbus_example():
"""
使用 minimalmodbus 进行 Modbus RTU 通信
"""
# 创建仪器对象
instrument = minimalmodbus.Instrument('/dev/ttyUSB0', 1) # 端口, 从站地址
instrument.serial.baudrate = 9600
instrument.serial.timeout = 1
try:
# 读取保持寄存器（功能码 03）
value = instrument.read_register(0, 0) # 寄存器地址, 小数位数
print(f"寄存器 0 的值: {value}")
# 写入保持寄存器（功能码 06）
instrument.write_register(0, 100)
print("写入成功")
except Exception as e:
print(f"Modbus 错误: {e}")
finally:
instrument.serial.close()
if __name__ == '__main__':
modbus_example()
5.2 pyusb（底层 USB 控制）
pip install pyusb
import usb.core
import usb.util
def usb_serial_example():
"""
使用 pyusb 进行底层 USB 设备操作
"""
# 查找 USB 设备（需要知道 VID 和 PID）
dev = usb.core.find(idVendor=0x1234, idProduct=0x5678)
if dev is None:
print("未找到指定 USB 设备")
return
# 设置配置
dev.set_configuration()
# 获取接口
cfg = dev.get_active_configuration()
intf = cfg[(0,0)]
# 查找端点
ep_out = usb.util.find_descriptor(
intf,
custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT
)
ep_in = usb.util.find_descriptor(
intf,
custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN
)
# 发送数据
ep_out.write(b'Hello USB!')
# 接收数据
data = ep_in.read(64)
print(f"收到: {bytes(data)}")
if __name__ == '__main__':
usb_serial_example()
6. 常见问题与调试
6.1 权限问题（Linux）
# 临时解决：将用户加入 dialout 组
sudo usermod -a -G dialout $USER
# 临时赋予权限
sudo chmod 666 /dev/ttyUSB0
6.2 常见错误排查
| 错误信息 | 原因 | 解决方案 |
|---|---|---|
Permission denied |
权限不足 | 加入 dialout 组或使用 sudo |
Device or resource busy |
串口被占用 | 关闭其他使用该串口的程序 |
Could not open port |
端口不存在 | 检查端口名是否正确 |
Read timeout |
读取超时 | 检查设备是否响应，调整 timeout |
6.3 调试技巧
import serial
import logging
# 启用详细日志
logging.basicConfig(level=logging.DEBUG)
ser = serial.Serial('/dev/ttyUSB0', 9600, timeout=1)
# 查看串口状态
print(f"是否打开: {ser.is_open}")
print(f"输入缓冲区: {ser.in_waiting} 字节")
print(f"输出缓冲区: {ser.out_waiting} 字节")
# 清空缓冲区
ser.reset_input_buffer()
ser.reset_output_buffer()
7. 综合示例
7.1 完整串口通信类
import serial
import serial.tools.list_ports
import threading
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class SerialCommunicator:
"""
通用串口通信类
支持：自动连接、数据监听、回调处理、线程安全
"""
def __init__(self, port=None, baudrate=9600, timeout=1):
self.port = port
self.baudrate = baudrate
self.timeout = timeout
self.serial = None
self._running = False
self._thread = None
self._callbacks = []
self._lock = threading.Lock()
def auto_connect(self, keyword='USB'):
"""自动查找并连接串口"""
ports = serial.tools.list_ports.comports()
for p in ports:
if keyword.lower() in p.description.lower():
self.port = p.device
return self.connect()
logger.error(f"未找到包含 '{keyword}' 的串口")
return False
def connect(self):
"""连接串口"""
try:
self.serial = serial.Serial(
port=self.port,
baudrate=self.baudrate,
timeout=self.timeout
)
logger.info(f"已连接串口: {self.port} @ {self.baudrate}bps")
return True
except serial.SerialException as e:
logger.error(f"连接失败: {e}")
return False
def disconnect(self):
"""断开连接"""
self._running = False
if self._thread and self._thread.is_alive():
self._thread.join(timeout=2)
if self.serial and self.serial.is_open:
self.serial.close()
logger.info("串口已断开")
def send(self, data):
"""发送数据"""
if not self.serial or not self.serial.is_open:
logger.warning("串口未连接")
return 0
with self._lock:
if isinstance(data, str):
data = data.encode('utf-8')
bytes_written = self.serial.write(data)
logger.debug(f"发送 {bytes_written} 字节")
return bytes_written
def send_line(self, text):
"""发送一行文本（自动添加换行）"""
if not text.endswith('\n'):
text += '\n'
return self.send(text)
def register_callback(self, callback):
"""注册数据接收回调函数"""
self._callbacks.append(callback)
def start_listening(self):
"""启动后台监听线程"""
self._running = True
self._thread = threading.Thread(target=self._listen_loop, daemon=True)
self._thread.start()
logger.info("监听线程已启动")
def _listen_loop(self):
"""后台监听循环"""
buffer = b''
while self._running:
try:
if self.serial and self.serial.in_waiting > 0:
data = self.serial.read(self.serial.in_waiting)
buffer += data
# 按行分割处理
while b'\n' in buffer:
line, buffer = buffer.split(b'\n', 1)
line = line.decode('utf-8', errors='ignore').strip()
if line:
self._handle_data(line)
time.sleep(0.01)
except Exception as e:
logger.error(f"监听错误: {e}")
break
def _handle_data(self, data):
"""处理接收到的数据"""
logger.info(f"收到: {data}")
for callback in self._callbacks:
try:
callback(data)
except Exception as e:
logger.error(f"回调错误: {e}")
def __enter__(self):
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.disconnect()
# ============ 使用示例 ============
def on_data_received(data):
"""数据接收回调"""
print(f"[回调] 收到数据: {data}")
if __name__ == '__main__':
# 方式一：使用上下文管理器
with SerialCommunicator(port='/dev/ttyUSB0', baudrate=9600) as comm:
if comm.serial and comm.serial.is_open:
comm.register_callback(on_data_received)
comm.start_listening()
# 发送测试数据
comm.send_line("Hello, Serial!")
# 保持运行
try:
while True:
user_input = input("输入要发送的内容 (或 'quit' 退出): ")
if user_input.lower() == 'quit':
break
comm.send_line(user_input)
except KeyboardInterrupt:
pass
# 方式二：自动连接
# comm = SerialCommunicator(baudrate=115200)
# if comm.auto_connect('USB'):
# comm.register_callback(on_data_received)
# comm.start_listening()
# # ... 业务逻辑
# comm.disconnect()
7.2 串口数据记录器
import serial
import csv
from datetime import datetime
import signal
import sys
class SerialDataLogger:
"""
串口数据记录器：将串口数据实时保存到 CSV 文件
"""
def __init__(self, port, baudrate, output_file='serial_data.csv'):
self.port = port
self.baudrate = baudrate
self.output_file = output_file
self.ser = None
self._running = False
def start(self):
self.ser = serial.Serial(self.port, self.baudrate, timeout=1)
self._running = True
with open(self.output_file, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['timestamp', 'data'])
print(f"开始记录数据到: {self.output_file}")
print("按 Ctrl+C 停止...")
while self._running:
try:
if self.ser.in_waiting > 0:
line = self.ser.readline().decode('utf-8', errors='ignore').strip()
if line:
timestamp = datetime.now().isoformat()
writer.writerow([timestamp, line])
f.flush() # 立即写入磁盘
print(f"[{timestamp}] {line}")
except serial.SerialException as e:
print(f"串口错误: {e}")
break
def stop(self):
self._running = False
if self.ser:
self.ser.close()
print("记录已停止")
if __name__ == '__main__':
logger = SerialDataLogger('/dev/ttyUSB0', 9600, 'sensor_data.csv')
def signal_handler(sig, frame):
logger.stop()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
logger.start()
附录
常用波特率
| 波特率 | 适用场景 |
|---|---|
| 9600 | 默认常用，兼容性好 |
| 19200 | 中等速度 |
| 38400 | 较高速率 |
| 57600 | 高速率 |
| 115200 | 高速通信，现代设备常用 |
数据格式
- 数据位：通常为 8 位（EIGHTBITS）
- 校验位：无校验（PARITY_NONE）最常用
- 停止位：1 位（STOPBITS_ONE）最常用
- 流控：无（无硬件/软件流控
更多推荐
7
0- 0
已为社区贡献4条内容
所有评论(0)