How to Listen For 'Usb Device Inserted' Events in Linux, in Python

How can I listen for 'usb device inserted' events in Linux, in Python?

Update: As said in comments, Hal is not supported in recent distributions, the standard now is udev, Here is a small example that makes use of glib loop and udev, I keep the Hal version for historical reasons.

This is basically the example in the pyudev documentation, adapted to work with older versions, and with the glib loop, notice that the filter should be customized for your specific needing:

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()

Old version with Hal and d-bus:

You can use D-Bus bindings and listen to DeviceAdded and DeviceRemoved signals.
You will have to check the capabilities of the Added device in order to select the storage devices only.

Here is a small example, you can remove the comments and try it.

import dbus
import gobject

class DeviceAddedListener:
    def __init__(self):

You need to connect to Hal Manager using the System Bus.

        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")

And you need to connect a listener to the signals you are interested on, in this case DeviceAdded.

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

I'm using a filter based on capabilities. It will accept any volume and will call do_something with if, you can read Hal documentation to find the more suitable queries for your needs, or more information about the properties of the Hal devices.

    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)

Example function that shows some information about the volume:

     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()

How can I detect using Python the insertion of only USBs and hard drives on Ubuntu/Linux?

I can detect it rather easily through monitoring the /dev/disks/by-label/ directory.

Detect inserted USB on Windows

Yes, you need to use the RegisterDeviceNotification Windows API call. As far as I know, there is no Python module that wraps this functionality, so you have to use ctypes to call this function.

Fortunately, you are not the first person who has wanted to do this, so there are some code samples floating around the web. WxPython provides a code sample, but as you are writing a daemon this may not interest you. You might want to try the following code sample, which relies on both ctypes and pywin32, lifted shameless from Tim Golden:

import win32serviceutil
import win32service
import win32event
import servicemanager

import win32gui
import win32gui_struct
struct = win32gui_struct.struct
pywintypes = win32gui_struct.pywintypes
import win32con

GUID_DEVINTERFACE_USB_DEVICE = "{A5DCBF10-6530-11D2-901F-00C04FB951ED}"
DBT_DEVICEARRIVAL = 0x8000
DBT_DEVICEREMOVECOMPLETE = 0x8004

import ctypes

#
# Cut-down clone of UnpackDEV_BROADCAST from win32gui_struct, to be
# used for monkey-patching said module with correct handling
# of the "name" param of DBT_DEVTYPE_DEVICEINTERFACE
#
def _UnpackDEV_BROADCAST (lparam):
  if lparam == 0: return None
  hdr_format = "iii"
  hdr_size = struct.calcsize (hdr_format)
  hdr_buf = win32gui.PyGetMemory (lparam, hdr_size)
  size, devtype, reserved = struct.unpack ("iii", hdr_buf)
  # Due to x64 alignment issues, we need to use the full format string over
  # the entire buffer.  ie, on x64:
  # calcsize('iiiP') != calcsize('iii')+calcsize('P')
  buf = win32gui.PyGetMemory (lparam, size)

  extra = {}
  if devtype == win32con.DBT_DEVTYP_DEVICEINTERFACE:
    fmt = hdr_format + "16s"
    _, _, _, guid_bytes = struct.unpack (fmt, buf[:struct.calcsize(fmt)])
    extra['classguid'] = pywintypes.IID (guid_bytes, True)
    extra['name'] = ctypes.wstring_at (lparam + struct.calcsize(fmt))
  else:
    raise NotImplementedError("unknown device type %d" % (devtype,))
  return win32gui_struct.DEV_BROADCAST_INFO(devtype, **extra)
win32gui_struct.UnpackDEV_BROADCAST = _UnpackDEV_BROADCAST

class DeviceEventService (win32serviceutil.ServiceFramework):

  _svc_name_ = "DevEventHandler"
  _svc_display_name_ = "Device Event Handler"
  _svc_description_ = "Handle device notification events"

  def __init__(self, args):
    win32serviceutil.ServiceFramework.__init__ (self, args)
    self.hWaitStop = win32event.CreateEvent (None, 0, 0, None)
    #
    # Specify that we're interested in device interface
    # events for USB devices
    #
    filter = win32gui_struct.PackDEV_BROADCAST_DEVICEINTERFACE (
      GUID_DEVINTERFACE_USB_DEVICE
    )
    self.hDevNotify = win32gui.RegisterDeviceNotification (
      self.ssh, # copy of the service status handle
      filter,
      win32con.DEVICE_NOTIFY_SERVICE_HANDLE
    )

  #
  # Add to the list of controls already handled by the underlying
  # ServiceFramework class. We're only interested in device events
  #
  def GetAcceptedControls(self):
    rc = win32serviceutil.ServiceFramework.GetAcceptedControls (self)
    rc |= win32service.SERVICE_CONTROL_DEVICEEVENT
    return rc

  #
  # Handle non-standard service events (including our device broadcasts)
  # by logging to the Application event log
  #
  def SvcOtherEx(self, control, event_type, data):
    if control == win32service.SERVICE_CONTROL_DEVICEEVENT:
      info = win32gui_struct.UnpackDEV_BROADCAST(data)
      #
      # This is the key bit here where you'll presumably
      # do something other than log the event. Perhaps pulse
      # a named event or write to a secure pipe etc. etc.
      #
      if event_type == DBT_DEVICEARRIVAL:
        servicemanager.LogMsg (
          servicemanager.EVENTLOG_INFORMATION_TYPE,
          0xF000,
          ("Device %s arrived" % info.name, '')
        )
      elif event_type == DBT_DEVICEREMOVECOMPLETE:
        servicemanager.LogMsg (
          servicemanager.EVENTLOG_INFORMATION_TYPE,
          0xF000,
          ("Device %s removed" % info.name, '')
        )

  #
  # Standard stuff for stopping and running service; nothing
  # specific to device notifications
  #
  def SvcStop(self):
    self.ReportServiceStatus (win32service.SERVICE_STOP_PENDING)
    win32event.SetEvent (self.hWaitStop)

  def SvcDoRun(self):
    win32event.WaitForSingleObject (self.hWaitStop, win32event.INFINITE)
    servicemanager.LogMsg (
      servicemanager.EVENTLOG_INFORMATION_TYPE,
      servicemanager.PYS_SERVICE_STOPPED,
      (self._svc_name_, '')
    )

if __name__=='__main__':
  win32serviceutil.HandleCommandLine (DeviceEventService)

Linux USB Mapping Question

You should probably ask HAL about that. You say you already get notifications from HAL by D-Bus... It maintains list of USB devices, together with their IDs and device names (block.device property).

Here's a nice example of how to get device file name together with the notification of new USB device: How can I listen for 'usb device inserted' events in Linux, in Python?

USB devices UDev and D-BUS

The udisks D-Bus service, obviously, only reports disks.

Just monitor udev directly (through libudev, through pyudev).

import pyudev
context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
observer = pyudev.pygtk.GUDevMonitorObserver(monitor)
observer.connect('device-added', device_added_callback)
observer.connect('device-changed', device_changed_callback)
monitor.enable_receiving()
mainloop = gobject.MainLoop()
mainloop.run()

---

Related Topics