gaworkflow/runner.py

   1 #!/usr/bin/env python
   2 import logging
   3 import os
   4 import re
   5 import sys
   6 import time
   7 import threading
   8
   9 from benderjab import rpc
  10
  11 from gaworkflow.pipeline.configure_run import *
  12 from gaworkflow.pipeline.monitors import startCmdLineStatusMonitor
  13
  14 #s_fc = re.compile('FC[0-9]+')
  15 s_fc = re.compile('_[0-9a-zA-Z]*$')
  16
  17
  18 def _get_flowcell_from_rundir(run_dir):
  19     """
  20     Returns flowcell string based on run_dir.
  21     Returns None and logs error if flowcell can't be found.
  22     """
  23     junk, dirname = os.path.split(run_dir)
  24     mo = s_fc.search(dirname)
  25     if not mo:
  26         logging.error('RunDir 2 FlowCell error: %s' % (run_dir))
  27         return None
  28
  29     return dirname[mo.start()+1:]
  30
  31
  32
  33 class Runner(rpc.XmlRpcBot):
  34     """
  35     Manage running pipeline jobs.
  36     """
  37     def __init__(self, section=None, configfile=None):
  38         #if configfile is None:
  39         #    self.configfile = "~/.gaworkflow"
  40         super(Runner, self).__init__(section, configfile)
  41
  42         self.cfg['notify_users'] = None
  43         self.cfg['genome_dir'] = None
  44
  45         self.conf_info_dict = {}
  46
  47         self.register_function(self.sequencingFinished)
  48         #self.eventTasks.append(self.update)
  49
  50
  51     def read_config(self, section=None, configfile=None):
  52         super(Runner, self).read_config(section, configfile)
  53
  54         self.genome_dir = self._check_required_option('genome_dir')
  55
  56
  57     def _parser(self, msg, who):
  58         """
  59         Parse xmpp chat messages
  60         """
  61         help = u"I can send [start] a run, or report [status]"
  62         if re.match(u"help", msg):
  63             reply = help
  64         elif re.match("status", msg):
  65             reply = u"not implemented"
  66         elif re.match(u"start", msg):
  67             words = msg.split()
  68             if len(words) == 2:
  69                 self.sequencingFinished(words[1])
  70                 reply = u"starting run for %s" % (words[1])
  71             else:
  72                 reply = u"need runfolder name"
  73         else:
  74             reply = u"I didn't understand '%s'" %(msg)
  75
  76         logging.debug("reply: " + str(reply))
  77         return reply
  78
  79
  80     def start(self, daemonize):
  81         """
  82         Start application
  83         """
  84         super(Runner, self).start(daemonize)
  85
  86
  87     def stop(self):
  88         """
  89         shutdown application
  90         """
  91         super(Runner, self).stop()
  92
  93
  94     def sequencingFinished(self, run_dir):
  95         """
  96         Sequenceing (and copying) is finished, time to start pipeline
  97         """
  98         logging.debug("received sequencing finished message")
  99
 100         # Setup config info object
 101         ci = ConfigInfo()
 102         ci.run_path = run_dir
 103
 104         # get flowcell from run_dir name
 105         flowcell = _get_flowcell_from_rundir(run_dir)
 106
 107         # Store ci object in dictionary
 108         self.conf_info_dict[flowcell] = ci
 109
 110
 111         # Launch the job in it's own thread and turn.
 112         self.launchJob(run_dir, flowcell, ci)
 113
 114
 115     def pipelineFinished(self, run_dir):
 116         # need to strip off self.watch_dir from rundir I suspect.
 117         logging.info("pipeline finished in" + str(run_dir))
 118         #pattern = self.watch_dir
 119         #if pattern[-1] != os.path.sep:
 120         #    pattern += os.path.sep
 121         #stripped_run_dir = re.sub(pattern, "", run_dir)
 122         #logging.debug("stripped to " + stripped_run_dir)
 123         #if self.notify_users is not None:
 124         #    for u in self.notify_users:
 125         #        self.send(u, 'Sequencing run %s finished' % #(stripped_run_dir))
 126         #if self.notify_runner is not None:
 127         #    for r in self.notify_runner:
 128         #        self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
 129
 130
 131     def _runner(self, run_dir, flowcell, conf_info):
 132         # retrieve config step
 133         cfg_filepath = os.path.abspath('config32auto.txt')
 134         status_retrieve_cfg = retrieve_config(conf_info,
 135                                           flowcell,
 136                                           cfg_filepath,
 137                                           self.genome_dir)
 138         if status_retrieve_cfg:
 139             logging.info("Runner: Retrieve config: success")
 140         else:
 141             logging.error("Runner: Retrieve config: failed")
 142
 143
 144         # configure step
 145         if status_retrieve_cfg:
 146             status = configure(ci)
 147             if status:
 148                 logging.info("Runner: Configure: success")
 149             else:
 150                 logging.error("Runner: Configure: failed")
 151
 152             #if successful, continue
 153             if status:
 154                 # Setup status cmdline status monitor
 155                 #startCmdLineStatusMonitor(ci)
 156
 157                 # running step
 158                 print 'Running pipeline now!'
 159                 run_status = run_pipeline(ci)
 160                 if run_status is True:
 161                     logging.info('Runner: Pipeline: success')
 162                     self.piplineFinished(run_dir)
 163                 else:
 164                     logging.info('Runner: Pipeline: failed')
 165
 166
 167     def launchJob(self, run_dir, flowcell, conf_info):
 168         """
 169         Starts up a thread for running the pipeline
 170         """
 171         t = threading.Thread(target=self._runner,
 172                         args=[run_dir, flowcell, conf_info])
 173         t.setDaemon(True)
 174         t.start()
 175
 176
 177
 178 def main(args=None):
 179     bot = Runner()
 180     bot.cfg['loglevel'] = 'DEBUG'
 181     return bot.main(args)
 182
 183 if __name__ == "__main__":
 184     sys.exit(main(sys.argv[1:]))
 185