merged in master

bethac07 · Nov 5, 2020 · May 13, 2020 · May 13, 2020 · May 13, 2020 · May 15, 2020
commit e680795e912747f1ec081f91d458c9b7c9efe97c
diff --git a/config.py b/config.py
@@ -37,3 +37,6 @@
 CHECK_IF_DONE_BOOL = 'True'  #True or False- should it check if there are a certain number of non-empty files and delete the job if yes?
 EXPECTED_NUMBER_FILES = 7    #What is the number of files that trigger skipping a job?
 MIN_FILE_SIZE_BYTES = 1      #What is the minimal number of bytes an object should be to "count"?
+
+# PLUGINS
+USE_PLUGINS = 'True'
diff --git a/run_batch_general.py b/run_batch_general.py
@@ -0,0 +1,155 @@
+import json
+import boto3
+import string
+import os
+import posixpath
+class JobQueue():
+
+    def __init__(self,name=None):
+        self.sqs = boto3.resource('sqs')
+        self.queue = self.sqs.get_queue_by_name(QueueName=name+'Queue')
+        self.inProcess = -1
+        self.pending = -1
+
+    def scheduleBatch(self, data):
+        msg = json.dumps(data)
+        response = self.queue.send_message(MessageBody=msg)
+        print 'Batch sent. Message ID:',response.get('MessageId')
+
+#project specific stuff
+topdirname='' #PROJECTNAME        
+projectname='' #PROJECTNAME
+batchsuffix='' #BATCHNAME
+rows=list(string.ascii_uppercase)[0:8]
+columns=range(1,13)
+sites=range(1,18)
+platelist=[] # PLATEFOLDERNAMES
+illumpipename='illum.cppipe'
+qcpipename='qc.cppipe'
+analysispipename='analysis.cppipe'
+batchpipenameillum='Batch_data_illum.h5'
+batchpipename='Batch_data.h5'
+
+#not project specific, unless you deviate from the structure
+startpath=posixpath.join('projects',topdirname)
+pipelinepath=posixpath.join(startpath,os.path.join('workspace/pipelines',batchsuffix))
+illumoutpath=posixpath.join(startpath,os.path.join(batchsuffix,'illum'))
+QCoutpath=posixpath.join(startpath,os.path.join('workspace/qc',os.path.join(batchsuffix,'results')))
+analysisoutpath=posixpath.join(startpath,os.path.join('workspace/analysis',batchsuffix))
+inputpath=posixpath.join(startpath,os.path.join('workspace/qc',os.path.join(batchsuffix,'rules')))
+datafilepath=posixpath.join(startpath,os.path.join('workspace/load_data_csv',batchsuffix))
+anlysisoutputstructure="Metadata_Plate/analysis/Metadata_Plate-Metadata_Well-Metadata_Site"
+batchpath=posixpath.join(startpath,os.path.join('workspace/batchfiles',batchsuffix))
+
+def MakeIllumJobs(mode='repurp'):    
+    illumqueue = JobQueue(projectname+'_Illum')
+    for toillum in platelist:
+        if mode=='repurp':
+            templateMessage_illum = {'Metadata': 'Metadata_Plate='+toillum,
+                                     'pipeline': posixpath.join(pipelinepath,illumpipename),'output': illumoutpath,
+                                     'input': inputpath, 'data_file':posixpath.join(datafilepath,toillum+'.csv')}
+        elif mode=='batch':
+            templateMessage_illum = {'Metadata': 'Metadata_Plate='+toillum,
+                                        'pipeline': posixpath.join(batchpath,batchpipenameillum),
+                                        'output': illumoutpath,
+                                        'input':inputpath,
+                                        'data_file': posixpath.join(batchpath,batchpipenameillum)
+                                        }
+        else:
+            templateMessage_illum = {'Metadata': 'Metadata_Plate='+toillum,
+                                     'pipeline': posixpath.join(pipelinepath,illumpipename),'output': illumoutpath,
+                                     'input': inputpath, 'data_file':posixpath.join(datafilepath,untruncatedplatedict[toillum],'load_data.csv')}
+
+        illumqueue.scheduleBatch(templateMessage_illum)
+
+    print 'Illum job submitted. Check your queue'
+
+def MakeQCJobs(repurp=False):
+    qcqueue = JobQueue(projectname+'_QC')
+    for toqc in platelist:
+        for eachrow in rows:
+            for eachcol in columns:
+                if repurp==False:
+                    templateMessage_qc = {'Metadata': 'Metadata_Plate='+toqc+',Metadata_Well='+eachrow+'%02d' %eachcol,
+                                    'pipeline': posixpath.join(pipelinepath,qcpipename),
+                                    'output': QCoutpath,
+                                    'input': inputpath,
+                                    'data_file': posixpath.join(datafilepath,toqc+'.csv')
+                                }
+                else:
+                    templateMessage_qc = {'Metadata': 'Metadata_Plate='+toqc+',Metadata_Well='+eachrow+'%02d' %eachcol,
+                                    'pipeline': posixpath.join(pipelinepath,qcpipename),
+                                    'output': QCoutpath,
+                                    'input': inputpath,
+                                    'data_file': posixpath.join(datafilepath,untruncatedplatedict[toqc],'load_data.csv')
+                                }
+                qcqueue.scheduleBatch(templateMessage_qc)
+
+    print 'QC job submitted. Check your queue'
+
+def MakeQCJobs_persite(repurp=False):
+    qcqueue = JobQueue(projectname+'_QC')
+    for toqc in platelist:
+        for eachrow in rows:
+            for eachcol in columns:
+                for eachsite in sites:
+                    if repurp==False:
+                        templateMessage_qc = {'Metadata': 'Metadata_Plate='+toqc+',Metadata_Well='+eachrow+'%02d' %eachcol+',Metadata_Site='+str(eachsite),
+                                        'pipeline': posixpath.join(pipelinepath,qcpipename),
+                                        'output': QCoutpath,
+                                        'input': inputpath,
+                                        'data_file': posixpath.join(datafilepath,toqc+'.csv')
+                                        }
+                    else:
+                        templateMessage_qc = {'Metadata': 'Metadata_Plate='+toqc+',Metadata_Well='+eachrow+'%02d' %eachcol+',Metadata_Site='+str(eachsite),
+                                        'pipeline': posixpath.join(pipelinepath,qcpipename),
+                                        'output': QCoutpath,
+                                        'input': inputpath,
+                                        'data_file': posixpath.join(datafilepath,untruncatedplatedict[toqc],'load_data.csv')
+                                        }
+
+                    qcqueue.scheduleBatch(templateMessage_qc)
+
+    print 'QC job submitted. Check your queue'
+
+
+def MakeAnalysisJobs(mode='repurp'):
+    analysisqueue = JobQueue(projectname+'_Analysis')
+    for toanalyze in platelist:
+        for eachrow in rows:
+            for eachcol in columns:
+                for eachsite in sites:
+                    if mode=='repurp':
+                        templateMessage_analysis = {'Metadata': 'Metadata_Plate='+toanalyze+',Metadata_Well='+eachrow+'%02d' %eachcol+',Metadata_Site='+str(eachsite),
+                                        'pipeline': posixpath.join(pipelinepath,analysispipename),
+                                        'output': analysisoutpath,
+                                        'output_structure':anlysisoutputstructure,
+                                        'input':inputpath,
+                                        'data_file': posixpath.join(datafilepath,toanalyze+'_with_illum.csv')
+                                        }
+		    elif mode=='batch':
+                        templateMessage_analysis = {'Metadata': 'Metadata_Plate='+toanalyze+',Metadata_Well='+eachrow+'%02d' %eachcol+',Metadata_Site='+str(eachsite),
+                                        'pipeline': posixpath.join(batchpath,batchpipename),
+                                        'output': analysisoutpath,
+                                        'output_structure':anlysisoutputstructure,
+                                        'input':inputpath,
+                                        'data_file': posixpath.join(batchpath,batchpipename)
+                                        }
+                    else:
+                        templateMessage_analysis = {'Metadata': 'Metadata_Plate='+toanalyze+',Metadata_Well='+eachrow+'%02d' %eachcol+',Metadata_Site='+str(eachsite),
+                                        'pipeline': posixpath.join(pipelinepath,analysispipename),
+                                        'output': analysisoutpath,
+                                        'output_structure':anlysisoutputstructure,
+                                        'input':inputpath,
+                                        'data_file': posixpath.join(datafilepath,untruncatedplatedict[toanalyze],'load_data_with_illum.csv')
+                                        }
+
+                    analysisqueue.scheduleBatch(templateMessage_analysis)
+
+    print 'Analysis job submitted. Check your queue'
+
+#MakeIllumJobs(mode='batch')
+#MakeQCJobs(repurp=True)
+#MakeQCJobs_persite(repurp=True)
+#MakeAnalysisJobs(mode='batch')
+
diff --git a/worker/Dockerfile b/worker/Dockerfile
@@ -5,6 +5,7 @@
 # This wraps the cellprofiler docker registry
 #
 
