如何用evdev实现Linux设备动态连接与断开的事件处理

我需要在Linux系统中使用evdev异步读取输入事件，同时实现设备连接与断开的动态处理。要求系统具备通用性、支持多设备，通过控制线程检测设备状态变化，调度或取消对应设备的处理任务。我编写了四个实现原型，但每个都存在问题：

1. 原型一：设备变更线程无权限添加任务导致崩溃 #

该方案逻辑直观，但设备变更线程无权限添加新任务，设备连接时程序崩溃。代码如下：

import asyncio
import evdev
import pyudev

# 处理设备变更（连接/断开）的函数
def handle_device_changes(devices, device_tasks):
    current_devices = [evdev.InputDevice(path) for path in evdev.list_devices()]
    added_devices = [device for device in current_devices if device not in devices]
    removed_devices = [device for device in devices if device not in current_devices]
    devices = current_devices
    for device in added_devices:
        print(f"Added device: {device.name}")
        # 为新增设备执行必要操作
        if "GuliKit" in device.name:
            device.grab()
            task = asyncio.create_task(print_events(device))
            device_tasks[device] = task
    for device in removed_devices:
        print(f"Removed device: {device.name}")
        # 为移除设备执行必要操作
        if device in device_tasks:
            task = device_tasks[device]
            task.cancel()
            del device_tasks[device]
    return devices

# 为每个捕获的设备创建的任务
async def print_events(device):
    try:
        async for event in device.async_read_loop():
            print(device.path, evdev.categorize(event), sep=': ')
    except OSError:
        print(f"Device {device.name} disconnected.")
        #handle_device_changes()

# 程序入口
async def main():
    devices = []
    device_tasks = {}
    devices = handle_device_changes(devices, device_tasks)
    queue = asyncio.Queue()

    monitor = pyudev.Monitor.from_netlink(pyudev.Context())
    monitor.filter_by(subsystem='input')
    def log_event(action, device):
        queue.put(handle_device_changes(devices, device_tasks))
    observer = pyudev.MonitorObserver(monitor, log_event)
    observer.start()

    try:
        while True:
            await queue.get()
    except KeyboardInterrupt:
        print("Keyboard interrupt received.")
    finally:
        observer.stop()
        await asyncio.gather(*device_tasks.values(), return_exceptions=True)

if __name__ == '__main__':
    asyncio.run(main(),debug=True)

2. 原型二：存在无法排查的运行错误 #

重新编写的版本，出现无法定位的运行错误。代码如下：

import evdev
import asyncio
import pyudev

# 处理设备事件的函数
async def handle_event(device):
    async for event in device.async_read_loop():
        # 在此处处理事件
        print("Event:", event)

# 处理设备变更（连接/断开）的函数
def handle_device_changes(devices, device_tasks):
    current_devices = [evdev.InputDevice(path) for path in evdev.list_devices()]
    added_devices = [device for device in current_devices if device not in devices]
    removed_devices = [device for device in devices if device not in current_devices]
    devices = current_devices

    for device in added_devices:
        print(f"Added device: {device.name}")
        # 为新增设备创建事件监听任务
        event_loop = asyncio.get_event_loop()
        event_task = event_loop.create_task(handle_event(device))
        device_tasks[device.path] = event_task
        if "GuliKit" in device.name:
            device.grab()

    for device in removed_devices:
        print(f"Removed device: {device.name}")
        # 取消并清理与移除设备关联的事件监听任务
        event_task = device_tasks.pop(device.path)
        event_task.cancel()

    return devices

async def main():
    devices = []
    device_tasks = {}

    # 初始设置：处理设备变更以建立事件监听
    devices = handle_device_changes(devices, device_tasks)

    # 监听设备变更
    context = pyudev.Context()
    monitor = pyudev.Monitor.from_netlink(context)
    monitor.filter_by(subsystem='input')

    loop = asyncio.get_event_loop()
    while True:
        try:
            async for device_event in loop.run_in_executor(None, monitor.poll):
                # 处理设备变更
                devices = handle_device_changes(devices, device_tasks)
        except KeyboardInterrupt:
            break

if __name__ == '__main__':
    asyncio.run(main())

3. 原型三：设备列表维护正常但无法上报事件 #

借助ChatGPT生成的实现，能准确维护设备列表，但无法上报设备事件。代码如下：

import evdev
import selectors
import pyudev

# 处理设备事件的函数
def handle_event(device, selector):
    for event in device.read():
        # 在此处处理事件
        print("event")

# 处理设备变更（连接/断开）的函数
def handle_device_changes(devices, device_tasks, selector):
    current_devices = [evdev.InputDevice(path) for path in evdev.list_devices()]
    added_devices = [device for device in current_devices if device not in devices]
    removed_devices = [device for device in devices if device not in current_devices]
    devices = current_devices

    for device in added_devices:
        print(f"Added device: {device.name}")
        # 为新增设备创建事件监听任务
        event_listener = selectors.EVENT_READ | selectors.EVENT_WRITE
        event_task = selector.register(device.fd, event_listener, handle_event)
        device_tasks[device.path] = event_task

    for device in removed_devices:
        print(f"Removed device: {device.name}")
        # 销毁与移除设备关联的事件监听任务
        event_task = device_tasks.pop(device.path)
        selector.unregister(device.fd)

    return devices

def main():
    devices = []
    device_tasks = {}
    selector = selectors.DefaultSelector()

    # 初始设置：处理设备变更以建立事件监听
    devices = handle_device_changes(devices, device_tasks, selector)

    # 监听设备变更
    context = pyudev.Context()
    monitor = pyudev.Monitor.from_netlink(context)
    monitor.filter_by(subsystem='input')

    for device_event in iter(monitor.poll, None):
        # 处理设备变更
        devices = handle_device_changes(devices, device_tasks, selector)

if __name__ == '__main__':
    main()

4. 原型四：使用废弃功能无法正常运行 #

参考旧帖实现，但使用了已废弃的功能，无法正常运行。代码如下：

import functools
import pyudev

from evdev import InputDevice
from select import select

context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by(subsystem='input')
monitor.start()

fds = {monitor.fileno(): monitor}
finalizers = []

while True:
    r, w, x = select(fds, [], [])

    if monitor.fileno() in r:
        r.remove(monitor.fileno())

        for udev in iter(functools.partial(monitor.poll, 0), None):
            # 我们只关注有设备节点的设备（如/dev/input/eventX）
            if not udev.device_node:
                break

            # 找到目标设备并添加到fds
            for name in (i['NAME'] for i in udev.ancestors if 'NAME' in i):
                # 我使用虚拟输入设备测试——请根据需求调整
                if u'GuliKit' in name:
                    if udev.action == u'add':
                        print('Device added: %s' % udev)
                        print(udev.device_node)
                        fds[dev.fd] = InputDevice(udev.device_node)
                        break
                    if udev.action == u'remove':
                        print('Device removed: %s' % udev)
                        def helper():
                            global fds
                            fds = {monitor.fileno(): monitor}
                        finalizers.append(helper)
                        break

    for fd in r:
        dev = fds[fd]
        for event in dev.read():
            print(event)

    for i in range(len(finalizers)):
        finalizers.pop()()

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