Notify Qml For 'usb Device Inserted' Events Using Pyqt5 And Pyudev

I have a GUI application (made with PyQt5 and QML) and would like to get notify when a usb device is plugged or unplugged from the computer. After some investigation, I have found

Solution 1:

The best thing to do is to modify the GUI in QML, for which the Monitor and Device object must be accessible from QML. Only QObjects receive notifications so I will create 2 classes that wrap with a light layer to both classes using q-properties and slots.

pyqtudev.py

from PyQt5 import QtCore
from pyudev import Context, Monitor, Device
from pyudev.pyqt5 import MonitorObserver

classQtUdevDevice(QtCore.QObject):
    def__init__(self, parent=None):
        super(QtUdevDevice, self).__init__(parent)
        self.m_dev = Nonedefinitialize(self, dev):
        self.m_dev = dev

    @QtCore.pyqtSlot(result=bool)defisValid(self):
        return self.m_dev isnotNone    @QtCore.pyqtProperty(str, constant=True)defdevType(self):
        ifnot self.isValid():
            return""if self.m_dev.device_type isNone:
            return""return self.m_dev.device_type

    @QtCore.pyqtProperty(str, constant=True)defsubsystem(self):
        ifnot self.isValid():
            return""return self.m_dev.subsystem

    @QtCore.pyqtProperty(str, constant=True)defname(self):
        ifnot self.isValid():
            return""return self.m_dev.sys_name

    @QtCore.pyqtProperty(str, constant=True)defdriver(self):
        ifnot self.isValid():
            return""if self.m_dev.driver isNone:
            return""return self.m_dev.driver

    @QtCore.pyqtProperty(str, constant=True)defdeviceNode(self):
        ifnot self.isValid():
            return""if self.m_dev.device_node isNone:
            return""return self.m_dev.device_node

    @QtCore.pyqtProperty(list, constant=True)defalternateDeviceSymlinks(self):
        returnlist(self.m_dev.device_links)

    @QtCore.pyqtProperty(str, constant=True)defsysfsPath(self):
        ifnot self.isValid():
            return""return self.m_dev.sys_path

    @QtCore.pyqtProperty(int, constant=True)defsysfsNumber(self):
        ifnot self.isValid():
            return -1if self.m_dev.sys_number isNone:
            return -1returnint(self.m_dev.sys_number)

    @QtCore.pyqtSlot(str, result=str)defproperty(self, name):
        ifnot self.isValid():
            return""
        v = self.m_dev.properties.get(name)
        return v if v isnotNoneelse""    @QtCore.pyqtSlot(str, result=bool)defhasProperty(self, name):
        ifnot self.isValid():
            returnFalsereturn self.m_dev.properties.get(name) isnotNone    @QtCore.pyqtProperty(list, constant=True)defdeviceProperties(self):
        ifnot self.isValid():
            return []
        returnlist(self.m_dev.properties)

    @QtCore.pyqtProperty(list, constant=True)defsysfsProperties(self):
        ifnot self.isValid():
            return []
        returnlist(self.m_dev.attributes.available_attributes)

    @QtCore.pyqtProperty(QtCore.QObject, constant=True)defparentDevice(self):
        ifnot self.isValid:
            returnif self.m_dev.parent:
            parent_device = QtUdevDevice()
            parent_device.initialize(self.m_dev.parent)
            return parent_device

    @QtCore.pyqtProperty(str, constant=True)defaction(self):
        ifnot self.isValid():
            return""if self.m_dev.action isNone:
            return""return self.m_dev.action

    def__repr__(self):
        if self.isValid():
            return"UdevDevice({})".format(self.sysfsPath())
        return"Invalid UdevDevice"classQtMonitorObserver(QtCore.QObject):
    deviceEvent = QtCore.pyqtSignal(QtUdevDevice, arguments=["device"])
    deviceAdded = QtCore.pyqtSignal(QtUdevDevice, arguments=["device"])
    deviceRemoved = QtCore.pyqtSignal(QtUdevDevice, arguments=["device"])
    deviceChanged = QtCore.pyqtSignal(QtUdevDevice, arguments=["device"])
    deviceOnlined = QtCore.pyqtSignal(QtUdevDevice, arguments=["device"])
    deviceOfflined = QtCore.pyqtSignal(QtUdevDevice, arguments=["device"])

    def__init__(self, parent=None):
        super(QtMonitorObserver, self).__init__(parent)
        context = Context()
        self._monitor = Monitor.from_netlink(context)
        self._observer = MonitorObserver(self._monitor, self)
        self._observer.deviceEvent.connect(self.setup_new_signals)

    @QtCore.pyqtSlot()defstart(self):
        self._monitor.start()

    @QtCore.pyqtSlot(str)deffilter_by(self, filter):
        self._monitor.filter_by(subsystem=filter)

    @QtCore.pyqtSlot(str)deffilter_by_tag(self, tag):
        self._monitor.filter_by_tag(tag)

    @QtCore.pyqtSlot()defremove_filter(self):
        self._monitor.remove_filter()

    @QtCore.pyqtSlot(Device)defsetup_new_signals(self, device):
        new_signals_map = {
            'add': self.deviceAdded,
            'remove': self.deviceRemoved,
            'change': self.deviceChanged,
            'online': self.deviceOnlined,
            'offline': self.deviceOfflined,
        }
        signal = new_signals_map.get(device.action)
        qtdevice = QtUdevDevice()
        qtdevice.initialize(device)
        if signal:
            signal.emit(qtdevice)
        self.deviceEvent.emit(qtdevice)

main.py

import os
import sys
from PyQt5 import QtCore, QtGui, QtQml

from pyqtudev import QtMonitorObserver

def run():
    app = QtGui.QGuiApplication(sys.argv)
    engine = QtQml.QQmlApplicationEngine()
    observer = QtMonitorObserver()
    engine.rootContext().setContextProperty("observer", observer)
    directory = os.path.dirname(os.path.abspath(__file__))
    engine.load(QtCore.QUrl.fromLocalFile(os.path.join(directory, 'main.qml')))
    if not engine.rootObjects():
        return -1
    return app.exec_()

if __name__ == "__main__":
    sys.exit(run())

main.qml

import QtQuick 2.11
import QtQuick.Window2.2
import QtQuick.Controls2.2

ApplicationWindow {    
    visible: true
    width: Screen.width/2
    height: Screen.height/2
    Connections {
        target: observer
        onDeviceEvent: {
            console.log(device, device.name, device.action, device.parentDevice)
            if(device.hasProperty("ID_VENDOR_ID")){
                console.log(device.property("ID_VENDOR_ID"))
            }
        }
    }
    Component.onCompleted: {
        observer.start()
        observer.filter_by("usb")
    } 
}