Zugriff auf Inhalt eines USB-Sticks mittels USB Device ID

[__deets__]

@sparrow: diese Ausgaben basieren auf den Ereignissen von udev, die man direkt abgreifen kann. Mit dem Code der oben steht.

[__deets__]

Und das Kommando lsblk liefert einem für blockdevices ihre mountpoints und anders, im JSON Format.

[sparrow]

Ich sehe noch ein Problem darin auf das Anstecken eins USB-Sticks zu reagieren.
Zumindest unter Kubuntu, da probiere ich das gerade aus, bedeutet das Einstecken eins USB-Sticks nicht, dass der auch automatisch gemounted wird. Der taucht dann zwar in der Geräteüberwachung auf, wird aber erst gemounted, wenn darüber darauf zugegriffen wird.

[__deets__]

Das hat erstmal nichts mit den Ereignissen zu tun. Das anstecken läuft garantiert in udev auf, was auch immer Ubuntu da sonst tun mag, folgt daraus.

Worauf wird denn da wie zugegriffen? Prinzipiell ist es ja zb mit pmount möglich auch ohne super User Geräte zu mounten. Was auch immer das dann auslöst.

[Atalanttore]

@sparrow: Danke für den vielen Code. Momentan bin ich allerdings mit dem Code von __deets__ für Linux noch vollkommen ausgelastet.


Bei der Erkennung von USB-Speichern konnte ich mein DVD-Laufwerk mittlerweile erfolgreich ausschließen. Ich habe noch etwas weiter an den Ausschlusskriterien geschraubt, damit auch per USB angeschlossene, externe DVD-Laufwerke nicht berücksichtigt werden.

Mittels `QSocketNotifier` konnte ich dann tatsächlich ein Signal auslösen, wenn ein USB-Stick an- oder abgesteckt wird. Allerdings ist da noch einiges im Argen. Der `QSocketNotifier` ist wohl nicht ganz korrekt eingebunden.

• USB-Stick bei Programmstart nicht angesteckt:
  1. Wenn man ohne angestecktem USB-Stick das Programm startet, wird auf der Qt-Oberfläche ganz korrekt kein Pfad angezeigt.
  2. Steckt man den USB-Stick ein, während das Programm läuft, wird `[]` angezeigt. Anscheinend ist die Erkennung des USB-Sticks nicht schnell genug und es wird eine leere Liste zurückgegeben.
  3. Wenn man den USB-Stick wieder entfernt hat, wird weiterhin `[]` angezeigt.
• USB-Stick bei Programmstart angesteckt:
  1. Mit angestecktem USB-Start beim Programmstart wird kein Pfad auf der Qt-Oberfläche angezeigt.
  2. Erst sobald man den USB-Stick entfernt, erscheint der Pfad und die Fehlermeldung unten läuft in Endlosschleife in der Konsole.
  3. Steckt man den USB-Stick wieder ein, erscheinen in der Konsole keine Fehlermeldungen mehr.

Fehlermeldung:

Code: Alles auswählen

Traceback (most recent call last):
  File "/home/ata/source/usb-detect/main.py", line 104, in <lambda>
    notifier.activated.connect(lambda: main_window.update_path(str(linux_device_monitor.devices)))
  File "/home/ata/source/usb-detect/main.py", line 50, in devices
    return [pyudev.Devices.from_path(self._context, device).get("DEVNAME") for device in self._devices]
  File "/home/ata/source/usb-detect/main.py", line 50, in <listcomp>
    return [pyudev.Devices.from_path(self._context, device).get("DEVNAME") for device in self._devices]
  File "/usr/lib/python3/dist-packages/pyudev/device/_device.py", line 83, in from_path
    return cls.from_sys_path(context, path)
  File "/usr/lib/python3/dist-packages/pyudev/device/_device.py", line 108, in from_sys_path
    raise DeviceNotFoundAtPathError(sys_path)
pyudev.device._errors.DeviceNotFoundAtPathError: No device at '/sys/devices/pci0000:00/0000:00:14.0/usb1/1-6/1-6:1.0/host4/target4:0:0/4:0:0:0/block/sdb'

main.py:

Code: Alles auswählen

from functools import partial
import os
import sys
from PyQt5.QtWidgets import QApplication, QMainWindow
from PyQt5.uic import loadUi
from PyQt5.QtCore import QSocketNotifier


class MainWindow(QMainWindow):

    def __init__(self, parent=None):
        super().__init__(parent)
        loadUi("mainwindow.ui", self)

    def update_path(self, path):
        self.label_usb_mass_storage_url.setText(path)


