gaworkflow/runner.py

   1 #!/usr/bin/env python
   2 import logging
   3 import os
   4 import re
   5 import sys
   6 import time
   7
   8 from benderjab import rpc
   9
  10 class Runner(rpc.XmlRpcBot):
  11     """
  12     Manage running pipeline jobs.
  13     """
  14     def __init__(self, section=None, configfile=None):
  15         #if configfile is None:
  16         #    self.configfile = "~/.gaworkflow"
  17         super(Runner, self).__init__(section, configfile)
  18
  19         self.cfg['notify_users'] = None
  20
  21         self.register_function(self.sequencingFinished)
  22         self.eventTasks.append(self.update)
  23
  24     def read_config(self, section=None, configfile=None):
  25         super(Runner, self).read_config(section, configfile)
  26
  27     def _parser(self, msg, who):
  28         """
  29         Parse xmpp chat messages
  30         """
  31         help = u"I can send [start] a run, or report [status]"
  32         if re.match(u"help", msg):
  33             reply = help
  34         elif re.match("status", msg):
  35             reply = u"not implemented"
  36         elif re.match(u"start", msg):
  37             words = msg.split()
  38             if len(words) == 2:
  39                 self.sequencingFinished(words[1])
  40                 reply = u"starting run for %s" % (words[1])
  41             else:
  42                 reply = u"need runfolder name"
  43         else:
  44             reply = u"I didn't understand '%s'" %(msg)
  45         return reply
  46
  47     def start(self, daemonize):
  48         """
  49         Start application
  50         """
  51         super(Runner, self).start(daemonize)
  52
  53     def stop(self):
  54         """
  55         shutdown application
  56         """
  57         super(Runner, self).stop()
  58
  59     def sequencingFinished(self, run_dir):
  60         """
  61         Sequenceing (and copying) is finished, time to start pipeline
  62         """
  63         logging.debug("received sequencing finished message")
  64
  65     def pipelineFinished(self, run_dir):
  66         # need to strip off self.watch_dir from rundir I suspect.
  67         logging.info("pipeline finished in" + str(run_dir))
  68         #pattern = self.watch_dir
  69         #if pattern[-1] != os.path.sep:
  70         #    pattern += os.path.sep
  71         #stripped_run_dir = re.sub(pattern, "", run_dir)
  72         #logging.debug("stripped to " + stripped_run_dir)
  73         #if self.notify_users is not None:
  74         #    for u in self.notify_users:
  75         #        self.send(u, 'Sequencing run %s finished' % #(stripped_run_dir))
  76         #if self.notify_runner is not None:
  77         #    for r in self.notify_runner:
  78         #        self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
  79
  80 def main(args=None):
  81     bot = Runner()
  82     return bot.main(args)
  83
  84 if __name__ == "__main__":
  85     sys.exit(main(sys.argv[1:]))
  86