pathwayProcessor.py

import json
import math
import os
import rpy2.robjects as robjects
import smtplib
import time
import logging

from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

from twisted.internet import reactor, defer
from PropertyUtil import PropertyUtil

from stompest.async import Stomp
from stompest.async.listener import SubscriptionListener
from stompest.async.listener import DisconnectListener
from stompest.config import StompConfig
from stompest.protocol import StompSpec

class pathwayProcessor(DisconnectListener):
  CONFIG = 'queue.config'
  NAME = 'queue.name'
  URL = 'queue.url'

  UPLOAD_FOLDER = 'pathway.folder.upload'
  MAIL_HOST = 'mail.host'
  MAIL_ADMIN = 'mail.admin'

  def composeMail(self,recipients,message,files=[]):
    config = PropertyUtil(r"config.ini")
    logging.info("sending message")
    if not isinstance(recipients,list):
      recipients = [recipients]
    packet = MIMEMultipart()
    packet['Subject'] = "Subject: Pathway Analysis Results"
    packet['From'] = "Pathway Analysis Tool <do.not.reply@nih.gov>"
    packet['To'] = ", ".join(recipients)
    logging.info(recipients)
    # print message
    packet.attach(MIMEText(message))
    for file in files:
      with open(file,"rb") as openfile:
        packet.attach(MIMEApplication(
          openfile.read(),
          Content_Disposition='attachment; filename="%s"' % os.path.basename(file),
          Name=os.path.basename(file)
        ))
    MAIL_HOST=config.getAsString('mail.host')
    logging.info(MAIL_HOST)
    smtp = smtplib.SMTP(MAIL_HOST)
    smtp.sendmail("do.not.reply@nih.gov",recipients,packet.as_string())

  def testQueue(self):
    logging.debug("tested")

  def rLength(self, tested):
    if tested is None:
      return 0
    if isinstance(tested,list) or isinstance(tested,set):
      return len(tested)
    else:
      return 1

  # @This is teh consume code which will listen to Queue server.
  def consume(self, client, frame):
    logging.info("In consume")
    config = PropertyUtil(r"config.ini")
    starttime = str(time.time())
    parameters = json.loads(frame.body)
    timestamp = frame.headers['timestamp']
    outfileString = os.path.join(parameters['outdir'], str(parameters['idstr']) + '.json')

    jsonout = {
      'submittedTime': parameters['idstr'], 
      'queuedTime':timestamp,
      'payload':parameters,
      'processStartTime': starttime
    }

    with open(outfileString,'w') as outfile:
      json.dump(jsonout,outfile)
    try:
      # Run R-Script
      artpResult = json.loads(self.r_runARTP(json.dumps(parameters))[0])
    except Exception as e:
      artpResult = {}
      artpResult["error"] = str(e)
    jsonout["processStopTime"] = str(time.time())
    message = ""
    if "warnings" in artpResult:
      jsonout["warnings"] = artpResult["warnings"]
      with open(outfileString,'w') as outfile:
        json.dump(jsonout,outfile)
      message += "\nWarnings:\n"
      if (isinstance(artpResult["warnings"],list)):
        for warning in artpResult["warnings"]:
          message += warning.strip() + "\n\n"
      else:
        message += artpResult["warnings"].strip() + "\n\n"
    if "error" in artpResult:
      jsonout["error"] = artpResult["error"]
      jsonout["status"] = "error"
      with open(outfileString,'w') as outfile:
        json.dump(jsonout,outfile)
      message = "Error: " + artpResult["error"].strip() + "\n" + message + "\n\n" +frame.body
      print(message)
      self.composeMail(config.getAsString(pathwayProcessor.MAIL_ADMIN).split(","), message)
      self.composeMail(parameters["email"], "Unfortunately there was an error processing your request. The site administrators have been alerted to the problem. Please contact " + config.getAsString(pathwayProcessor.MAIL_ADMIN) + " if any question.\n\n" + message)
      return
    # email results
    files = [ os.path.join(parameters['outdir'],parameters['idstr']+'.Rdata') ]
    saveValue = artpResult["saveValue"]
    jsonout["saveValue"] = saveValue
    jsonout["status"] = "success"

    with open(outfileString,'w') as outfile:
      json.dump(jsonout,outfile)
      
    message = ("Dear User,\n\n" +
              "We have analyzed your data using the ARTP2 package (version: " + ".".join([str(x) for x in saveValue['options']['version'][0]]) + "). " +
              "The sARTP test returned a pathway p-value " + str(round(saveValue['pathway.pvalue'],1-int(math.floor(math.log10(saveValue['pathway.pvalue']))))) + ". The p-value was estimated by " + str(saveValue['options']['nperm']) + " resampling steps.\n\n" +
              "Several gene/SNP filters were applied to the data based on specified options. " +
              "There are " + str(self.rLength(saveValue['deleted.genes'].get('Gene',None))) + " genes and " + str(self.rLength(set(saveValue['deleted.snps'].get('SNP',None)))) + " SNPs that were excluded from the analysis. " +
              "After that, " + str(self.rLength(saveValue['gene.pvalue']['Chr'])) + " unique genes and " + str(self.rLength(set(saveValue['pathway']['SNP']))) + " unique SNPs were used in testing. " +
              "Some warning messages (if any) can be found at the end of this email.\n\n" +
              "More detailed result of this pathway analysis is saved as an R list, saveValue, in the attached file. You can read it in R through function load(). For example, the pathway p-value is\n\n" +
              "saveValue$pathway.pvalue\n\n" +
              "and the gene-level p-values are saved in a data frame\n\n" +
              "saveValue$gene.pvalue\n\n" +
              "To understand the reason for each SNP/gene excluded from the analysis, check\n\n" +
              "saveValue$deleted.snps\n\n" +
              "and\n\n" +
              "saveValue$deleted.genes\n\n" +
              "A full list of genes and SNPs that were used in testing is saved in a data frame\n\n" +
              "saveValue$pathway\n\n" +
              "All the options are saved in \n\n" +
              "saveValue$options\n\n" +
              "For more information, please refer to the help document of function sARTP in R package ARTP2.\n\n" +
              message)

    print(message)
    self.composeMail(parameters['email'], message, files)

    # remove the already used files
    for study in parameters['studies']:
      os.remove(study['filename'])
    if (parameters['pathway_type'] == 'file_pathway'):
      os.remove(parameters['file_pathway'])
    for file in files:
      os.remove(file)

  @defer.inlineCallbacks
  def run(self):
    # client = yield Stomp(self.CONFIG[pathwayProcessor.CONFIG]).connect()
    client = Stomp(self.config)
    yield client.connect()
    headers = {
        # client-individual mode is necessary for concurrent processing
        # (requires ActiveMQ >= 5.2)
        StompSpec.ACK_HEADER: StompSpec.ACK_CLIENT_INDIVIDUAL,
        # the maximal number of messages the broker will let you work on at the same time
        'activemq.prefetchSize': '100',
    }
    # client.subscribe(self.CONFIG[pathwayProcessor.NAME], headers, listener=self)
    client.subscribe(self.QUEUE, headers, listener=SubscriptionListener(self.consume, errorDestination=self.ERROR_QUEUE))
    client.add(listener=self)

  # Consumer for Jobs in Queue, needs to be rewrite by the individual projects

  def onCleanup(self, connect):
    logging.info('In clean up ...')

  def onConnectionLost(self,connection,reason):
    # super(pathwayProcessor,self).onConnectionLost(connection,reason)
    self.run()

  def __init__(self):
    # super(pathwayProcessor,self).__init__(self.consume)
    config = PropertyUtil(r"config.ini")
    # config[pathwayProcessor.CONFIG] = StompConfig(uri="failover:("+config.getAsString(pathwayProcessor.URL)+")?startupMaxReconnectAttempts=-1,initialReconnectDelay=300000")
    self.QUEUE=config.getAsString(pathwayProcessor.NAME)
    self.ERROR_QUEUE=config.getAsString('queue.error.name')
    # config[pathwayProcessor.CONFIG] = StompConfig(config.getAsString(pathwayProcessor.URL))
    config = StompConfig(config.getAsString(pathwayProcessor.URL))
    self.config = config

    robjects.r('''source('ARTPWrapper.R')''')
    self.r_runARTP = robjects.r['runARTPWithHandlers']


# def main():
if __name__ == '__main__':
  logging.basicConfig(level=logging.DEBUG, filename='../logs/queue.log', filemode='w')
  logging.info("JPSurv processor has started")
  pathwayProcessor().run()
  reactor.run()