树莓派OS下多同IDUSB设备触发Python脚本异常求助

当前配置 #

udev规则（/etc/udev/rules.d/myrule.rules） #

ACTION=="add", SUBSYSTEM=="usb", ATTR{idVendor}=="1234", ATTR{idProduct}=="5678", ENV{SYSTEMD_WANTS}="usb-trigger.service"

注：1234和5678为示例ID。

systemd服务配置（/etc/systemd/system/usb-trigger.service） #

[Unit]
Description=USB Trigger Service

[Service]
Type=oneshot
ExecStart=/path/to/my/python/script/uart.py

[Install]
WantedBy=multi-user.target

Python脚本代码 #

import serial
import serial.tools.list_ports
import time
import csv
import datetime
import sys

def get_current_time():
    current_time = time.time()
    milliseconds = int((current_time - int(current_time)) * 1000)
    formatted_time = time.strftime("%H:%M:%S", time.localtime(current_time))
    formatted_time += f".{milliseconds:03d}"  # Adding milliseconds
    return formatted_time

def write_to_csv(writer, data):
    try:
        writer.writerow(data)
    except Exception as e:
        print(f"Error writing to CSV: {e}")

def find_com_port():
    try:
        com_ports = serial.tools.list_ports.comports()
        for port in com_ports:
            if port.device.startswith('/dev/ttyUSB') or port.device.startswith('/dev/ttyACM'):
                return port.device
    except Exception as e:
        print(f"Error finding COM port: {e}")

def main():
    try:
        while True:
            com_port = find_com_port()
            if com_port is not None:
                print(f"COM port found: {com_port}")
                break  # Exit the loop once a COM port is found
            else:
                print("No COM port found. Retrying...")
                time.sleep(1)      

        if com_port is None:
            print("No COM port found.")
            return

        filename = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + ".csv"
        bottle_id = None
        bottle_id_collected = False  # Flag to track if bottle ID has been collected

        with serial.Serial(port=com_port, baudrate=460800, timeout=2) as ser, open(filename, 'w', newline='') as file:
            print(f"Connected to {com_port} successfully.")
            writer = csv.writer(file)

            while True:
                received_bytes = ser.readline().decode('ascii', errors='replace').strip()
                if received_bytes:
                    if not bottle_id_collected and 'Bottle ID' in received_bytes:
                        bottle_id = received_bytes.split('Bottle ID')[1].strip()
                        writer.writerow(['Bottle ID', bottle_id])
                        writer.writerow(['Input 1', 'Input 2', 'Input 3', 'time stamp'])
                        bottle_id_collected = True
                    else:
                        parts = received_bytes.split(', ')
                        try:
                            numbers = [float(part) for part in parts]
                            data_row = numbers + [get_current_time()]
                            writer.writerow(data_row)
                            print(f"Writing to CSV: {data_row}") 
                            file.flush()  # Force flushing
                        except ValueError as e:
                            pass
                time.sleep(0.01)

    except serial.SerialException as e:
        print(f"Error opening serial port {com_port}: {e}")

    except KeyboardInterrupt:
        print("Program terminated by user.")

    except Exception as e:
        print(f"An error occurred: {e}")

    finally:
        try:
            if 'ser' in locals() and ser.is_open:
                ser.close()
        except Exception as e:
            print(f"Error closing serial port: {e}")

        try:
            if 'file' in locals() and not file.closed:
                file.close()


        except Exception as e:
            print(f"Error closing CSV file: {e}")

        sys.exit()

if __name__ == "__main__":
    main()

问题现象 #

• 脚本触发不稳定：有时执行，有时不执行
• 插入超过2个设备时，后续设备无法触发脚本启动
• 偶尔所有设备都无法正常触发脚本

原因分析 #

1. udev规则触发时机错误：使用SUBSYSTEM=="usb"会在USB设备枚举初期触发，此时对应的ttyUSB/ACM串口端口尚未创建，脚本无法找到可用端口
2. 多设备触发冲突：相同VID/PID的设备触发同一个usb-trigger.service，而oneshot类型服务默认串行执行，后续设备的触发请求会被阻塞
3. 脚本端口抢占：find_com_port()只返回第一个找到的串口，多设备场景下多个脚本实例会抢占同一个端口，导致其他设备无法绑定到正确端口
4. 无设备拔出处理：脚本未监听设备拔出事件，也没有检测串口连接状态，设备拔出后脚本可能仍占用端口或无响应

