htsworkflow/automation/spoolwatcher.py

   1 #!/usr/bin/env python
   2 import logging
   3 import os
   4 import re
   5 import shlex
   6 import sys
   7 import time
   8
   9 from htsworkflow.util import mount
  10 from htsworkflow.automation.solexa import is_runfolder, get_top_dir
  11
  12 # this uses pyinotify
  13 import pyinotify
  14 from pyinotify import EventsCodes
  15 IN_CREATE = EventsCodes.ALL_FLAGS['IN_CREATE']
  16 IN_UNMOUNT = EventsCodes.ALL_FLAGS['IN_UNMOUNT']
  17
  18 from benderjab import rpc
  19
  20 class WatcherEvent(object):
  21     """
  22     Track information about a file event
  23
  24     Currently its time, and if it was an indication we've completed the run.
  25     """
  26     def __init__(self, event_root=None):
  27         self.time = time.time()
  28         self.event_root = event_root
  29         self.complete = False
  30
  31     def __unicode__(self):
  32         if self.complete:
  33            complete = "(completed)"
  34         else:
  35            complete = ""
  36         return u"<WatchEvent: %s %s %s>" % (time.ctime(self.time), self.event_root, complete)
  37
  38 class Handler(pyinotify.ProcessEvent):
  39     def __init__(self, watchmanager, bot, completion_files=None):
  40         """
  41         Completion file contains current "completion" filename
  42         """
  43         self.last_event = {}
  44         self.watchmanager = watchmanager
  45         self.bot = bot
  46         self.log = bot.log
  47         if completion_files is not None:
  48             completion_files = [ x.lower() for x in completion_files ]
  49         self.completion_files = completion_files
  50
  51     def process_IN_CREATE(self, event):
  52         for wdd in self.bot.wdds:
  53             for watch_path in self.bot.watchdirs:
  54                 run_already_complete = False
  55                 # I only care about things created inside the watch directory, so
  56                 # the event path needs to be longer than the watch path in addition to
  57                 # starting with the watch_path
  58                 if len(event.path) > len(watch_path) and event.path.startswith(watch_path):
  59                     # compute name of the top level directory that had an event
  60                     # in the current watch path
  61                     target = get_top_dir(watch_path, event.path)
  62                     runfolder = os.path.join(watch_path, target)
  63
  64                     if not is_runfolder(target):
  65                         self.log.debug("Skipping %s, not a runfolder" % (target,))
  66                         continue
  67
  68                     # grab the previous events for this watch path
  69                     watch_path_events = self.last_event.setdefault(watch_path, {})
  70
  71                     # if we've already seen an event in this directory (AKA runfolder)
  72                     # keep track if its already hit the "completed" flag
  73                     if watch_path_events.has_key(target):
  74                        run_already_complete = watch_path_events[target].complete
  75
  76                     watch_path_events[target] = WatcherEvent(target)
  77                     #self.last_event.setdefault(watch_path, {})[target] = WatcherEvent(target)
  78
  79                     msg = "Create: %s %s %s %s" % (watch_path, target, event.path, event.name)
  80
  81                     # the ReadPrep step uses some of the same file completion flags as the
  82                     # main analysis, which means this completion code might get tripped because of it
  83                     # so we need to make sure we're getting the completion file in the root of the
  84                     # runfolder
  85                     event_name = event.name.lower()
  86                     if (event_name in self.completion_files and event.path == runfolder) \
  87                       or run_already_complete:
  88                         self.last_event[watch_path][target].complete = True
  89                         msg += "(completed)"
  90
  91                     self.log.debug(msg)
  92
  93     def process_IN_DELETE(self, event):
  94         self.log.debug("Remove: %s" %  os.path.join(event.path, event.name))
  95         pass
  96
  97     def process_IN_UNMOUNT(self, event):
  98         pathname = os.path.join(event.path, event.name)
  99         self.log.debug("IN_UNMOUNT: %s" % (pathname,))
 100         self.bot.unmount_watch(event.path)
 101
 102 class SpoolWatcher(rpc.XmlRpcBot):
 103     """
 104     Watch a directory and send a message when another process is done writing.
 105
 106     This monitors a directory tree using inotify (linux specific) and
 107     after some files having been written will send a message after <timeout>
 108     seconds of no file writing.
 109
 110     (Basically when the solexa machine finishes dumping a round of data
 111     this'll hopefully send out a message saying hey look theres data available
 112
 113     """
 114     # these params need to be in the config file
 115     # I wonder where I should put the documentation
 116     #:Parameters:
 117     #    `watchdirs` - list of directories to monitor for modifications
 118     #    `profile` - specify which .htsworkflow profile to use
 119     #    `write_timeout` - how many seconds to wait for writes to finish to
 120     #                      the spool
 121     #    `notify_timeout` - how often to timeout from notify
 122     #    `completion_files` - what files indicates we've finished sequencing
 123     #                        defaults to: netcopy_complete.txt
 124
 125     def __init__(self, section=None, configfile=None):
 126         #if configfile is None:
 127         #    self.configfile = "~/.htsworkflow"
 128         super(SpoolWatcher, self).__init__(section, configfile)
 129
 130         self.cfg['watchdirs'] = None
 131         self.cfg['write_timeout'] = 10
 132         self.cfg['notify_users'] = None
 133         self.cfg['notify_runner'] = None
 134         self.cfg['completion_files'] = 'ImageAnalysis_Netcopy_complete_READ2.txt ImageAnalysis_Netcopy_complete_SINGLEREAD.txt'
 135
 136         self.watchdirs = []
 137         self.watchdir_url_map = {}
 138         self.notify_timeout = 0.001
 139
 140         self.wm = None
 141         self.notify_users = None
 142         self.notify_runner = None
 143         self.wdds = []
 144
 145         # keep track if the specified mount point is currently mounted
 146         self.mounted_points = {}
 147         # keep track of which mount points tie to which watch directories
 148         # so maybe we can remount them.
 149         self.mounts_to_watches = {}
 150
 151         self.eventTasks.append(self.process_notify)
 152
 153     def read_config(self, section=None, configfile=None):
 154         # Don't give in to the temptation to use logging functions here,
 155         # need to wait until after we detach in start
 156         super(SpoolWatcher, self).read_config(section, configfile)
 157
 158         self.watchdirs = shlex.split(self._check_required_option('watchdirs'))
 159         # see if there's an alternate url that should be used for the watchdir
 160         for watchdir in self.watchdirs:
 161             self.watchdir_url_map[watchdir] = self.cfg.get(watchdir, watchdir)
 162
 163         self.write_timeout = int(self.cfg['write_timeout'])
 164         self.completion_files = shlex.split(self.cfg['completion_files'])
 165
 166         self.notify_users = self._parse_user_list(self.cfg['notify_users'])
 167         try:
 168           self.notify_runner = \
 169              self._parse_user_list(self.cfg['notify_runner'],
 170                                    require_resource=True)
 171         except bot.JIDMissingResource:
 172             msg = 'need a full jabber ID + resource for xml-rpc destinations'
 173             raise bot.JIDMissingResource(msg)
 174
 175         self.handler = None
 176         self.notifier = None
 177
 178     def add_watch(self, watchdirs=None):
 179         """
 180         start watching watchdir or self.watchdir
 181         we're currently limited to watching one directory tree.
 182         """
 183         # create the watch managers if we need them
 184         if self.wm is None:
 185             self.wm = pyinotify.WatchManager()
 186             self.handler = Handler(self.wm, self, self.completion_files)
 187             self.notifier = pyinotify.Notifier(self.wm, self.handler)
 188
 189         # the one tree limit is mostly because self.wdd is a single item
 190         # but managing it as a list might be a bit more annoying
 191         if watchdirs is None:
 192             watchdirs = self.watchdirs
 193
 194         mask = IN_CREATE | IN_UNMOUNT
 195         # rec traverses the tree and adds all the directories that are there
 196         # at the start.
 197         # auto_add will add in new directories as they are created
 198         for w in watchdirs:
 199             mount_location = mount.find_mount_point_for(w)
 200             self.mounted_points[mount_location] = True
 201             mounts = self.mounts_to_watches.get(mount_location, [])
 202             if w not in mounts:
 203                 mounts.append(w)
 204                 self.mounts_to_watches[mount_location] = mounts
 205
 206             self.log.info(u"Watching:"+unicode(w))
 207             self.wdds.append(self.wm.add_watch(w, mask, rec=True, auto_add=True))
 208
 209     def unmount_watch(self, event_path):
 210         # remove backwards so we don't get weirdness from
 211         # the list getting shorter
 212         for i in range(len(self.wdds),0, -1):
 213             wdd = self.wdds[i]
 214             self.log.info(u'unmounting: '+unicode(wdd.items()))
 215             self.wm.rm_watch(wdd.values())
 216             del self.wdds[i]
 217         self.mounted = False
 218
 219     def make_copy_url(self, watchdir, list_event_dir):
 220         root_copy_url = self.watchdir_url_map[watchdir]
 221         if root_copy_url[-1] != '/':
 222             root_copy_url += '/'
 223         copy_url = root_copy_url + list_event_dir
 224         self.log.debug('Copy url: %s' % (copy_url,))
 225         return copy_url
 226
 227     def process_notify(self, *args):
 228         if self.notifier is None:
 229             # nothing to do yet
 230             return
 231         # process the queue of events as explained above
 232         self.notifier.process_events()
 233         #check events waits timeout
 234         if self.notifier.check_events(self.notify_timeout):
 235             # read notified events and enqeue them
 236             self.notifier.read_events()
 237             # should we do something?
 238         # has something happened?
 239         for watchdir, last_events in self.handler.last_event.items():
 240             for last_event_dir, last_event_detail in last_events.items():
 241                 time_delta = time.time() - last_event_detail.time
 242                 if time_delta > self.write_timeout:
 243                     print "timeout", unicode(last_event_detail)
 244                     copy_url = self.make_copy_url(watchdir, last_event_dir)
 245                     self.startCopy(copy_url)
 246                     if last_event_detail.complete:
 247                         self.sequencingFinished(last_event_detail.event_root)
 248
 249                     self.handler.last_event[watchdir] = {}
 250         # handle unmounted filesystems
 251         for mount_point, was_mounted in self.mounted_points.items():
 252             if not was_mounted and mount.is_mounted(mount_point):
 253                 # we've been remounted. Huzzah!
 254                 # restart the watch
 255                 for watch in self.mounts_to_watches[mount_point]:
 256                     self.add_watch(watch)
 257                     self.logg.info(
 258                         "%s was remounted, restarting watch" % \
 259                             (mount_point)
 260                     )
 261                 self.mounted_points[mount_point] = True
 262
 263     def _parser(self, msg, who):
 264         """
 265         Parse xmpp chat messages
 266         """
 267         help = u"I can send [copy] message, or squencer [finished]"
 268         if re.match(u"help", msg):
 269             reply = help
 270         elif re.match("copy", msg):
 271             self.startCopy(msg)
 272             reply = u"sent copy message"
 273         elif re.match(u"finished", msg):
 274             words = msg.split()
 275             if len(words) == 2:
 276                 self.sequencingFinished(words[1])
 277                 reply = u"sending sequencing finished for %s" % (words[1])
 278             else:
 279                 reply = u"need runfolder name"
 280         else:
 281             reply = u"I didn't understand '%s'" %(msg)
 282         return reply
 283
 284     def run(self):
 285         """
 286         Start application
 287         """
 288         # we have to configure pyinotify after BenderJab.start is called
 289         # as weird things happen to pyinotify if the stdio is closed
 290         # after it's initialized.
 291         self.add_watch()
 292         super(SpoolWatcher, self).run()
 293
 294     def stop(self):
 295         """
 296         shutdown application
 297         """
 298         # destroy the inotify's instance on this interrupt (stop monitoring)
 299         if self.notifier is not None:
 300             self.notifier.stop()
 301         super(SpoolWatcher, self).stop()
 302
 303     def startCopy(self, copy_url=None):
 304         self.log.debug("writes seem to have stopped")
 305         if self.notify_runner is not None:
 306             for r in self.notify_runner:
 307                 self.rpc_send(r, tuple([copy_url]), 'startCopy')
 308         if self.notify_users is not None:
 309             for u in self.notify_users:
 310                 self.send(u, 'startCopy %s.' % (copy_url,))
 311
 312     def sequencingFinished(self, run_dir):
 313         # need to strip off self.watchdirs from rundir I suspect.
 314         self.log.info("run.completed in " + str(run_dir))
 315         for watch in self.watchdirs:
 316             if not run_dir.startswith(watch):
 317                 print "%s didn't start with %s" % (run_dir, watch)
 318                 continue
 319             if watch[-1] != os.path.sep:
 320                 watch += os.path.sep
 321             stripped_run_dir = re.sub(watch, "", run_dir)
 322         else:
 323             stripped_run_dir = run_dir
 324
 325         self.log.debug("stripped to " + stripped_run_dir)
 326         if self.notify_users is not None:
 327             for u in self.notify_users:
 328                 self.send(u, 'Sequencing run %s finished' % \
 329                           (stripped_run_dir))
 330         if self.notify_runner is not None:
 331             for r in self.notify_runner:
 332                 self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
 333
 334 def main(args=None):
 335     bot = SpoolWatcher()
 336     return bot.main(args)
 337
 338 if __name__ == "__main__":
 339     ret = main(sys.argv[1:])
 340     #sys.exit(ret)
 341
 342 # TODO:
 343 # send messages to copier specifying which mount to copy