问Linux:在PyQt5图形用户界面上显示当前连接的所有U盘的名称
EN

Stack Overflow用户

提问于 2019-03-27 04:55:34

回答 2查看 202关注 0票数 2

下面的代码显示了一个新插入的U盘的名称(作为PyQt5图形用户界面的替代)。

不幸的是，一拔出U盘就会在控制台中出现pyudev.device._errors.DeviceNotFoundAtPathError，而没有正确地弹出它。

需要进行哪些更改才能修复此错误？

main.py

from functools import partial
import os
import sys

import pyudev

from PyQt5.QtWidgets import QApplication
from PyQt5.QtCore import QSocketNotifier, QObject, pyqtSignal


class MainWindow():

    def __init__(self, parent=None):
        super().__init__()
        # GUI code
        pass

    def print_name(self, name):
        print(name)


class LinuxDeviceMonitor(QObject):
    devices_changed = pyqtSignal(list)

    def __init__(self):
        super().__init__()
        self._context = pyudev.Context()

        self._monitor = pyudev.Monitor.from_netlink(self._context)
        self._monitor.start()

        self._devices = set()

        self._process_devices(self._context.list_devices(), action="add")

    def fileno(self):
        return self._monitor.fileno()

    @property
    def device_names(self):
        return [pyudev.Devices.from_path(self._context, device).get("DEVNAME") for device in self._devices]

    def process_incoming(self):
        read_device = partial(pyudev._util.eintr_retry_call, self._monitor.poll, timeout=0)
        self._process_devices(iter(read_device, None))
        self.devices_changed.emit(self.device_names)

    def _process_devices(self, devices, action=None):
        for device in devices:
            action = device.action if action is None else action

            if action in ("add", "change") and self._is_usb_mass_storage_device(device):
                self._devices.add(device.sys_path)
            elif action == "remove" and device.sys_path in self._devices:
                self._devices.remove(device.sys_path)

    @classmethod
    def _read_device_flag(self, device, name):
        path = os.path.join(device.sys_path, name)
        try:
            with open(path) as data:
                return bool(int(data.read()))
        except (IOError, ValueError):
            return False

    def _is_usb_mass_storage_device(self, device):
        is_removable = self._read_device_flag(device, "removable")
        has_size = self._read_device_flag(device, "size")
        has_usb = device.get("ID_BUS") == "usb"
        has_no_disc = device.get("ID_CDROM") is None
        return is_removable and has_size and has_usb and has_no_disc


def main():
    app = QApplication(sys.argv)

    main_window = MainWindow()

    linux_device_monitor = LinuxDeviceMonitor()

    notifier = QSocketNotifier(linux_device_monitor.fileno(), QSocketNotifier.Read)
    notifier.activated.connect(linux_device_monitor.process_incoming)

    linux_device_monitor.devices_changed.connect(main_window.print_name)

    sys.exit(app.exec_())


if __name__ == '__main__':
    main()

python

pyqt5

pyudev

回答 2

Stack Overflow用户

回答已采纳

发布于 2019-03-29 06:09:20

我的第一个答案没有错，但是你可以用你自己的函数覆盖它。下面是针对您的问题的示例。主要的变化是：

添加：

from pyudev._util import ensure_byte_string

并重写了from_sys_path函数：

        pyudev.Devices.from_sys_path = self.from_sys_path

    def from_sys_path(self, context, sys_path):
        device = context._libudev.udev_device_new_from_syspath(
            context, ensure_byte_string(sys_path))
        if not device:
            return None
        return pyudev.Device(context, device)

已更改：

    @property
    def device_names(self):
        return [pyudev.Devices.from_path(self._context, device).get("DEVNAME") for device in self._devices]

至：

    def device_names(self):
        devices = []
        for device in self._devices:
            dev = pyudev.Devices.from_path(self._context, device)
            if dev is not None:
                devices.append(dev.get("DEVNAME"))
        return devices

和

        self.devices_changed.emit(self.device_names)

至

        self.devices_changed.emit(self.device_names())

整个代码如下所示：

from functools import partial
import os
import sys

import pyudev

from PyQt5.QtWidgets import QApplication
from PyQt5.QtCore import QSocketNotifier, QObject, pyqtSignal
from pyudev._util import ensure_byte_string


class MainWindow():

    def __init__(self, parent=None):
        super().__init__()
        # GUI code
        pass

    def print_name(self, name):
        print(name)


class LinuxDeviceMonitor(QObject):
    devices_changed = pyqtSignal(list)

    def __init__(self):
        super().__init__()
        self._context = pyudev.Context()

        self._monitor = pyudev.Monitor.from_netlink(self._context)
        self._monitor.start()

        self._devices = set()

        self._process_devices(self._context.list_devices(), action="add")
        pyudev.Devices.from_sys_path = self.from_sys_path

    def from_sys_path(self, context, sys_path):
        device = context._libudev.udev_device_new_from_syspath(
            context, ensure_byte_string(sys_path))
        if not device:
            return None
        return pyudev.Device(context, device)

    def fileno(self):
        return self._monitor.fileno()

    def device_names(self):
        devices = []
        for device in self._devices:
            dev = pyudev.Devices.from_path(self._context, device)
            if dev is not None:
                devices.append(dev.get("DEVNAME"))
        return devices

    def process_incoming(self):
        read_device = partial(pyudev._util.eintr_retry_call, self._monitor.poll, timeout=0)
        self._process_devices(iter(read_device, None))
        self.devices_changed.emit(self.device_names())

    def _process_devices(self, devices, action=None):
        for device in devices:
            action = device.action if action is None else action

            if action in ("add", "change") and self._is_usb_mass_storage_device(device):
                self._devices.add(device.sys_path)
            elif action == "remove" and device.sys_path in self._devices:
                self._devices.remove(device.sys_path)

    @classmethod
    def _read_device_flag(self, device, name):
        path = os.path.join(device.sys_path, name)
        try:
            with open(path) as data:
                return bool(int(data.read()))
        except (IOError, ValueError):
            return False

    def _is_usb_mass_storage_device(self, device):
        is_removable = self._read_device_flag(device, "removable")
        has_size = self._read_device_flag(device, "size")
        has_usb = device.get("ID_BUS") == "usb"
        has_no_disc = device.get("ID_CDROM") is None
        return is_removable and has_size and has_usb and has_no_disc


