gaworkflow/automation/spoolwatcher.py

   1 #!/usr/bin/env python
   2 import logging
   3 import os
   4 import re
   5 import sys
   6 import time
   7 #import glob
   8
   9 #from gaworkflow.pipeline.recipe_parser import get_cycles
  10
  11 # this uses pyinotify
  12 import pyinotify
  13 from pyinotify import EventsCodes
  14
  15 from benderjab import rpc
  16
  17
  18 class WatcherEvents(object):
  19     # two events need to be tracked
  20     # one to send startCopy
  21     # one to send OMG its broken
  22     # OMG its broken needs to stop when we've seen enough
  23     #  cycles
  24     # this should be per runfolder.
  25     # read the xml files
  26     def __init__(self):
  27         pass
  28
  29
  30 class Handler(pyinotify.ProcessEvent):
  31     def __init__(self, watchmanager, bot):
  32         self.last_event_time = None
  33         self.watchmanager = watchmanager
  34         self.bot = bot
  35
  36     def process_IN_CREATE(self, event):
  37         self.last_event_time = time.time()
  38         msg = "Create: %s" %  os.path.join(event.path, event.name)
  39         if event.name.lower() == "run.completed":
  40             try:
  41                 self.bot.sequencingFinished(event.path)
  42             except IOError, e:
  43                 logging.error("Couldn't send sequencingFinished")
  44         logging.debug(msg)
  45
  46     def process_IN_DELETE(self, event):
  47         logging.debug("Remove: %s" %  os.path.join(event.path, event.name))
  48
  49     def process_IN_UNMOUNT(self, event):
  50         self.bot.unmount_watch()
  51
  52 class SpoolWatcher(rpc.XmlRpcBot):
  53     """
  54     Watch a directory and send a message when another process is done writing.
  55
  56     This monitors a directory tree using inotify (linux specific) and
  57     after some files having been written will send a message after <timeout>
  58     seconds of no file writing.
  59
  60     (Basically when the solexa machine finishes dumping a round of data
  61     this'll hopefully send out a message saying hey look theres data available
  62
  63     """
  64     # these params need to be in the config file
  65     # I wonder where I should put the documentation
  66     #:Parameters:
  67     #    `watchdir` - which directory tree to monitor for modifications
  68     #    `profile` - specify which .gaworkflow profile to use
  69     #    `write_timeout` - how many seconds to wait for writes to finish to
  70     #                      the spool
  71     #    `notify_timeout` - how often to timeout from notify
  72
  73     def __init__(self, section=None, configfile=None):
  74         #if configfile is None:
  75         #    self.configfile = "~/.gaworkflow"
  76         super(SpoolWatcher, self).__init__(section, configfile)
  77
  78         self.cfg['watchdir'] = None
  79         self.cfg['write_timeout'] = 10
  80         self.cfg['notify_users'] = None
  81         self.cfg['notify_runner'] = None
  82
  83         self.notify_timeout = 0.001
  84         self.wm = pyinotify.WatchManager()
  85         self.handler = Handler(self.wm, self)
  86         self.notifier = pyinotify.Notifier(self.wm, self.handler)
  87         self.wdd = None
  88
  89         self.notify_users = None
  90         self.notify_runner = None
  91
  92         self.eventTasks.append(self.process_notify)
  93
  94     def read_config(self, section=None, configfile=None):
  95         super(SpoolWatcher, self).read_config(section, configfile)
  96
  97         self.watch_dir = self._check_required_option('watchdir')
  98         self.write_timeout = int(self.cfg['write_timeout'])
  99
 100         self.notify_users = self._parse_user_list(self.cfg['notify_users'])
 101         try:
 102           self.notify_runner = \
 103              self._parse_user_list(self.cfg['notify_runner'],
 104                                    require_resource=True)
 105         except bot.JIDMissingResource:
 106             msg = 'need a full jabber ID + resource for xml-rpc destinations'
 107             logging.FATAL(msg)
 108             raise bot.JIDMissingResource(msg)
 109
 110     def add_watch(self, watchdir=None):
 111         """
 112         start watching watchdir or self.watch_dir
 113         we're currently limited to watching one directory tree.
 114         """
 115         # the one tree limit is mostly because self.wdd is a single item
 116         # but managing it as a list might be a bit more annoying
 117         if watchdir is None:
 118             watchdir = self.watch_dir
 119         logging.info("Watching:"+str(watchdir))
 120         mask = EventsCodes.IN_CREATE | EventsCodes.IN_UNMOUNT
 121         # rec traverses the tree and adds all the directories that are there
 122         # at the start.
 123         # auto_add will add in new directories as they are created
 124         self.wdd = self.wm.add_watch(watchdir, mask, rec=True, auto_add=True)
 125
 126     def unmount_watch(self):
 127         if self.wdd is not None:
 128             logging.debug("disabling watch")
 129             logging.debug(str(self.wdd))
 130             self.wm.rm_watch(self.wdd.values())
 131             self.wdd = None
 132
 133     def process_notify(self, *args):
 134         # process the queue of events as explained above
 135         self.notifier.process_events()
 136         #check events waits timeout
 137         if self.notifier.check_events(self.notify_timeout):
 138             # read notified events and enqeue them
 139             self.notifier.read_events()
 140             # should we do something?
 141         last_event_time = self.handler.last_event_time
 142         if last_event_time is not None:
 143             time_delta = time.time() - last_event_time
 144             if time_delta > self.write_timeout:
 145                 self.startCopy()
 146                 self.handler.last_event_time = None
 147
 148     def _parser(self, msg, who):
 149         """
 150         Parse xmpp chat messages
 151         """
 152         help = u"I can send [copy] message, or squencer [finished]"
 153         if re.match(u"help", msg):
 154             reply = help
 155         elif re.match("copy", msg):
 156             self.startCopy()
 157             reply = u"sent copy message"
 158         elif re.match(u"finished", msg):
 159             words = msg.split()
 160             if len(words) == 2:
 161                 self.sequencingFinished(words[1])
 162                 reply = u"sending sequencing finished for %s" % (words[1])
 163             else:
 164                 reply = u"need runfolder name"
 165         else:
 166             reply = u"I didn't understand '%s'" %(msg)
 167         return reply
 168
 169     def start(self, daemonize):
 170         """
 171         Start application
 172         """
 173         self.add_watch()
 174         super(SpoolWatcher, self).start(daemonize)
 175
 176     def stop(self):
 177         """
 178         shutdown application
 179         """
 180         # destroy the inotify's instance on this interrupt (stop monitoring)
 181         self.notifier.stop()
 182         super(SpoolWatcher, self).stop()
 183
 184     def startCopy(self):
 185         logging.debug("writes seem to have stopped")
 186         if self.notify_runner is not None:
 187             for r in self.notify_runner:
 188                 self.rpc_send(r, tuple(), 'startCopy')
 189
 190     def sequencingFinished(self, run_dir):
 191         # need to strip off self.watch_dir from rundir I suspect.
 192         logging.info("run.completed in " + str(run_dir))
 193         pattern = self.watch_dir
 194         if pattern[-1] != os.path.sep:
 195             pattern += os.path.sep
 196         stripped_run_dir = re.sub(pattern, "", run_dir)
 197         logging.debug("stripped to " + stripped_run_dir)
 198         if self.notify_users is not None:
 199             for u in self.notify_users:
 200                 self.send(u, 'Sequencing run %s finished' % (stripped_run_dir))
 201         if self.notify_runner is not None:
 202             for r in self.notify_runner:
 203                 self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
 204
 205 def main(args=None):
 206     bot = SpoolWatcher()
 207     return bot.main(args)
 208
 209 if __name__ == "__main__":
 210     sys.exit(main(sys.argv[1:]))
 211