Python下无需插拔USB即可维持RFID串口通信的解决方法

开发Python应用使用PySerial通过串口连接UHF RFID阅读器，持续读取数据。但遇到以下问题：用Ctrl+C停止程序或发生错误后，重启程序无法从串口读取数据，必须物理插拔RFID阅读器的USB线缆才能恢复。指示灯显示命令已发送至阅读器，但无数据返回。

环境信息 #

Language: Python 3.12
Library: PySerial
Operating System: macOS
RFID Reader Model: [你的RFID阅读器型号]
Connection Details: USB转串口接口

当前行为 #

程序正常关闭或异常终止后，重启无法读取串口数据，唯一解决方法是物理插拔USB线缆。

预期行为 #

无需物理操作硬件，重启程序后串口连接自动恢复并持续读取数据。

已尝试方案 #

• 使用ser.reset_input_buffer()和ser.reset_output_buffer()清除缓冲区
• 在异常处理块中实现重连策略，关闭串口并重试打开
• 添加详细日志检查连接状态，显示除非插拔线缆，否则重启后无数据读取

原始代码示例 #

import serial
import threading
import time
import json
from datetime import datetime
from flask_cors import CORS
from queue import Queue
import logging
from flask import Flask, render_template, jsonify  # 补充原代码遗漏的导入

logging.basicConfig(level=logging.INFO)

app = Flask(__name__)
CORS(app, resources={r"/*": {"origins": "http://localhost:5000"}})
assigned_numbers_queue = Queue(maxsize=4)

SERIAL_PORT = '/dev/tty.usbmodem3417240901531'
BAUD_RATE = 921600
RESET_COMMAND = '$ba-bnx0-re\n'
INVENTORY_COMMAND = '$ir-bnx0-sex0-tax0-slx0-dtx1-mcx2-iqx4-fqx0-hqxF-lqx0-svx0-rfxF1-stx0-srx0-sdx1388-trx3001\n'
GO_COMMAND = '$ba-bnx0-go\n'

try:
    with open('epc_mappings.json') as f:
        epc_to_number = json.load(f)
except FileNotFoundError:
    print("EPC mappings file not found.")

def process_tag(raw_epc):
    prefix_to_remove = "TR: | EP: "
    epc = raw_epc[len(prefix_to_remove):].strip().upper()
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"Detected Tag ID: {epc} - Timestamp: {timestamp}")

def read_from_serial(ser):
    while True:
        line = ser.readline().decode().strip()
        if line:
            print(f"Received: {line}")
            process_tag(line)

def send_go_command(ser):
    while True:
        ser.write(GO_COMMAND.encode())
        time.sleep(7)

def read_from_rfid():
    with serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=1) as ser:
        ser.write(RESET_COMMAND.encode())
        ser.write(INVENTORY_COMMAND.encode())
        threading.Thread(target=send_go_command, args=(ser,), daemon=True).start()
        while True:
            line = ser.readline().decode().strip()
            if line:
                print(f"Received: {line}")
                process_tag(line)

threading.Thread(target=read_from_rfid, daemon=True).start()

@app.route('/')
def index():
    return render_template('index.html', assigned_numbers=list(assigned_numbers_queue.queue))

@app.route('/tags')
def get_tags():
    return jsonify({'assigned_numbers': list(assigned_numbers_queue.queue)})

if __name__ == '__main__':
    app.run(debug=True)

1. macOS USB串口资源残留：macOS的USB转串口驱动（如CDC-ACM）在程序异常终止时，可能未正确释放端口资源，导致系统认为端口仍被占用，或阅读器处于半连接状态。
2. RFID阅读器状态锁死：程序异常退出时，阅读器可能未收到终止指令，仍停留在之前的工作状态，后续发送的初始化指令无法触发响应。
3. PySerial连接未彻底关闭：虽然使用了with上下文管理器，但在KeyboardInterrupt等异常下，可能存在串口关闭不彻底的情况，导致阅读器的串口逻辑未重置。

1. 完善异常处理与串口清理 #

捕获所有可能的异常（包括KeyboardInterrupt），确保串口被正确关闭，并清空缓冲区。

2. 利用DTR/RTS信号重置阅读器 #

在连接前切换DTR/RTS信号状态，模拟物理插拔的重置效果，触发阅读器内部状态重置。