+
 FROM cellprofiler/cellprofiler:3.1.9
 
 # Install S3FS 
@@ -56,6 +57,12 @@ COPY instance-monitor.py .
 COPY run-worker.sh .
 RUN chmod 755 run-worker.sh
 
+RUN git clone https://github.com/CellProfiler/CellProfiler-plugins.git
+WORKDIR /home/ubuntu/CellProfiler-plugins
+RUN \
+    pip install -r requirements.txt
+
+WORKDIR /home/ubuntu
 ENTRYPOINT ["./run-worker.sh"]
 CMD [""]
 
diff --git a/worker/cp-worker.py b/worker/cp-worker.py
@@ -17,6 +17,7 @@
 
 DATA_ROOT = '/home/ubuntu/bucket'
 LOCAL_OUTPUT = '/home/ubuntu/local_output'
+PLUGIN_DIR = '/home/ubuntu/CellProfiler-plugins'
 QUEUE_URL = os.environ['SQS_QUEUE_URL']
 AWS_BUCKET = os.environ['AWS_BUCKET']
 LOG_GROUP_NAME= os.environ['LOG_GROUP_NAME']
@@ -26,6 +27,10 @@
     MIN_FILE_SIZE_BYTES = 1
 else:
     MIN_FILE_SIZE_BYTES = int(os.environ['MIN_FILE_SIZE_BYTES'])
