cola/inotify.py

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