class LinuxDeviceMonitor:

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

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

        self._devices = set()
        self._listeners = []
        self.rescan()

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

    def add_listener(self, listener):
        self._listeners.append(listener)

    def remove_listener(self, listener):
        self._listeners.remove(listener)

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

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

    def process_incoming(self, *_args):
        from pyudev._util import eintr_retry_call
        read_device = partial(eintr_retry_call, self._monitor.poll, timeout=0)
        self._process_devices(iter(read_device, None))

    def _process_devices(self, devices, action=None):
        old = set(self._devices)
        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 "remove" == action and device.sys_path in self._devices:
                self._devices.remove(device.sys_path)
        if self._devices != old:
            devices = self.devices
            for listener in self._listeners:
                listener(devices)

    def _is_usb_mass_storage_device(self, device):
        def dev_path(device, name):
            return "{}/{}".format(device.sys_path, name)

        def removable(device):
            removable_path = dev_path(device, "removable")
            try:
                return os.path.exists(removable_path) and int(open(removable_path).read())
            except ValueError:
                return False

        def has_size(device):
            size_path = dev_path(device, "size")
            try:
                return os.path.exists(size_path) and int(open(size_path).read())
            except ValueError:
                return False

        def has_usb(device):
            return device.get("ID_BUS") == "usb"

        def has_no_disc(device):
            return device.get("ID_CDROM") is None

        return removable(device) and has_size(device) and has_usb(device) and has_no_disc(device)


def main():
    app = QApplication(sys.argv)
    main_window = MainWindow()
    main_window.show()
    linux_device_monitor = LinuxDeviceMonitor()

    notifier = QSocketNotifier(linux_device_monitor.fileno(), QSocketNotifier.Read, linux_device_monitor.process_incoming())
    notifier.activated.connect(lambda: main_window.update_path(str(linux_device_monitor.devices)))

    sys.exit(app.exec_())

if __name__ == '__main__':
    main()

mainwindow.ui:

Code: Alles auswählen

<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
 <class>MainWindow</class>
 <widget class="QMainWindow" name="MainWindow">
  <property name="geometry">
   <rect>
    <x>0</x>
    <y>0</y>
    <width>500</width>
    <height>250</height>
   </rect>
  </property>
  <property name="windowTitle">
   <string>MainWindow</string>
  </property>
  <widget class="QWidget" name="centralwidget">
   <layout class="QGridLayout" name="gridLayout">
    <item row="0" column="0">
     <widget class="QLabel" name="label_path_to_mass_storage">
      <property name="text">
       <string>USB-Massenspeicher:</string>
      </property>
     </widget>
    </item>
    <item row="0" column="1">
     <widget class="QLabel" name="label_usb_mass_storage_url">
      <property name="text">
       <string/>
      </property>
     </widget>
    </item>
   </layout>
  </widget>
  <widget class="QMenuBar" name="menubar">
   <property name="geometry">
    <rect>
     <x>0</x>
     <y>0</y>
     <width>500</width>
     <height>28</height>
    </rect>
   </property>
  </widget>
  <widget class="QStatusBar" name="statusbar"/>
 </widget>
 <resources/>
 <connections/>
</ui>

Gruß
Atalanttore

[sparrow]

Wann genau willst du eigentlich die USB Device ID auswerten, die du im Threadtitel geschriebebn hast?
Und was meinst du mit der Device ID?

[__deets__]

Du rufst process_incoming AUF, statt es nur als callback zu übergeben.

[__deets__]

Ok, nochmal genauer hingeschaut. Das ist ja noch etwas kruder... die process_incoming-Methode muss aufgerufen werden, wenn etwas auf dem Socket passiert. Was du stattdessen machst ist, deren Ruckgabewert (der zufällig None ist) als Parent an den notifier zu übergeben. WTF?

Stattdessen musst du den im activate Signal die process_incoming aufrufen lassen. Und die kann dann ggf zu Änderungen der GUI auffordern.

[Atalanttore]

Wann genau willst du eigentlich die USB Device ID auswerten, die du im Threadtitel geschriebebn hast?
Und was meinst du mit der Device ID?

Mit Device ID meinte ich die Vendor ID und Product ID von USB-Geräten. Eigentlich wollte ich damit den Einhängepunkt eines USB-Sticks herausfinden, aber das funktioniert offensichtlich nicht so einfach und auch nicht plattformunabhängig.

[Atalanttore]

