Add --use_cluster flag to CLI and pass it to the nextflow executor

psot/main.py

@@ -25,6 +25,7 @@ def main():
     analyze_parser.add_argument('--config', '-c', help='The config to use')
     analyze_parser.add_argument('--debug', '-d', action='store_true', help='Debug mode, computation directory will not be removed after computation')
     analyze_parser.add_argument('--execution_dir', '-e', help='Use the specified execution directory and do not delete it after the computation')
+    analyze_parser.add_argument('--use_cluster', '-C', action='store_true', help='Use compute cluster for execution')
     analyze_parser.set_defaults(func=analyze)
 
     args = parser.parse_args()
@@ -53,6 +54,7 @@ def cleanup(execution):
 def generate_execution(config, args):
     execution = {}
     execution['debug'] = args.debug
+    execution['use_cluster'] = args.use_cluster
     execution['mode'] = 'live' if args.live else 'complete'
     execution['bin_path'] = config['app']['bin_path']
     execution['script_path'] = os.path.abspath(os.path.join(os.path.dirname(config['app']['bin_path']), 'scripts'))
@@ -62,7 +64,10 @@ def generate_execution(config, args):
     if args.execution_dir:
         execution['directory'] = os.path.abspath(args.execution_dir)
     else:
-        execution['directory'] = tempfile.mkdtemp(dir='/vol/sge-tmp')
+        if args.use_cluster:
+            execution['directory'] = tempfile.mkdtemp(dir='/vol/sge-tmp')
+        else:
+            execution['directory'] = tempfile.mkdtemp()
     execution['modules'] = generate_execution_modules_for_profile(config, args.profile)
     return execution
     

psot/nextflow.py

@@ -17,9 +17,9 @@ def flatten(d, parent_key='', sep='_'):
 
 analysis_template = Template ('''
 process ${id} {
-    executor 'drmaa'
+    executor '${executor}'
     input:
-    file fasta from for_${id}.splitFasta(by:100, file:'input')
+    file fasta from for_${id}${chunks}
 
     output:
     file "$${fasta}.${id}.results" into ${id}_results
@@ -141,7 +141,15 @@ Channel.fromPath(params.fasta).set{fasta}''')
     fragments.append('fasta.into{'+';'.join(target_channels)+';}')
 
     for m in modules:
-        fragments.append(analysis_template.substitute(flatten(m)))
+        config = flatten(m)
+        if execution['use_cluster']:
+            config['executor'] = 'drmaa'
+            config['chunks'] = ".splitFasta(by:300, file:'input')"
+        else:
+            config['executor'] = 'local'
+            config['chunks'] = ''
+
+        fragments.append(analysis_template.substitute(config))
         if execution['mode'] == 'live':
             fragments.append(convert_live_template.substitute(flatten(m)))
             copy = deepcopy(m)