10 from pyinotify import EventsCodes
12 from benderjab import rpc
15 class Handler(pyinotify.ProcessEvent):
16 def __init__(self, watchmanager, bot):
17 self.last_event_time = None
18 self.watchmanager = watchmanager
21 def process_IN_CREATE(self, event):
22 self.last_event_time = time.time()
23 msg = "Create: %s" % os.path.join(event.path, event.name)
24 if event.name.lower() == "run.completed":
26 self.bot.runFinished(event.path)
28 logging.error("Couldn't send runFinished")
31 def process_IN_DELETE(self, event):
32 logging.debug("Remove: %s" % os.path.join(event.path, event.name))
34 def process_IN_UNMOUNT(self, event):
35 self.bot.unmount_watch()
37 class SpoolWatcher(rpc.XmlRpcBot):
39 Watch a directory and send a message when another process is done writing.
41 This monitors a directory tree using inotify (linux specific) and
42 after some files having been written will send a message after <timeout>
43 seconds of no file writing.
45 (Basically when the solexa machine finishes dumping a round of data
46 this'll hopefully send out a message saying hey look theres data available
49 # these params need to be in the config file
50 # I wonder where I should put the documentation
52 # `watchdir` - which directory tree to monitor for modifications
53 # `profile` - specify which .gaworkflow profile to use
54 # `write_timeout` - how many seconds to wait for writes to finish to
56 # `notify_timeout` - how often to timeout from notify
58 def __init__(self, section=None, configfile=None):
59 #if configfile is None:
60 # self.configfile = "~/.gaworkflow"
61 super(SpoolWatcher, self).__init__(section, configfile)
63 self.cfg['watchdir'] = None
64 self.cfg['write_timeout'] = 10
65 self.cfg['notify_users'] = None
66 self.cfg['notify_runner'] = None
68 self.notify_timeout = 0.001
69 self.wm = pyinotify.WatchManager()
70 self.handler = Handler(self.wm, self)
71 self.notifier = pyinotify.Notifier(self.wm, self.handler)
74 self.notify_users = None
75 self.notify_runner = None
77 self.eventTasks.append(self.process_notify)
79 def read_config(self, section=None, configfile=None):
80 super(SpoolWatcher, self).read_config(section, configfile)
82 self.watch_dir = self._check_required_option('watchdir')
83 self.write_timeout = int(self.cfg['write_timeout'])
85 self.notify_users = self._parse_user_list(self.cfg['notify_users'])
87 self.notify_runner = \
88 self._parse_user_list(self.cfg['notify_runner'],
89 require_resource=True)
90 except bot.JIDMissingResource:
91 msg = 'need a full jabber ID + resource for xml-rpc destinations'
93 raise bot.JIDMissingResource(msg)
95 def add_watch(self, watchdir=None):
97 start watching watchdir or self.watch_dir
98 we're currently limited to watching one directory tree.
100 # the one tree limit is mostly because self.wdd is a single item
101 # but managing it as a list might be a bit more annoying
103 watchdir = self.watch_dir
104 logging.info("Watching:"+str(watchdir))
105 mask = EventsCodes.IN_CREATE | EventsCodes.IN_UNMOUNT
106 # rec traverses the tree and adds all the directories that are there
108 # auto_add will add in new directories as they are created
109 self.wdd = self.wm.add_watch(watchdir, mask, rec=True, auto_add=True)
111 def unmount_watch(self):
112 if self.wdd is not None:
113 logging.debug("disabling watch")
114 logging.debug(str(self.wdd))
115 self.wm.rm_watch(self.wdd)
118 def process_notify(self, *args):
119 # process the queue of events as explained above
120 self.notifier.process_events()
121 #check events waits timeout
122 if self.notifier.check_events(self.notify_timeout):
123 # read notified events and enqeue them
124 self.notifier.read_events()
125 # should we do something?
126 last_event_time = self.handler.last_event_time
127 if last_event_time is not None:
128 time_delta = time.time() - last_event_time
129 if time_delta > self.write_timeout:
131 self.handler.last_event_time = None
133 def start(self, daemonize):
138 super(SpoolWatcher, self).start(daemonize)
144 # destroy the inotify's instance on this interrupt (stop monitoring)
146 super(SpoolWatcher, self).stop()
149 logging.debug("writes seem to have stopped")
150 if self.notify_runner is not None:
151 for r in self.notify_runner:
152 self.rpc_send(r, tuple(), 'startCopy')
154 def runFinished(self, run_dir):
155 # need to strip off self.watch_dir from rundir I suspect.
156 logging.info("run.completed in " + str(run_dir))
157 pattern = self.watch_dir
158 if pattern[-1] != os.path.sep:
159 pattern += os.path.sep
160 stripped_run_dir = re.sub(pattern, "", run_dir)
161 logging.debug("stripped to " + stripped_run_dir)
162 if self.notify_users is not None:
163 for u in self.notify_users:
164 self.send(u, 'Sequencing run %s finished' % (stripped_run_dir))
165 if self.notify_runner is not None:
166 for r in self.notify_runner:
167 self.rpc_send(r, (stripped_run_dir,), 'runFinished')
171 return bot.main(args)
173 if __name__ == "__main__":
174 sys.exit(main(sys.argv[1:]))