Du rufst process_incoming AUF, statt es nur als callback zu übergeben.

Ok, nochmal genauer hingeschaut. Das ist ja noch etwas kruder... die process_incoming-Methode muss aufgerufen werden, wenn etwas auf dem Socket passiert. Was du stattdessen machst ist, deren Ruckgabewert (der zufällig None ist) als Parent an den notifier zu übergeben. WTF?

Stattdessen musst du den im activate Signal die process_incoming aufrufen lassen. Und die kann dann ggf zu Änderungen der GUI auffordern.

Ohne Beispielcode musste ich mir den Code mithilfe der Doku selbst erstellen. Dabei kommt es bei mir regelmäßig zu Kollateralschäden, aber mit deinen Verbesserungsvorschlägen funktioniert es jetzt.

Nach dem Abziehen des USB-Sticks (ohne ordnungsgemäßem, vorherigen Aushängen) erscheint allerdings immer folgende Fehlermeldung in der Konsole:

Code: Alles auswählen

Traceback (most recent call last):
  File "/home/ata/source//usb-detect/main.py", line 58, in process_incoming
    self.main_window.update_path(str(self.devices))
  File "/home/ata/source//usb-detect/main.py", line 52, in devices
    return [pyudev.Devices.from_path(self._context, device).get("DEVNAME") for device in self._devices]
  File "/home/ata/source//usb-detect/main.py", line 52, in <listcomp>
    return [pyudev.Devices.from_path(self._context, device).get("DEVNAME") for device in self._devices]
  File "/usr/lib/python3/dist-packages/pyudev/device/_device.py", line 83, in from_path
    return cls.from_sys_path(context, path)
  File "/usr/lib/python3/dist-packages/pyudev/device/_device.py", line 108, in from_sys_path
    raise DeviceNotFoundAtPathError(sys_path)
pyudev.device._errors.DeviceNotFoundAtPathError: No device at '/sys/devices/pci0000:00/0000:00:14.0/usb1/1-6/1-6:1.0/host4/target4:0:0/4:0:0:0/block/sdb'

Wenn man den USB-Stick ordnungsgemäß aushängt, kommt es beim Abziehen zu keiner Fehlermeldung.

Lässt sich diese Fehlermeldung vermeiden oder ist das normal, wenn man ein USB-Gerät einfach abzieht?

main.py:

Code: Alles auswählen

from functools import partial
import os
import sys
from PyQt5.QtWidgets import QApplication, QMainWindow
from PyQt5.uic import loadUi
from PyQt5.QtCore import QSocketNotifier


class MainWindow(QMainWindow):

    def __init__(self, parent=None):
        super().__init__(parent)
        loadUi("mainwindow.ui", self)

    def update_path(self, path):
        self.label_usb_mass_storage_url.setText(path)


class LinuxDeviceMonitor:

    def __init__(self, main_window):
        import pyudev
        self._context = pyudev.Context()

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

        self._devices = set()
        self._listeners = []
        self.rescan()

        self.main_window = main_window

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

    def add_listener(self, listener):
        self._listeners.append(listener)

    def remove_listener(self, listener):
        self._listeners.remove(listener)

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

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

    def process_incoming(self, *_args):
        from pyudev._util import eintr_retry_call
        read_device = partial(eintr_retry_call, self._monitor.poll, timeout=0)
        self._process_devices(iter(read_device, None))
        self.main_window.update_path(str(self.devices))

    def _process_devices(self, devices, action=None):
        old = set(self._devices)
        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 "remove" == action and device.sys_path in self._devices:
                self._devices.remove(device.sys_path)
        if self._devices != old:
            devices = self.devices
            for listener in self._listeners:
                listener(devices)

    def _is_usb_mass_storage_device(self, device):
        def dev_path(device, name):
            return "{}/{}".format(device.sys_path, name)

        def removable(device):
            removable_path = dev_path(device, "removable")
            try:
                return os.path.exists(removable_path) and int(open(removable_path).read())
            except ValueError:
                return False

        def has_size(device):
            size_path = dev_path(device, "size")
            try:
                return os.path.exists(size_path) and int(open(size_path).read())
            except ValueError:
                return False

        def has_usb(device):
            return device.get("ID_BUS") == "usb"

        def has_no_disc(device):
            return device.get("ID_CDROM") is None

        return removable(device) and has_size(device) and has_usb(device) and has_no_disc(device)


