9 from htsworkflow.util import mount
13 from pyinotify import EventsCodes
15 from benderjab import rpc
17 def get_top_dir(root, path):
19 Return the directory in path that is a subdirectory of root.
22 >>> print get_top_dir('/a/b/c', '/a/b/c/d/e/f')
24 >>> print get_top_dir('/a/b/c/', '/a/b/c/d/e/f')
26 >>> print get_top_dir('/a/b/c', '/g/e/f')
28 >>> print get_top_dir('/a/b/c', '/a/b/c')
31 if path.startswith(root):
32 subpath = path[len(root):]
33 if subpath.startswith('/'):
35 return subpath.split(os.path.sep)[0]
39 class WatcherEvents(object):
40 # two events need to be tracked
41 # one to send startCopy
42 # one to send OMG its broken
43 # OMG its broken needs to stop when we've seen enough
45 # this should be per runfolder.
51 class Handler(pyinotify.ProcessEvent):
52 def __init__(self, watchmanager, bot, ipar=False):
54 ipar flag indicates we should wait for ipar to finish, instead of
55 just the run finishing
58 self.watchmanager = watchmanager
62 self.last_file = 'IPAR_Netcopy_Complete.txt'.lower()
64 self.last_file = "run.completed".lower()
66 def process_IN_CREATE(self, event):
67 for wdd in self.bot.wdds:
68 for watch_path in self.bot.watchdirs:
69 if event.path.startswith(watch_path):
70 target = get_top_dir(watch_path, event.path)
71 self.last_event.setdefault(watch_path, {})[target] = time.time()
73 msg = "Create: %s %s %s" % (event.path, event.name, target)
75 if event.name.lower() == self.last_file:
77 self.bot.sequencingFinished(event.path)
79 logging.error("Couldn't send sequencingFinished")
82 def process_IN_DELETE(self, event):
83 logging.debug("Remove: %s" % os.path.join(event.path, event.name))
86 def process_IN_UNMOUNT(self, event):
87 pathname = os.path.join(event.path, event.name)
88 logging.debug("IN_UNMOUNT: %s" % (pathname,))
89 self.bot.unmount_watch(event.path)
91 class SpoolWatcher(rpc.XmlRpcBot):
93 Watch a directory and send a message when another process is done writing.
95 This monitors a directory tree using inotify (linux specific) and
96 after some files having been written will send a message after <timeout>
97 seconds of no file writing.
99 (Basically when the solexa machine finishes dumping a round of data
100 this'll hopefully send out a message saying hey look theres data available
103 # these params need to be in the config file
104 # I wonder where I should put the documentation
106 # `watchdirs` - list of directories to monitor for modifications
107 # `profile` - specify which .htsworkflow profile to use
108 # `write_timeout` - how many seconds to wait for writes to finish to
110 # `notify_timeout` - how often to timeout from notify
112 def __init__(self, section=None, configfile=None):
113 #if configfile is None:
114 # self.configfile = "~/.htsworkflow"
115 super(SpoolWatcher, self).__init__(section, configfile)
117 self.cfg['watchdirs'] = None
118 self.cfg['write_timeout'] = 10
119 self.cfg['notify_users'] = None
120 self.cfg['notify_runner'] = None
121 self.cfg['wait_for_ipar'] = 0
124 self.watchdir_url_map = {}
125 self.notify_timeout = 0.001
128 self.notify_users = None
129 self.notify_runner = None
132 # keep track if the specified mount point is currently mounted
133 self.mounted_points = {}
134 # keep track of which mount points tie to which watch directories
135 # so maybe we can remount them.
136 self.mounts_to_watches = {}
138 self.eventTasks.append(self.process_notify)
140 def read_config(self, section=None, configfile=None):
141 # Don't give in to the temptation to use logging functions here,
142 # need to wait until after we detach in start
143 super(SpoolWatcher, self).read_config(section, configfile)
145 self.watchdirs = shlex.split(self._check_required_option('watchdirs'))
146 # see if there's an alternate url that should be used for the watchdir
147 for watchdir in self.watchdirs:
148 self.watchdir_url_map[watchdir] = self.cfg.get(watchdir, watchdir)
150 self.write_timeout = int(self.cfg['write_timeout'])
151 self.wait_for_ipar = int(self.cfg['wait_for_ipar'])
153 self.notify_users = self._parse_user_list(self.cfg['notify_users'])
155 self.notify_runner = \
156 self._parse_user_list(self.cfg['notify_runner'],
157 require_resource=True)
158 except bot.JIDMissingResource:
159 msg = 'need a full jabber ID + resource for xml-rpc destinations'
160 raise bot.JIDMissingResource(msg)
165 def add_watch(self, watchdirs=None):
167 start watching watchdir or self.watchdir
168 we're currently limited to watching one directory tree.
170 # create the watch managers if we need them
172 self.wm = pyinotify.WatchManager()
173 self.handler = Handler(self.wm, self, self.wait_for_ipar)
174 self.notifier = pyinotify.Notifier(self.wm, self.handler)
176 # the one tree limit is mostly because self.wdd is a single item
177 # but managing it as a list might be a bit more annoying
178 if watchdirs is None:
179 watchdirs = self.watchdirs
181 mask = EventsCodes.IN_CREATE | EventsCodes.IN_UNMOUNT
182 # rec traverses the tree and adds all the directories that are there
184 # auto_add will add in new directories as they are created
186 mount_location = mount.find_mount_point_for(w)
187 self.mounted_points[mount_location] = True
188 mounts = self.mounts_to_watches.get(mount_location, [])
191 self.mounts_to_watches[mount_location] = mounts
193 logging.info(u"Watching:"+unicode(w))
194 self.wdds.append(self.wm.add_watch(w, mask, rec=True, auto_add=True))
196 def unmount_watch(self, event_path):
197 # remove backwards so we don't get weirdness from
198 # the list getting shorter
199 for i in range(len(self.wdds),0, -1):
201 logging.info(u'unmounting: '+unicode(wdd.items()))
202 self.wm.rm_watch(wdd.values())
206 def make_copy_url(self, watchdir, list_event_dir):
207 root_copy_url = self.watchdir_url_map[watchdir]
208 if root_copy_url[-1] != '/':
210 copy_url = root_copy_url + list_event_dir
211 logging.debug('Copy url: %s' % (copy_url,))
214 def process_notify(self, *args):
215 if self.notifier is None:
218 # process the queue of events as explained above
219 self.notifier.process_events()
220 #check events waits timeout
221 if self.notifier.check_events(self.notify_timeout):
222 # read notified events and enqeue them
223 self.notifier.read_events()
224 # should we do something?
225 # has something happened?
226 for watchdir, last_events in self.handler.last_event.items():
227 for last_event_dir, last_event_time in last_events.items():
228 time_delta = time.time() - last_event_time
229 if time_delta > self.write_timeout:
230 copy_url = self.make_copy_url(watchdir, last_event_dir)
231 self.startCopy(copy_url)
232 self.handler.last_event[watchdir] = {}
233 # handle unmounted filesystems
234 for mount_point, was_mounted in self.mounted_points.items():
235 if not was_mounted and mount.is_mounted(mount_point):
236 # we've been remounted. Huzzah!
238 for watch in self.mounts_to_watches[mount_point]:
239 self.add_watch(watch)
241 "%s was remounted, restarting watch" % \
244 self.mounted_points[mount_point] = True
246 def _parser(self, msg, who):
248 Parse xmpp chat messages
250 help = u"I can send [copy] message, or squencer [finished]"
251 if re.match(u"help", msg):
253 elif re.match("copy", msg):
255 reply = u"sent copy message"
256 elif re.match(u"finished", msg):
259 self.sequencingFinished(words[1])
260 reply = u"sending sequencing finished for %s" % (words[1])
262 reply = u"need runfolder name"
264 reply = u"I didn't understand '%s'" %(msg)
271 # we have to configure pyinotify after BenderJab.start is called
272 # as weird things happen to pyinotify if the stdio is closed
273 # after it's initialized.
275 super(SpoolWatcher, self).run()
281 # destroy the inotify's instance on this interrupt (stop monitoring)
282 if self.notifier is not None:
284 super(SpoolWatcher, self).stop()
286 def startCopy(self, copy_url=None):
287 logging.debug("writes seem to have stopped")
288 if self.notify_runner is not None:
289 for r in self.notify_runner:
290 self.rpc_send(r, tuple([copy_url]), 'startCopy')
291 if self.notify_users is not None:
292 for u in self.notify_users:
293 self.send(u, 'startCopy %s.' % (copy_urls,))
295 def sequencingFinished(self, run_dir):
296 # need to strip off self.watchdirs from rundir I suspect.
297 logging.info("run.completed in " + str(run_dir))
298 for watch in self.watchdirs:
299 if not run_dir.startswith(watch):
301 if watch[-1] != os.path.sep:
303 stripped_run_dir = re.sub(watch, "", run_dir)
304 logging.debug("stripped to " + stripped_run_dir)
305 if self.notify_users is not None:
306 for u in self.notify_users:
307 self.send(u, 'Sequencing run %s finished' % \
309 if self.notify_runner is not None:
310 for r in self.notify_runner:
311 self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
315 return bot.main(args)
317 if __name__ == "__main__":
318 ret = main(sys.argv[1:])
322 # send messages to copier specifying which mount to copy