cola/inotify.py

   1 import ctypes
   2 import ctypes.util
   3 import errno
   4 import os
   5
   6 # constant from Linux include/uapi/linux/limits.h
   7 NAME_MAX = 255
   8
   9 # constants from Linux include/uapi/linux/inotify.h
  10 IN_MODIFY       = 0x00000002
  11 IN_ATTRIB       = 0x00000004
  12 IN_CLOSE_WRITE  = 0x00000008
  13 IN_MOVED_FROM   = 0x00000040
  14 IN_MOVED_TO     = 0x00000080
  15 IN_CREATE       = 0x00000100
  16 IN_DELETE       = 0x00000200
  17
  18 IN_Q_OVERFLOW   = 0x00004000
  19
  20 IN_ONLYDIR      = 0x01000000
  21 IN_EXCL_UNLINK  = 0x04000000
  22 IN_ISDIR        = 0x80000000
  23
  24
  25 class inotify_event(ctypes.Structure):
  26     _fields_ = [
  27         ('wd', ctypes.c_int),
  28         ('mask', ctypes.c_uint32),
  29         ('cookie', ctypes.c_uint32),
  30         ('len', ctypes.c_uint32),
  31     ]
  32
  33 MAX_EVENT_SIZE = ctypes.sizeof(inotify_event) + NAME_MAX + 1
  34
  35
  36 def _errcheck(result, func, arguments):
  37     if result >= 0:
  38         return result
  39     err = ctypes.get_errno()
  40     if err == errno.EINTR:
  41         return func(*arguments)
  42     raise OSError(err, os.strerror(err))
  43
  44
  45 try:
  46     _libc = ctypes.CDLL(ctypes.util.find_library('c'), use_errno=True)
  47     _read = _libc.read
  48     init = _libc.inotify_init
  49     add_watch = _libc.inotify_add_watch
  50     rm_watch = _libc.inotify_rm_watch
  51 except AttributeError:
  52     raise ImportError('Could not load inotify functions from libc')
  53
  54
  55 _read.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_size_t]
  56 _read.errcheck = _errcheck
  57
  58 init.argtypes = []
  59 init.errcheck = _errcheck
  60
  61 add_watch.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_uint32]
  62 add_watch.errcheck = _errcheck
  63
  64 rm_watch.argtypes = [ctypes.c_int, ctypes.c_int]
  65 rm_watch.errcheck = _errcheck
  66
  67
  68 def read_events(inotify_fd, count=64):
  69     buf = ctypes.create_string_buffer(MAX_EVENT_SIZE * count)
  70     n = _read(inotify_fd, buf, ctypes.sizeof(buf))
  71
  72     addr = ctypes.addressof(buf)
  73     while n:
  74         assert n >= ctypes.sizeof(inotify_event)
  75         event = inotify_event.from_address(addr)
  76         addr += ctypes.sizeof(inotify_event)
  77         n -= ctypes.sizeof(inotify_event)
  78         if event.len:
  79             assert n >= event.len
  80             name = ctypes.string_at(addr)
  81             addr += event.len
  82             n -= event.len
  83         else:
  84             name = None
  85         yield event.wd, event.mask, event.cookie, name