add tool and test cases

Author: lindaxiang

payload-gen-variant-processing/Dockerfile

@@ -2,8 +2,14 @@ FROM python:3.7.5-slim-buster
 
 LABEL org.opencontainers.image.source https://github.com/icgc-argo/data-processing-utility-tools
 
+RUN groupadd -g 1000 ubuntu &&\
+    useradd -l -u 1000 -g ubuntu ubuntu &&\
+    install -d -m 0755 -o ubuntu -g ubuntu /home/ubuntu
+
 ENV PATH="/tools:${PATH}"
 
 COPY *.py /tools/
 
+USER ubuntu
+
 CMD ["/bin/bash"]

payload-gen-variant-processing/main.nf

@@ -2,17 +2,17 @@
 
 /*
   Copyright (C) 2021,  Ontario Institute for Cancer Research
-
+  
   This program is free software: you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.
-
+  
   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Affero General Public License for more details.
-
+  
   You should have received a copy of the GNU Affero General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
@@ -24,7 +24,7 @@
 /* this block is auto-generated based on info from pkg.json where   */
 /* changes can be made if needed, do NOT modify this block manually */
 nextflow.enable.dsl = 2
-version = '0.1.0'  // package version
+version = '0.1.0'
 
 container = [
     'ghcr.io': 'ghcr.io/icgc-argo/data-processing-utility-tools.payload-gen-variant-processing'
@@ -44,32 +44,53 @@ params.publish_dir = ""  // set to empty string will disable publishDir
 
 
 // tool specific parmas go here, add / change as needed
-params.input_file = ""
-params.output_pattern = "*"  // output file name pattern
+params.analysis = ""
+params.files_to_upload = []
+params.wf_name = ""
+params.wf_short_name = ""
+params.wf_version = ""
+params.controlled = false
+params.process_name = ""
+params.analysis_type = ""
 
 
-process payloadGenVariantProcessing {
+process payloadGenVariantFiltering {
   container "${params.container ?: container[params.container_registry ?: default_container_registry]}:${params.container_version ?: version}"
   publishDir "${params.publish_dir}/${task.process.replaceAll(':', '_')}", mode: "copy", enabled: params.publish_dir
 
   cpus params.cpus
   memory "${params.mem} GB"
 
   input:  // input, make update as needed
-    path input_file
+    path analysis
+    path files_to_upload
+    val wf_name
+    val wf_short_name
+    val wf_version
+    val controlled
+    val process_name
+    val analysis_type
 
   output:  // output, make update as needed
-    path "output_dir/${params.output_pattern}", emit: output_file
+    path "*.payload.json", emit: payload
+    path "out/*{.vcf.gz,.vcf.gz.tbi}", emit: files_to_upload
 
   script:
     // add and initialize variables here as needed
+    arg_controlled = controlled ? "-c" : ""
+    arg_process_name = process_name ? "-p ${process_name}" : ""
+    arg_analysis_type = analysis_type ? "-t ${analysis_type}" : ""
 
-    """
-    mkdir -p output_dir
-
+    """   
     main.py \
-      -i ${input_file} \
-      -o output_dir
+      -a ${analysis} \
+      -f ${files_to_upload} \
+      -w ${wf_name} \
+      -s ${wf_short_name} \
+      -v ${wf_version} \
+      -r ${workflow.runName} \
+      -j ${workflow.sessionId} \
+      ${arg_controlled} ${arg_process_name} ${arg_analysis_type}
 
     """
 }
@@ -78,7 +99,14 @@ process payloadGenVariantProcessing {
 // this provides an entry point for this main script, so it can be run directly without clone the repo
 // using this command: nextflow run <git_acc>/<repo>/<pkg_name>/<main_script>.nf -r <pkg_name>.v<pkg_version> --params-file xxx
 workflow {
-  payloadGenVariantProcessing(
-    file(params.input_file)
+  payloadGenVariantFiltering(
+    file(params.analysis),
+    Channel.fromPath(params.files_to_upload).collect(),
+    params.wf_name,
+    params.wf_short_name,
+    params.wf_version,
+    params.controlled,
+    params.process_name,
+    params.analysis_type
   )
 }

payload-gen-variant-processing/main.py

@@ -3,17 +3,17 @@
 
 """
   Copyright (C) 2021,  Ontario Institute for Cancer Research
-
+  
   This program is free software: you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.
-
+  
   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Affero General Public License for more details.
-
+  
   You should have received a copy of the GNU Affero General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
@@ -22,32 +22,140 @@
 """
 
 import os
-import sys
 import argparse
-import subprocess
-
+import json
+import uuid
+import hashlib
+import copy
+
+variant_type_to_data_type_etc = {
+    'snv': ['Simple Nucleotide Variation', 'Raw SNV Calls', ['CaVEMan', 'bcftools'], ['GATK-Mutect2', 'bcftools']],   # dataCategory, dataType, analysis_tools
+    'indel': ['Simple Nucleotide Variation', 'Raw InDel Calls', ['Pindel', 'bcftools'], ['GATK-Mutect2', 'bcftools']]
+}
+
+workflow_full_name = {
+    'open-access-variant-filtering': 'Open Access Variant Filtering'
+}
+
+def calculate_size(file_path):
+    return os.stat(file_path).st_size
+
+
+def calculate_md5(file_path):
+    md5 = hashlib.md5()
+    with open(file_path, 'rb') as f:
+        for chunk in iter(lambda: f.read(1024 * 1024), b''):
+            md5.update(chunk)
+    return md5.hexdigest()
+
+
+def get_files_info(file_to_upload, args):
+    fname = os.path.basename(file_to_upload).split(".")
+    input_wf = fname[5]
+    variant_type = fname[7]
+    process_name = args.process_name if args.process_name else args.wf_short_name
+    new_fname = ".".join(fname[0:8]+[process_name, 'vcf.gz']+(['tbi'] if file_to_upload.endswith('.tbi') else []))
+    file_info = {
+        'fileName': new_fname,
+        'fileType': 'VCF' if new_fname.endswith('.vcf.gz') else new_fname.split(".")[-1].upper(),
+        'fileSize': calculate_size(file_to_upload),
+        'fileMd5sum': calculate_md5(file_to_upload),
+        'fileAccess': 'open' if not args.controlled else 'controlled',
+        'info': {
+            'data_category': variant_type_to_data_type_etc[variant_type][0]
+        }
+    }
+
+    if file_to_upload.endswith('.vcf.gz'):
+        file_info['dataType'] = variant_type_to_data_type_etc[variant_type][1]
+    elif file_to_upload.endswith('.vcf.gz.tbi'):
+        file_info['dataType'] = 'VCF Index'
+    else:
+        pass
+
+    if input_wf in (['sanger-wgs', 'sanger-wxs']):
+        file_info['info']['analysis_tools'] = variant_type_to_data_type_etc[variant_type][2]
+    elif input_wf in (['gatk-mutect2']):
+        file_info['info']['analysis_tools'] = variant_type_to_data_type_etc[variant_type][3]
+
+    new_dir = 'out'
+    try:
+        os.mkdir(new_dir)
+    except FileExistsError:
+        pass
+
+    dst = os.path.join(os.getcwd(), new_dir, new_fname)
+    print(dst)
+    os.symlink(os.path.abspath(file_to_upload), dst)
+
+    return file_info
+
+def get_sample_info(sample_list):
+    samples = copy.deepcopy(sample_list)
+    for sample in samples:
+        for item in ['info', 'sampleId', 'specimenId', 'donorId', 'studyId']:
+            sample.pop(item, None)
+            sample['specimen'].pop(item, None)
+            sample['donor'].pop(item, None)
+
+    return samples
 
 def main():
     """
     Python implementation of tool: payload-gen-variant-processing
-
-    This is auto-generated Python code, please update as needed!
     """
 
     parser = argparse.ArgumentParser(description='Tool: payload-gen-variant-processing')
-    parser.add_argument('-i', '--input-file', dest='input_file', type=str,
-                        help='Input file', required=True)
-    parser.add_argument('-o', '--output-dir', dest='output_dir', type=str,
-                        help='Output directory', required=True)
+    parser.add_argument("-a", dest="analysis", required=True,
+                        help="json file containing sequencing_alignment SONG analysis for tumour sample")
+    parser.add_argument("-f", dest="files_to_upload", type=str, nargs="+", help="Files to be uploaded", required=True)
+    parser.add_argument("-w", dest="wf_name", type=str, help="workflow full name", required=True)
+    parser.add_argument("-s", dest="wf_short_name", type=str, help="workflow short name", required=True)
+    parser.add_argument("-v", dest="wf_version", type=str, required=True, help="workflow version")
+    parser.add_argument("-r", dest="wf_run", type=str, required=True, help="workflow run ID")
+    parser.add_argument("-j", dest="wf_session", type=str, required=True, help="workflow session ID")
+    parser.add_argument("-c", dest="controlled", action='store_true', help="set file to be controlled access")
+    parser.add_argument("-p", dest="process_name", type=str, help="variant process name")
+    parser.add_argument("-t", dest="analysis_type", type=str, default="variant_processing", help="specify output song analysis type")
     args = parser.parse_args()
 
-    if not os.path.isfile(args.input_file):
-        sys.exit('Error: specified input file %s does not exist or is not accessible!' % args.input_file)
-
-    if not os.path.isdir(args.output_dir):
-        sys.exit('Error: specified output dir %s does not exist or is not accessible!' % args.output_dir)
-
-    subprocess.run(f"cp {args.input_file} {args.output_dir}/", shell=True, check=True)
+    analysis = {}
+    with open(args.analysis, 'r') as f:
+        analysis = json.load(f)
+
+    input_analysis_type = analysis.get('analysisType').get('name')
+    
+    payload = {
+        'analysisType': {
+            'name': args.analysis_type
+        },
+        'studyId': analysis.get('studyId'),  
+        'experiment': analysis.get('experiment'),
+        'samples': get_sample_info(analysis.get('samples')),
+        'files': [],
+        'workflow': {
+            'workflow_name': workflow_full_name.get(args.wf_name),
+            'workflow_short_name': args.wf_short_name,
+            'workflow_version': args.wf_version,
+            'run_id': args.wf_run,
+            'session_id': args.wf_session,
+            'inputs': [
+                {
+                    'input_analysis_id': analysis.get('analysisId'),
+                    'analysis_type': input_analysis_type
+                }
+            ],
+            'genome_build': 'GRCh38_hla_decoy_ebv'
+        },
+        'variant_class': analysis.get('variant_class')
+    }
+
+    for f in args.files_to_upload:
+        file_info = get_files_info(f, args)
+        payload['files'].append(file_info)
+
+    with open("%s.%s.payload.json" % (str(uuid.uuid4()), args.analysis_type), 'w') as f:
+        f.write(json.dumps(payload, indent=2))
 
 
 if __name__ == "__main__":