3. 实现带重试的重连机制 #

每次连接失败后，等待一段时间重试，确保系统端口资源释放。

4. 完整的初始化流程 #

每次重新连接时，发送完整的重置、配置指令，确保阅读器回到可工作状态。

import serial
import threading
import time
import json
from datetime import datetime
from flask_cors import CORS
from queue import Queue
import logging
from flask import Flask, render_template, jsonify
import signal

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)
CORS(app, resources={r"/*": {"origins": "http://localhost:5000"}})
assigned_numbers_queue = Queue(maxsize=4)

SERIAL_PORT = '/dev/tty.usbmodem3417240901531'
BAUD_RATE = 921600
RESET_COMMAND = '$ba-bnx0-re\n'
INVENTORY_COMMAND = '$ir-bnx0-sex0-tax0-slx0-dtx1-mcx2-iqx4-fqx0-hqxF-lqx0-svx0-rfxF1-stx0-srx0-sdx1388-trx3001\n'
GO_COMMAND = '$ba-bnx0-go\n'

try:
    with open('epc_mappings.json') as f:
        epc_to_number = json.load(f)
except FileNotFoundError:
    logger.warning("EPC mappings file not found.")

# 全局串口与线程对象，用于优雅关闭
ser = None
go_thread = None

def process_tag(raw_epc):
    prefix_to_remove = "TR: | EP: "
    epc = raw_epc[len(prefix_to_remove):].strip().upper()
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    logger.info(f"Detected Tag ID: {epc} - Timestamp: {timestamp}")

def send_go_command():
    global ser
    while ser and ser.is_open:
        try:
            ser.write(GO_COMMAND.encode())
            time.sleep(7)
        except Exception as e:
            logger.error(f"Failed to send GO command: {str(e)}")
            break

def read_from_rfid():
    global ser, go_thread
    while True:
        try:
            # 模拟物理插拔：切换DTR/RTS信号重置阅读器
            with serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=1) as temp_ser:
                temp_ser.dtr = False
                temp_ser.rts = False
                time.sleep(0.5)
                temp_ser.dtr = True
                temp_ser.rts = True
                time.sleep(0.5)
            
            # 正式建立串口连接
            ser = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=1)
            logger.info(f"Connected to serial port {SERIAL_PORT}")

            # 清空输入输出缓冲区
            ser.reset_input_buffer()
            ser.reset_output_buffer()

            # 发送初始化指令序列
            ser.write(RESET_COMMAND.encode())
            time.sleep(0.2)
            ser.write(INVENTORY_COMMAND.encode())
            time.sleep(0.2)

            # 启动GO指令发送线程
            go_thread = threading.Thread(target=send_go_command, daemon=True)
            go_thread.start()

            # 持续读取串口数据
            while ser.is_open:
                line = ser.readline().decode().strip()
                if line:
                    logger.info(f"Received: {line}")
                    process_tag(line)
        
        except serial.SerialException as e:
            logger.error(f"Serial connection error: {str(e)}. Retrying in 3 seconds...")
            if ser and ser.is_open:
                ser.close()
            time.sleep(3)
        except KeyboardInterrupt:
            logger.info("Received interrupt, closing serial connection...")
            if ser and ser.is_open:
                ser.close()
            break
        except Exception as e:
            logger.error(f"Unexpected error: {str(e)}. Retrying in 3 seconds...")
            if ser and ser.is_open:
                ser.close()
            time.sleep(3)

def handle_shutdown(signum, frame):
    """处理系统退出信号，优雅关闭串口"""
    global ser, go_thread
    logger.info("Shutting down gracefully...")
    if ser and ser.is_open:
        ser.close()
    exit(0)

# 注册信号处理，捕获Ctrl+C和系统终止信号
signal.signal(signal.SIGINT, handle_shutdown)
signal.signal(signal.SIGTERM, handle_shutdown)

# 启动RFID读取线程
threading.Thread(target=read_from_rfid, daemon=True).start()

@app.route('/')
def index():
    return render_template('index.html', assigned_numbers=list(assigned_numbers_queue.queue))

@app.route('/tags')
def get_tags():
    return jsonify({'assigned_numbers': list(assigned_numbers_queue.queue)})

if __name__ == '__main__':
    app.run(debug=True)

内容的提问来源于stack exchange，提问作者William Hawkins