+if 'USE_PLUGINS' not in os.environ:
+    USE_PLUGINS = 'False'
+else:
+    USE_PLUGINS = os.environ['USE_PLUGINS']
 
 #################################
 # CLASS TO HANDLE THE SQS QUEUE
@@ -127,18 +132,19 @@ def runCellProfiler(message):
     remoteOut= os.path.join(message['output'],metadataID)
     replaceValues = {'PL':message['pipeline'], 'OUT':localOut, 'FL':message['data_file'],
             'DATA': DATA_ROOT, 'Metadata': message['Metadata'], 'IN': message['input'], 
-            'MetadataID':metadataID }
+            'MetadataID':metadataID, 'PLUGINS':PLUGIN_DIR }
     # See if this is a message you've already handled, if you've so chosen
     if CHECK_IF_DONE_BOOL.upper() == 'TRUE':
         try:
             s3client=boto3.client('s3')
             bucketlist=s3client.list_objects(Bucket=AWS_BUCKET,Prefix=remoteOut+'/')
             objectsizelist=[k['Size'] for k in bucketlist['Contents']]
             objectsizelist = [i for i in objectsizelist if i >= MIN_FILE_SIZE_BYTES]
-            if len(objectsizelist)>=int(EXPECTED_NUMBER_FILES):
-                return 'SUCCESS'
-        except KeyError: #Returned if that folder does not exist
-            pass
+        if len(objectsizelist)>=int(EXPECTED_NUMBER_FILES):
+            logger.removeHandler(watchtowerlogger)
+            return 'SUCCESS'
+    except KeyError: #Returned if that folder does not exist
+        pass
 
     # Start loggging now that we have a job we care about
     watchtowerlogger=watchtower.CloudWatchLogHandler(log_group=LOG_GROUP_NAME, stream_name=metadataID,create_log_group=False)