解决方案 #

1. 修改udev规则，精准触发并传递设备标识 #

将规则改为针对tty子系统，确保串口端口创建后再触发，同时为每个设备生成独立的服务实例：

ACTION=="add", SUBSYSTEM=="tty", ATTRS{idVendor}=="1234", ATTRS{idProduct}=="5678", ENV{SYSTEMD_WANTS}="usb-trigger@%k.service"

• %k会替换为当前设备名（如ttyUSB0），实现每个设备对应独立的服务实例
• ATTRS{}（带S）用于匹配父USB设备的VID/PID，确保tty设备属于目标USB设备

2. 修改systemd服务为模板服务 #

将usb-trigger.service重命名为usb-trigger@.service，内容修改为：

[Unit]
Description=USB Trigger Service for %I

[Service]
Type=simple
ExecStart=/usr/bin/python3 /path/to/my/python/script/uart.py %I
Restart=on-failure
RestartSec=1
User=pi  # 替换为你的用户名，避免权限问题

[Install]
WantedBy=multi-user.target

• Type=simple适配长期运行的脚本（原oneshot是一次性执行，不适合串口监听）
• %I会传递设备名参数给脚本，指定要连接的串口
• Restart配置确保脚本异常退出后自动重启

3. 修改Python脚本，支持指定端口并处理设备断开 #

修改脚本，直接接收命令行传递的端口参数，移除自动查找逻辑，同时添加串口连接状态检测：

import serial
import time
import csv
import datetime
import sys

def get_current_time():
    current_time = time.time()
    milliseconds = int((current_time - int(current_time)) * 1000)
    formatted_time = time.strftime("%H:%M:%S", time.localtime(current_time))
    formatted_time += f".{milliseconds:03d}"
    return formatted_time

def main():
    if len(sys.argv) != 2:
        print("Usage: python3 uart.py <serial_port>")
        sys.exit(1)
    # 拼接完整串口路径，比如传入ttyUSB0则变为/dev/ttyUSB0
    com_port = f"/dev/{sys.argv[1]}"
    # 文件名添加设备标识，避免多设备文件冲突
    filename = f"{datetime.datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}_{sys.argv[1]}.csv"

    bottle_id = None
    bottle_id_collected = False

    try:
        with serial.Serial(port=com_port, baudrate=460800, timeout=2) as ser, open(filename, 'w', newline='') as file:
            print(f"Connected to {com_port} successfully.")
            writer = csv.writer(file)

            # 循环检测串口连接状态
            while ser.is_open:
                try:
                    received_bytes = ser.readline().decode('ascii', errors='replace').strip()
                    if received_bytes:
                        if not bottle_id_collected and 'Bottle ID' in received_bytes:
                            bottle_id = received_bytes.split('Bottle ID')[1].strip()
                            writer.writerow(['Bottle ID', bottle_id])
                            writer.writerow(['Input 1', 'Input 2', 'Input 3', 'time stamp'])
                            bottle_id_collected = True
                        else:
                            parts = received_bytes.split(', ')
                            try:
                                numbers = [float(part) for part in parts]
                                data_row = numbers + [get_current_time()]
                                writer.writerow(data_row)
                                print(f"Writing to CSV: {data_row}")
                                file.flush()
                            except ValueError:
                                pass
                except serial.SerialException:
                    print(f"Lost connection to {com_port}")
                    break
                time.sleep(0.01)

    except serial.SerialException as e:
        print(f"Error opening serial port {com_port}: {e}")
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        print(f"Exiting service for {com_port}")
        sys.exit()

if __name__ == "__main__":
    main()

4. 重新加载配置并测试 #

执行以下命令使配置生效：

sudo udevadm control --reload-rules
sudo systemctl daemon-reload

插入设备后，可通过以下命令查看服务状态和日志：

# 查看某个设备的服务状态
systemctl status usb-trigger@ttyUSB0.service
# 实时查看服务日志
journalctl -u usb-trigger@ttyUSB0.service -f

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