Refactored nextflow code generation and execution into module. Implemented postprocessing

bin/PSOT

@@ -3,6 +3,7 @@ import argparse
 import os
 from config import load_config
 import copy
+from nextflow import setup_execution_directory, execute_analysis
 
 def main():
     parser = argparse.ArgumentParser(description='Make bioinformatic observations on aminoacid sequences')
@@ -44,24 +45,6 @@ def generate_execution(config, args):
     modules = generate_execution_modules_for_profile(config, args.profile)
     execution['modules'] = modules
     return execution
-
-def setup_execution_directory(execution):
-    directory = execution['directory']
-    if not os.path.exists(directory):
-        os.mkdir(directory)
-    if not os.path.isdir(directory):
-        exit()
-
-    nextflow_script = generate_nextflow_script(execution['modules'])
-    with open(directory + '/main.nf', 'w') as script_file:
-        script_file.write(nextflow_script)
-
-    if not os.path.exists(directory+'/bin'):
-        os.symlink(execution['bin_path'], directory + '/bin')
-
-def execute_analysis(execution):
-    os.chdir(execution['directory'])
-    os.system('nextflow run main.nf --fasta ' + execution['fasta'] + ' --output ' + execution['output'])
     
 def generate_execution_modules_for_profile(config, profile):
     # find profile by name
@@ -87,50 +70,6 @@ def generate_params_string(options):
         params = ' '.join(l)
     return params
 
-def generate_nextflow_script(modules):
-
-    fragments = []
-    fragments.append('''params.fasta = "example/proteins.fas"
-Channel.fromPath(params.fasta).set{fasta}''')
-    target_channels = ["for_"+m['id'] for m in modules]
-    fragments.append('fasta.into{'+';'.join(target_channels)+';}')
-
-    from string import Template
-    analysis_template = Template ('''
-process ${id} {
-    input:
-    file fasta from for_${id}
-
-    output:
-    file "$${fasta}.${id}.results" into ${id}_results
-
-    script:
-    """
-    ${analysis_script} --fasta $$fasta --output $${fasta}.${id}.results ${params}
-    """
-}
-''')
-    convert_template = Template ('''
-process convert_${id}_to_json {
-    input:
-    file result from ${id}_results
-
-    output:
-    file "$${result}.json" into ${id}_json
-
-    script:
-    """
-    ${converter_script} --result $$result --output $${result}.json
-    """
-}
-''')
-    for m in modules:
-        fragments.append(analysis_template.substitute(m))
-        fragments.append(convert_template.substitute(m))
-
-    # TODO add postprocessing
-    nextflow_script = '\n'.join(fragments)
-    return nextflow_script
 
 def show_analyses(config):
     print('Profiles:')

bin/nextflow.py

+from string import Template
+import os.path
+import os
+
+analysis_template = Template ('''
+process ${id} {
+    input:
+    file fasta from for_${id}
+
+    output:
+    file "$${fasta}.${id}.results" into ${id}_results
+
+    script:
+    """
+    ${analysis_script} --fasta $$fasta --output $${fasta}.${id}.results ${params}
+    """
+}
+''')
+convert_template = Template ('''
+process convert_${id}_to_json {
+    input:
+    file result from ${id}_results
+
+    output:
+    file "$${result}.json" into ${id}_json
+
+    script:
+    """
+    ${converter_script} --result $$result --output $${result}.json
+    """
+}
+''')
+input_template = Template('''    file ${id}_result from ${id}_json''')
+join_jsons_template = Template('''
+process join_documents {
+
+    input:
+    ${inputs}
+
+    output:
+    file "joined.json" into joined_json
+
+    script:
+    """
+    join_json_files.py --output joined.json *.json
+    """
+}
+''')
+split_jsons_template = Template('''
+process split_documents {
+
+    publishDir "${output}", mode: 'copy'
+
+    input:
+    file "input/json.json" from joined_json
+
+    output:
+    file "*.json" into result_documents
+
+    script:
+    """
+    split_json_into_separate_files.py --json 'input/json.json' --output .
+    """
+}
+''')
+
+def setup_execution_directory(execution):
+    directory = execution['directory']
+    if not os.path.exists(directory):
+        os.mkdir(directory)
+    if not os.path.isdir(directory):
+        exit()
+
+    nextflow_script = generate_nextflow_script(execution)
+    with open(directory + '/main.nf', 'w') as script_file:
+        script_file.write(nextflow_script)
+
+    if not os.path.exists(directory+'/bin'):
+        os.symlink(execution['bin_path'], directory + '/bin')
+
+def execute_analysis(execution):
+    os.chdir(execution['directory'])
+    os.system('nextflow run main.nf --fasta ' + execution['fasta'] + ' --output ' + execution['output'])
+
+def generate_nextflow_script(execution):
+    modules = execution['modules']
+
+    fragments = []
+    fragments.append('''params.fasta = "example/proteins.fas"
+Channel.fromPath(params.fasta).set{fasta}''')
+    target_channels = ["for_"+m['id'] for m in modules]
+    fragments.append('fasta.into{'+';'.join(target_channels)+';}')
+
+    for m in modules:
+        fragments.append(analysis_template.substitute(m))
+        fragments.append(convert_template.substitute(m))
+
+    json_inputs = []
+    for m in modules:
+        json_inputs.append(input_template.substitute(m))
+
+    fragments.append(join_jsons_template.substitute({'inputs': '\n'.join(json_inputs)}))
+    fragments.append(split_jsons_template.substitute(execution))
+        
+    nextflow_script = '\n'.join(fragments)
+    return nextflow_script