ctypes-inotify-sample.py

#!/usr/bin/env python

from ctypes import CDLL, CFUNCTYPE, c_char_p, c_int, c_uint32
from ctypes.util import find_library
import argparse
import os
import struct
import sys
import select

__libc = CDLL(find_library("c"))

inotify_init = CFUNCTYPE(c_int)(("inotify_init", __libc), ())
inotify_init1 = CFUNCTYPE(c_int, c_int)(("inotify_init", __libc), ((1, "flags"),))

inotify_add_watch = CFUNCTYPE(c_int, c_int, c_char_p, c_uint32)(
    ("inotify_add_watch", __libc),
    ((1, "fd"), (1, "pathname"), (1, "mask"))
)

inotify_rm_watch = CFUNCTYPE(c_int, c_int, c_int)(
    ("inotify_rm_watch", __libc),
    ((1, "fd"), (1, "wd"))
)

IN_ACCESS           = 0x00000001
IN_MODIFY           = 0x00000002
IN_ATTRIB           = 0x00000004
IN_CLOSE_WRITE      = 0x00000008
IN_CLOSE_NOWRITE    = 0x00000010
IN_OPEN             = 0x00000020
IN_MOVED_FROM       = 0x00000040
IN_MOVED_TO         = 0x00000080
IN_CREATE           = 0x00000100
IN_DELETE           = 0x00000200
IN_DELETE_SELF      = 0x00000400
IN_MOVE_SELF        = 0x00000800

IN_UNMOUNT          = 0x00002000
IN_Q_OVERFLOW       = 0x00004000
IN_IGNORED          = 0x00008000

IN_CLOSE            = IN_CLOSE_WRITE | IN_CLOSE_NOWRITE
IN_MOVE             = IN_MOVED_FROM | IN_MOVED_TO

IN_ONLYDIR          = 0x01000000
IN_DONT_FOLLOW      = 0x02000000
IN_EXCL_UNLINK      = 0x04000000
IN_MASK_ADD         = 0x20000000
IN_ISDIR            = 0x40000000
IN_ONESHOT          = 0x80000000

#IN_ALL_EVENTS       = 0xfff

def _decode_flag(flag):
    flag_names = [x for x in globals().keys() if x.startswith('IN_')]
    flag_names.remove('IN_CLOSE')
    flag_names.remove('IN_MOVE')
    flag_names.sort(key=lambda x: globals()[x])
    r = []
    for n in flag_names:
        if globals()[n] & flag != 0:
            r.append(n)
    return r

def main():

    def integer(x):
        return int(x, base=0)
    parser = argparse.ArgumentParser()
    parser.add_argument('path', nargs='?', default='.',
        help='directory or file to watch')
    parser.add_argument('-f', '--flag', action='append', default=[],
        choices=[x for x in globals().keys() if x.startswith('IN_')],
        metavar='FLAG', help='event name to watch')
    parser.add_argument('-m', '--mask', default=0, type=integer,
        help='numeric event mask to watch')
    parser.add_argument('-a', '--all', action='store_const',
        dest='mask', const=0xfff, help='watch all event(-m 0xfff)')
    args = parser.parse_args()

    mask = args.mask
    for x in args.flag:
        mask |= globals()[x]
    if mask == 0:
        mask = IN_ACCESS

    print('watching {} for inotify events: {}'.format(
        args.path, ",".join(_decode_flag(mask))))

    fd = inotify_init()
    wd = inotify_add_watch(fd, args.path.encode("utf-8"), mask)

    epoll = select.epoll()
    epoll.register(fd, select.EPOLLIN)

    try:

        while True:

            events = epoll.poll()
            for fileno, event in events:
                if event & select.EPOLLIN:
                    buf = os.read(fd, 4096)
                    i = 0
                    fmt = 'iIII'
                    fmt_size = struct.calcsize(fmt)
                    while i < len(buf):
                        wd, mask, cookie, name_len = struct.unpack_from(fmt, buf, i)
                        i += fmt_size
                        name = buf[i:i+name_len]
                        i += name_len
                        print("{:08x}({}) {} {}".format(
                            mask, ",".join(_decode_flag(mask)), cookie, name.decode("utf-8").strip("\0")))

    except KeyboardInterrupt as e:
        pass
    else:
        inotify_rm_watch(fd, wd)
        epoll.unregister(fd)
        epoll.close()

if __name__ == '__main__':
    sys.exit(main())