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
  34 MAX_EVENT_SIZE = ctypes.sizeof(inotify_event) + NAME_MAX + 1
  35
  36
  37 def _errcheck(result, func, arguments):
  38     if result >= 0:
  39         return result
  40     err = ctypes.get_errno()
  41     if err == errno.EINTR:
  42         return func(*arguments)
  43     raise OSError(err, os.strerror(err))
  44
  45
  46 try:
  47     _libc = ctypes.CDLL(ctypes.util.find_library('c'), use_errno=True)
  48     _read = _libc.read
  49     init = _libc.inotify_init
  50     add_watch = _libc.inotify_add_watch
  51     rm_watch = _libc.inotify_rm_watch
  52 except AttributeError:
  53     raise ImportError('Could not load inotify functions from libc')
  54
  55
  56 _read.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_size_t]
  57 _read.errcheck = _errcheck
  58
  59 init.argtypes = []
  60 init.errcheck = _errcheck
  61
  62 add_watch.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_uint32]
  63 add_watch.errcheck = _errcheck
  64
  65 rm_watch.argtypes = [ctypes.c_int, ctypes.c_int]
  66 rm_watch.errcheck = _errcheck
  67
  68
  69 def read_events(inotify_fd, count=64):
  70     buf = ctypes.create_string_buffer(MAX_EVENT_SIZE * count)
  71     num = _read(inotify_fd, buf, ctypes.sizeof(buf))
  72
  73     addr = ctypes.addressof(buf)
  74     while num:
  75         assert num >= ctypes.sizeof(inotify_event)
  76         event = inotify_event.from_address(addr)
  77         addr += ctypes.sizeof(inotify_event)
  78         num -= ctypes.sizeof(inotify_event)
  79         if event.len:
  80             assert num >= event.len
  81             name = ctypes.string_at(addr)
  82             addr += event.len
  83             num -= event.len
  84         else:
  85             name = None
  86         yield event.wd, event.mask, event.cookie, name