适用于 Linux 和 Windows 的多线程键盘检测器

Posted 2023-03-30

【中文标题】适用于 Linux 和 Windows 的多线程键盘检测器

【英文标题】：Multithreaded Keyboard Detector for Linux and Windows

【发布时间】：2020-07-17 14:04:48

【问题描述】：

我想在我现有的Keyboard Detector for Windows 中添加一个适用于 Linux 的键盘检测。所以我用pyudev创建了一个LinuxKeyboardDetector。

脚本可以启动，出现图形用户界面，可惜键盘检测没有识别任何东西，也没有报错。

我怀疑使用QRunnable 的多线程存在问题。

代码

import sys
from datetime import datetime
import platform

from PyQt5. QtCore import QObject, QRunnable, QThreadPool, pyqtSignal
from PyQt5.QtWidgets import QApplication, QMainWindow, QTableWidget, QTableWidgetItem, QHeaderView


current_platform = platform.system()
if current_platform == "Windows":
    import pythoncom
    import wmi
elif current_platform == "Linux":
    import pyudev
    from pyudev.pyqt5 import MonitorObserver


def create_keyboard_detector():
    keyboard_detector = None
    if current_platform == "Windows":
        keyboard_detector = WindowsKeyboardDetector()
    elif current_platform == "Linux":
        keyboard_detector = LinuxKeyboardDetector()
    return keyboard_detector


class KeyboardDetectorSignals(QObject):
    keyboard_changed = pyqtSignal(str)


class WindowsKeyboardDetector(QRunnable):
    def __init__(self):
        super().__init__()

        self.signals = KeyboardDetectorSignals()

    def run(self):
        pythoncom.CoInitialize()
        device_connected_wql = "SELECT * FROM __InstanceCreationEvent WITHIN 2 WHERE TargetInstance ISA \'Win32_Keyboard\'"
        device_disconnected_wql = "SELECT * FROM __InstanceDeletionEvent WITHIN 2 WHERE TargetInstance ISA \'Win32_Keyboard\'"

        c = wmi.WMI()
        connected_watcher = c.watch_for(raw_wql=device_connected_wql)
        disconnected_watcher = c.watch_for(raw_wql=device_disconnected_wql)

        while True:
            try:
                connected = connected_watcher(timeout_ms=10)
            except wmi.x_wmi_timed_out:
                pass
            else:
                if connected:
                    self.signals.keyboard_changed.emit("Keyboard connected.")

            try:
                disconnected = disconnected_watcher(timeout_ms=10)
            except wmi.x_wmi_timed_out:
                pass
            else:
                if disconnected:
                    self.signals.keyboard_changed.emit("Keyboard disconnected.")


class LinuxKeyboardDetector(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = KeyboardDetectorSignals()
        self.context = pyudev.Context()
        self.monitor = pyudev.Monitor.from_netlink(self.context)
        self.observer = MonitorObserver(self.monitor)

    def run(self):
        self.monitor.filter_by(subsystem="usb", device_type="usb_device")
        self.observer.deviceEvent.connect(self.process_device_event)
        self.monitor.start()

    def process_device_event(self, device):
        if device['ID_INPUT_KEYBOARD'] == '1':
            if device.action == "add":
                self.signals.keyboard_changed.emit("Keyboard connected.")
            if device.action == "remove":
                self.signals.keyboard_changed.emit("Keyboard disconnected.")


class MainWindow(QMainWindow):

    def __init__(self):
        super().__init__()

        self.setGeometry(100, 100, 500, 500)
        self.setWindowTitle("Keyboard Logger")

        self.log_table = QTableWidget()
        self.log_table.setColumnCount(2)
        self.log_table.setShowGrid(True)
        self.log_table.setHorizontalHeaderLabels(["Time", "Event"])
        self.log_table.horizontalHeader().setStretchLastSection(True)
        self.log_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeToContents)
        self.setCentralWidget(self.log_table)
        self.show()

        self.threadpool = QThreadPool()
        keyboard_detector = create_keyboard_detector()
        keyboard_detector.signals.keyboard_changed.connect(self.add_row)
        self.threadpool.start(keyboard_detector)

    def add_row(self, event: str):
        now = datetime.now()
        datetime_string = now.strftime("%Y-%m-%d %H:%M:%S")

        row_count = self.log_table.rowCount()
        self.log_table.insertRow(row_count)
        self.log_table.setItem(row_count, 0, QTableWidgetItem(datetime_string))
        self.log_table.setItem(row_count, 1, QTableWidgetItem(event))


