小编典典

如何在Linux和Python中侦听“插入USB设备”事件?

linux

我想在Linux中为Amarok编写Python脚本,以自动将stackoverflow播客复制到播放器中。当我插入播放器时,它将安装驱动器,复制所有未决的播客,然后弹出播放器。如何收听“插入”事件?我浏览了一下,但找不到一个很好的例子。


阅读 483

收藏
2020-06-02

共1个答案

小编典典

更新 :如评论中所述,最近的发行版不支持Hal,现在的标准是udev,这是一个使用glib loop和 udev
的小示例,出于历史原因,我保留Hal版本。

这基本上是pyudev文档中示例,适用于较早版本和glib循环,请注意,应根据您的特定需求自定义过滤器:

import glib

from pyudev import Context, Monitor

try:
    from pyudev.glib import MonitorObserver

    def device_event(observer, device):
        print 'event {0} on device {1}'.format(device.action, device)
except:
    from pyudev.glib import GUDevMonitorObserver as MonitorObserver

    def device_event(observer, action, device):
        print 'event {0} on device {1}'.format(action, device)

context = Context()
monitor = Monitor.from_netlink(context)

monitor.filter_by(subsystem='usb')
observer = MonitorObserver(monitor)

observer.connect('device-event', device_event)
monitor.start()

glib.MainLoop().run()

具有Hal和d-bus的旧版本:

您可以使用D-Bus绑定并收听DeviceAddedDeviceRemoved信号。您必须检查“添加的设备”的功能才能仅选择存储设备。

这是一个小示例,您可以删除评论并尝试。

import dbus
import gobject

class DeviceAddedListener:
    def __init__(self):

您需要使用系统总线连接到Hal Manager。

        self.bus = dbus.SystemBus()
        self.hal_manager_obj = self.bus.get_object(
                                              "org.freedesktop.Hal", 
                                              "/org/freedesktop/Hal/Manager")
        self.hal_manager = dbus.Interface(self.hal_manager_obj,
                                          "org.freedesktop.Hal.Manager")

在这种情况下,您需要将一个侦听器连接到您感兴趣的信号DeviceAdded

        self.hal_manager.connect_to_signal("DeviceAdded", self._filter)

我正在使用基于功能的过滤器。如果您可以阅读Hal文档以找到更适合您需要的查询,或者有关Hal设备属性的更多信息,它将接受任何要求volumedo_something与之联系。

    def _filter(self, udi):
        device_obj = self.bus.get_object ("org.freedesktop.Hal", udi)
        device = dbus.Interface(device_obj, "org.freedesktop.Hal.Device")

        if device.QueryCapability("volume"):
            return self.do_something(device)

示例函数显示有关卷的一些信息:

     def do_something(self, volume):
        device_file = volume.GetProperty("block.device")
        label = volume.GetProperty("volume.label")
        fstype = volume.GetProperty("volume.fstype")
        mounted = volume.GetProperty("volume.is_mounted")
        mount_point = volume.GetProperty("volume.mount_point")
        try:
            size = volume.GetProperty("volume.size")
        except:
            size = 0

        print "New storage device detectec:"
        print "  device_file: %s" % device_file
        print "  label: %s" % label
        print "  fstype: %s" % fstype
        if mounted:
            print "  mount_point: %s" % mount_point
        else:
            print "  not mounted"
        print "  size: %s (%.2fGB)" % (size, float(size) / 1024**3)

if __name__ == '__main__':
    from dbus.mainloop.glib import DBusGMainLoop
    DBusGMainLoop(set_as_default=True)
    loop = gobject.MainLoop()
    DeviceAddedListener()
    loop.run()
2020-06-02