寻求基于Python实现两台Raspberry Pi串口传输各类文件的开发起点
串口跨树莓派传输文件开发起点建议
一、先搞定硬件连接与串口基础配置
- 硬件接线:用树莓派GPIO的硬件UART引脚(TXD/GPIO14、RXD/GPIO15),交叉连接两台设备——A的TX接B的RX,A的RX接B的TX,同时将两台设备的GND引脚相连(必须共地,否则通信不稳定)。
- 禁用串口控制台:树莓派默认把硬件UART用作系统控制台,需要关闭。执行
sudo raspi-config,依次选择Interfacing Options > Serial,关闭"Would you like a login shell to be accessible over serial?",开启"Would you like the serial port hardware to be enabled?",重启设备生效。 - 确认串口设备:重启后用
ls /dev/tty*查看,硬件UART对应的设备通常是/dev/ttyS0。
二、搭建Python串口开发环境
- 安装
pyserial库:这是Python串口开发的标准依赖,执行pip install pyserial完成安装。 - 先做基础通信测试:写极简代码验证双向通信是否正常,避免后续文件传输时排查硬件问题。
- 发送端测试代码:
import serial import time ser = serial.Serial('/dev/ttyS0', baudrate=9600, timeout=1) time.sleep(2) # 给串口初始化预留时间 ser.write(b"Test from RPi 1") ser.close() - 接收端测试代码:
import serial ser = serial.Serial('/dev/ttyS0', baudrate=9600, timeout=5) data = ser.readline() if data: print(f"Received: {data.decode('utf-8')}") ser.close()
- 发送端测试代码:
三、文件传输核心逻辑设计
串口是字节流传输,必须约定简单的传输规则保证文件完整性:
- 先传文件元数据:发送端先发送文件名长度、文件名、文件总字节数,让接收端明确要接收的数据量和保存路径。
- 分块传输大文件:将文件拆成固定大小的块(比如1024字节/块)逐块发送,接收端逐块写入本地文件,避免一次性加载大文件占内存。
- 可选校验机制:给每个数据块加CRC校验,接收端验证后再写入,防止传输错误。
基础功能示例代码
发送端(支持任意格式文件)
import serial import os import time def send_file(serial_port, file_path, baudrate=9600): ser = serial.Serial(serial_port, baudrate, timeout=2) time.sleep(2) # 读取文件元数据 file_name = os.path.basename(file_path) file_size = os.path.getsize(file_path) # 发送元数据:文件名长度(2字节)+文件名+文件大小(4字节) ser.write(len(file_name).to_bytes(2, byteorder='big')) ser.write(file_name.encode('utf-8')) ser.write(file_size.to_bytes(4, byteorder='big')) # 分块发送文件内容 with open(file_path, 'rb') as f: while chunk := f.read(1024): ser.write(chunk) # 可选:等待接收端确认,避免丢包 # ack = ser.read(1) # if ack != b'OK': # f.seek(-len(chunk), os.SEEK_CUR) ser.close() print("File sent successfully") if __name__ == "__main__": send_file('/dev/ttyS0', 'sample.pdf')
接收端
import serial import os def receive_file(serial_port, save_dir='./received_files', baudrate=9600): os.makedirs(save_dir, exist_ok=True) ser = serial.Serial(serial_port, baudrate, timeout=5) # 读取文件名长度和文件名 name_len = int.from_bytes(ser.read(2), byteorder='big') file_name = ser.read(name_len).decode('utf-8') # 读取文件总大小 file_size = int.from_bytes(ser.read(4), byteorder='big') save_path = os.path.join(save_dir, file_name) received_bytes = 0 with open(save_path, 'wb') as f: while received_bytes < file_size: # 计算剩余需要接收的字节数,避免最后一块超出大小 chunk_size = min(1024, file_size - received_bytes) chunk = ser.read(chunk_size) if not chunk: print("Connection timeout, transfer incomplete") break f.write(chunk) received_bytes += len(chunk) # 可选:发送确认信号 # ser.write(b'OK') ser.close() if received_bytes == file_size: print(f"File saved to {save_path}") else: print(f"Transfer failed: received {received_bytes}/{file_size} bytes") if __name__ == "__main__": receive_file('/dev/ttyS0')
四、优化与扩展方向
- 提升传输速度:根据硬件支持,将波特率调高到115200甚至更高(两台设备必须设置一致)。
- 异常处理:给代码加入
try-except块,捕获串口断开、文件读写失败等异常,提高稳定性。 - 批量传输:扩展代码支持多文件批量发送,比如先发送文件总数,再逐个传输每个文件的元数据和内容。
- 加密传输:如果需要安全性,可在发送前对文件内容加密,接收端解密后再保存。
内容的提问来源于stack exchange,提问作者Michael
相关产品推荐
相关产品推荐