def main():
    app = QApplication(sys.argv)
    window = MainWindow()
    sys.exit(app.exec_())


if __name__ == '__main__':
    main()

---

编辑 1： 更新 LinuxKeyboardDetector 类以使用基本的 pyudev.MonitorObserver，而不是专用的 pyqt 版本。

class LinuxKeyboardDetector(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = KeyboardDetectorSignals()
        self.context = pyudev.Context()
        self.monitor = pyudev.Monitor.from_netlink(self.context)
        # self.observer = MonitorObserver(self.monitor)
        self.observer = pyudev.MonitorObserver(self.monitor, self.process_device_event)

    def run(self):
        self.monitor.filter_by(subsystem="usb", device_type="usb_device")
        # self.observer.deviceEvent.connect(self.process_device_event)
        # self.monitor.start()
        self.observer.start()

    def process_device_event(self, device):
        if device['ID_INPUT_KEYBOARD'] == '1':
            if device.action == "add":
                self.signals.keyboard_changed.emit("Keyboard connected.")
            if device.action == "remove":
                self.signals.keyboard_changed.emit("Keyboard disconnected.")

结果1：插入或关闭USB键盘时出现以下错误信息。

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/home/ata/source/venv/lib/python3.6/site-packages/pyudev/monitor.py", line 532, in run
    self._callback(device)
  File "/home/ata/source/venv/lib/python3.6/site-packages/pyudev/monitor.py", line 508, in <lambda>
    callback = lambda d: event_handler(d.action, d)
TypeError: process_device_event() takes 2 positional arguments but 3 were given

【问题讨论】：

@eyllanesc 我假设您的意思是 keyboard_detector 在MainWindow 类的__init__() 方法中。如果是这样，我必须通知您，更改为 self.keyboard_detector 很遗憾并没有解决问题。

为什么不使用像pynput这样的包。它是跨平台的异步和同步键盘和鼠标监视器/控制器

@PraysonW.Daniel 非常感谢您的提示。我还不知道pynput 库。

【参考方案1】：

根据this answer，您必须在Qt 应用程序事件循环之前 启动监视器。在这种情况下，您必须不使用 QRunnable，因为监视器将作为标准 QObject 工作，异步工作并在需要时发送信号。

如果你还想保持相同的界面，使用QRunnable，我认为唯一的解决方案是使用基本的pyudev.MonitorObserver，而不是专用的pyqt版本。

class LinuxKeyboardDetector(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = KeyboardDetectorSignals()
        self.context = pyudev.Context()
        self.monitor = pyudev.Monitor.from_netlink(self.context)
        self.observer = pyudev.MonitorObserver(self.monitor, self.process_device_event)

    def run(self):
        self.monitor.filter_by(subsystem="usb", device_type="usb_device")
        self.observer.start()

【讨论】：

保持界面不变就好了。不幸的是，您建议的解决方案会出现一条错误消息，请参阅结果 1。

这可能是由于参数签名，如果我没记错的话，pyudev 的 MonitorObserver 使用两个参数，第一个是动作类型，第二个是设备。您可以通过更改为process_device_event(self, *args):保持相同的界面，然后使用args[-1]获取设备。

除了检查USB设备是否是键盘（KeyError: 'ID_INPUT_KEYBOARD'），现在可以了，谢谢！

@Atalanttore 我没有包含该部分，因为我认为您已经弄清楚了，并且您只提供了最少的代码。由于您实际上是在获取字典，因此请确保仅进一步了解 if 'ID_INPUT_KEYBOARD' in device。

感谢您的提示，但由于libudev 没有提供适当的功能，我已经在检查USB 设备是否为键盘时失败了一次。见github.com/pyudev/pyudev/issues/361