def main():
    app = QApplication(sys.argv)
    main_window = MainWindow()
    main_window.show()
    linux_device_monitor = LinuxDeviceMonitor(main_window)

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

    sys.exit(app.exec_())

if __name__ == '__main__':
    main()

Gruß
Atalanttore

[sparrow]

Mit Device ID meinte ich die Vendor ID und Product ID von USB-Geräten. Eigentlich wollte ich damit den Einhängepunkt eines USB-Sticks herausfinden, aber das funktioniert offensichtlich nicht so einfach und auch nicht plattformunabhängig.

Trotz __deets__ Anmerkungen: du hast gesehen, dass mein Code das für Windows und Linux tut?

[Atalanttore]

@sparrow: Ja, wenn man beim Aufruf von `get_path_for_usbid()` die Vendor ID und Product ID des USB-Sticks übergibt, dann erhält man den Pfad zum USB-Stick. Unter Ubuntu hat es bei mir funktioniert. Unter Windows habe ich es noch nicht getestet.

Der Code von __deets__ dagegen liefert zur Laufzeit die Namen aller gerade eingesteckten USB-Sticks unter Linux zurück. Mit `lsblk -o MOUNTPOINT <Name des USB-Sticks>` konnte ich anschließend den Pfad zum USB-Stick ausgeben lassen.

Eine Kombination der Vorteile von beiden Programmen wäre ideal. Also ohne Polling eine ständig aktuelle Liste mit den Pfaden zu allen gerade eingesteckten USB-Sticks unter Linux bzw. Windows.

Gruß
Atalanttore

[__deets__]

Warum du str() um das set aufrufst ist mir unklar. Die Kopplung des device monitors an das Fenster ist ebenfalls extrem unschön. Du arbeitest doch die ganze Zeit mit Qt. Das Prinzip von Dingen, die Signale aussenden, und Dingen, die sei dann verarbeiten ohne von ersteren intime Kenntnisse zu haben wird da doch andauern vorgelebt. Warum probierst du das nicht mal so?

Und ich vermute mal, das das abrupfen des Sticks Events auslöst, die der bestehende Code nicht korrekt verarbeitet, und dadurch bleibt een device Eintrag bestehen, auch wenn das device nicht mehr existiert. Du kannst also entweder die richtigen Events das device entfernen lassen, oder den Fehler als das dazu gehörige Ereignis benuzten.

[Atalanttore]

@__deets__: Die Methode `setText()` erwartet einen String. Deshalb die Umwandlung mittels `str()`. Mit einer Liste kommt es zu folgendem Fehler:

Code: Alles auswählen

TypeError: setText(self, str): argument 1 has unexpected type 'list'

Was meinst du mit "intimen Kenntnissen"?

Gruß
Atalanttore

[__deets__]

Das ist nicht, was ich meinte. Du programmierst falsch, wenn du ein eine solche Kopplung von "dieses Objekt hier hat Daten" und "dieses Objekt will diese Daten aber auf genau diese art und weise haben" herstellst. Wenn der Linux-Device-Monitor 10 unterschiedliche Quellen in deinem Programm bedient, schreibst du dann 10 mal speziellen Code in dem Monitor? Das wäre keine gute Idee.

Stattdessen macht man ein Signal, dass die Devices als set verschickt. Und jedes, das sich dafür interessiert, muss das dann so umwandeln, wie es das benötigt.

[Atalanttore]

Ist ein Signal eine Art globaler Variable, wo a) alle damit verbundenen Slots bei einer Änderung benachrichtigt werden und b) man an jeder Stelle im Code darauf zugreifen kann?

Gruß
Atalanttore

[__deets__]

Nein. Warum global? Du benutzt doch andauernd Signale und Slots in Qt. Sind die global? Hast du also nur ein clicked für alle buttons die in deinem Programm sind?

[sls]

@Atalanttore: bei deiner GUI-Anwendung gibt es einen Main-Thread der ewig läuft, der empfängt Systemnachrichten (Signale) und kann dann wiederrum andere Threads benachrichtigen. Es kann nicht jeder Thread einfach so auf ein Signal zugreifen.

[__deets__]

@sls: das Konzept der Signale hat mit dem System erstmal wenig zu tun. Und was hat das mit Threads zu tun?

Oh. Moment. Meinst du Signale wie im signals Modul? Das ist hier nicht gemeint. Sondern signal/slots in Qt. Die Attlantore ja auch munter benutzt.

[sls]

@__deets__: richtig, sorry, hätte mir erst deinen letzten Post auf Seite 3 durchlesen sollen. Leider falscher Kontext.