def main():
    app = QApplication(sys.argv)

    main_window = MainWindow()

    linux_device_monitor = LinuxDeviceMonitor()

    notifier = QSocketNotifier(linux_device_monitor.fileno(), QSocketNotifier.Read)
    notifier.activated.connect(linux_device_monitor.process_incoming)

    linux_device_monitor.devices_changed.connect(main_window.print_name)

    sys.exit(app.exec_())


if __name__ == '__main__':
    main()

您的集成开发环境可能会抱怨您正在访问一个受保护的模块成员(pyudev._util)，但它仍然可以工作。

票数 1

Stack Overflow用户

发布于 2019-03-27 07:01:27

您可以简单地捕获异常：

变化

    @property
    def device_names(self):
        return [pyudev.Devices.from_path(self._context, device).get("DEVNAME") for device in self._devices]

至

    def device_names(self):
        devices = []
        for device in self._devices:
            try:
                dev_name = pyudev.Devices.from_path(self._context, device).get("DEVNAME")
                devices.append(dev_name)
            except pyudev.DeviceNotFoundAtPathError:
                pass
        return devices

和

        self.devices_changed.emit(self.device_names)

至

        self.devices_changed.emit(self.device_names())

完整的代码：

from functools import partial
import os
import sys

import pyudev

from PyQt5.QtWidgets import QApplication
from PyQt5.QtCore import QSocketNotifier, QObject, pyqtSignal


class MainWindow():

    def __init__(self, parent=None):
        super().__init__()
        # GUI code
        pass

    def print_name(self, name):
        print(name)


class LinuxDeviceMonitor(QObject):
    devices_changed = pyqtSignal(list)

    def __init__(self):
        super().__init__()
        self._context = pyudev.Context()

        self._monitor = pyudev.Monitor.from_netlink(self._context)
        self._monitor.start()

        self._devices = set()

        self._process_devices(self._context.list_devices(), action="add")

    def fileno(self):
        return self._monitor.fileno()

    def device_names(self):
        devices = []
        for device in self._devices:
            try:
                dev_name = pyudev.Devices.from_path(self._context, device).get("DEVNAME")
                devices.append(dev_name)
            except pyudev.DeviceNotFoundAtPathError:
                pass
        return devices

    def process_incoming(self):
        read_device = partial(pyudev._util.eintr_retry_call, self._monitor.poll, timeout=0)
        self._process_devices(iter(read_device, None))
        self.devices_changed.emit(self.device_names())

    def _process_devices(self, devices, action=None):
        for device in devices:
            action = device.action if action is None else action

            if action in ("add", "change") and self._is_usb_mass_storage_device(device):
                self._devices.add(device.sys_path)
            elif action == "remove" and device.sys_path in self._devices:
                self._devices.remove(device.sys_path)

    @classmethod
    def _read_device_flag(self, device, name):
        path = os.path.join(device.sys_path, name)
        try:
            with open(path) as data:
                return bool(int(data.read()))
        except (IOError, ValueError):
            return False

    def _is_usb_mass_storage_device(self, device):
        is_removable = self._read_device_flag(device, "removable")
        has_size = self._read_device_flag(device, "size")
        has_usb = device.get("ID_BUS") == "usb"
        has_no_disc = device.get("ID_CDROM") is None
        return is_removable and has_size and has_usb and has_no_disc


def main():
    app = QApplication(sys.argv)

    main_window = MainWindow()

    linux_device_monitor = LinuxDeviceMonitor()

    notifier = QSocketNotifier(linux_device_monitor.fileno(), QSocketNotifier.Read)
    notifier.activated.connect(linux_device_monitor.process_incoming)

    linux_device_monitor.devices_changed.connect(main_window.print_name)

    sys.exit(app.exec_())


if __name__ == '__main__':
    main()

插入和移除usb记忆棒时的输出：

[]
[]
[]
['/dev/sdc']
['/dev/sdc']
['/dev/sdc']
[]
[]
[]

票数 1

页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持

原文链接：

https://stackoverflow.com/questions/55366051

复制

相似问题

问Linux:在PyQt5图形用户界面上显示当前连接的所有U盘的名称
EN

回答 2

Stack Overflow用户

Stack Overflow用户

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问Linux:在PyQt5图形用户界面上显示当前连接的所有U盘的名称EN

回答 2

Stack Overflow用户

Stack Overflow用户

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问Linux:在PyQt5图形用户界面上显示当前连接的所有U盘的名称
EN