WIP

snappy_pipeline/apps/snappy_snake.py

@@ -52,6 +52,7 @@
     tcell_crg_report,
     tumor_mutational_burden,
     varfish_export,
+    varfish_export_external,
     variant_annotation,
     variant_calling,
     variant_checking,
@@ -108,6 +109,7 @@
     "tcell_crg_report": tcell_crg_report,
     "tumor_mutational_burden": tumor_mutational_burden,
     "varfish_export": varfish_export,
+    "varfish_export_external": varfish_export_external,
     "variant_annotation": variant_annotation,
     "variant_calling": variant_calling,
     "variant_checking": variant_checking,

snappy_pipeline/workflows/varfish_export/__init__.py

@@ -55,6 +55,7 @@
 """
 
 from itertools import chain
+from pathlib import Path
 import re
 import typing
 import warnings
@@ -243,7 +244,9 @@ def _is_pedigree_good(self, pedigree: Pedigree) -> bool:
     def _get_input_files_annotate_seqvars(self, wildcards):
         yield "ped", "work/write_pedigree.{index_ngs_library}/out/{index_ngs_library}.ped"
 
-        variant_calling = self.parent.sub_workflows["variant_calling"]
+        variant_calling = self.parent.sub_workflows.get("variant_calling")
+        if variant_calling is None:
+            variant_calling = lambda p: str(Path(self.parent.work_dir) / p)
 
         path = (
             "output/{mapper}.{var_caller}.{index_ngs_library}/out/"
@@ -287,13 +290,14 @@ def _get_output_files_annotate_seqvars(self):
 
     def _get_params_annotate_seqvars(self, wildcards: Wildcards) -> typing.Dict[str, typing.Any]:
         pedigree = self.index_ngs_library_to_pedigree[wildcards.index_ngs_library]
+        step_name = self.parent.name
         for donor in pedigree.donors:
             if (
                 donor.dna_ngs_library
                 and donor.dna_ngs_library.extra_infos.get("libraryType") == "WGS"
             ):
-                return {"step_name": "varfish_export"}
-        return {"step_name": "varfish_export"}
+                return {"step_name": step_name}
+        return {"step_name": step_name}
 
     @dictify
     def _get_input_files_annotate_strucvars(self, wildcards):
@@ -365,7 +369,10 @@ def _get_input_files_annotate_strucvars(self, wildcards):
             )
         yield "vcf", vcfs
 
-        ngs_mapping = self.parent.sub_workflows["ngs_mapping"]
+        ngs_mapping = self.parent.sub_workflows("ngs_mapping")
+        if ngs_mapping is None:
+            ngs_mapping = lambda p: str(Path(self.parent.work_dir) / p)
+
         pedigree = self.index_ngs_library_to_pedigree[wildcards.index_ngs_library]
         library_names = [
             donor.dna_ngs_library.name for donor in pedigree.donors if donor.dna_ngs_library

snappy_pipeline/workflows/varfish_export_external/Snakefile

@@ -0,0 +1,120 @@
+import os
+
+from snappy_pipeline import expand_ref
+from snappy_pipeline.workflows.varfish_export_external import VarfishExportExternalWorkflow
+
+__author__ = "Max Xiaohang Zhao <[email protected]>"
+
+
+# Configuration ===============================================================
+
+
+configfile: "config.yaml"
+
+
+# Expand "$ref" JSON pointers in configuration (also works for YAML)
+config, lookup_paths, config_paths = expand_ref("config.yaml", config)
+
+# WorkflowImpl Object Setup ===================================================
+
+wf = VarfishExportExternalWorkflow(workflow, config, lookup_paths, config_paths, os.getcwd())
+
+# Rules =======================================================================
+
+
+localrules:
+    varfish_export_write_pedigree_run,
+
+
+rule all:
+    input:
+        wf.get_result_files(),
+
+
+# Export for VarFish ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+# Run link in external dragen files
+rule varfish_export_link_in_external_run:
+    input:
+        wf.get_input_files("link_in_external", "run")
+    output:
+        wf.get_output_files("link_in_external", "run")
+    run:
+        wf.substep_dispatch("link_in_external", "run", wildcards, output)
+
+# Write out pedigree file -----------------------------------------------------
+
+
+rule varfish_export_write_pedigree_run:
+    input:
+        unpack(wf.get_input_files("write_pedigree_with_sample_name", "run")),
+    output:
+        wf.get_output_files("write_pedigree_with_sample_name", "run"),
+    run:
+        wf.substep_dispatch("write_pedigree_with_sample_name", "run", wildcards, output)
+
+
+# Run varfish-annotator-cli annotate-seqvars -----------------------------------
+
+
+rule varfish_export_mehari_annotate_seqvars:
+    input:
+        unpack(wf.get_input_files("mehari", "annotate_seqvars")),
+    output:
+        **wf.get_output_files("mehari", "annotate_seqvars"),
+    threads: wf.get_resource("mehari", "annotate_seqvars", "threads")
+    resources:
+        time=wf.get_resource("mehari", "annotate_seqvars", "time"),
+        memory=wf.get_resource("mehari", "annotate_seqvars", "memory"),
+        partition=wf.get_resource("mehari", "annotate_seqvars", "partition"),
+        tmpdir=wf.get_resource("mehari", "annotate_seqvars", "tmpdir"),
+    log:
+        **wf.get_log_file("mehari", "annotate_seqvars"),
+    params:
+        **{"args": wf.get_params("mehari", "annotate_seqvars")},
+    wrapper:
+        wf.wrapper_path("mehari/annotate_seqvars")
+
+
+# Run varfish-annotator-cli annotate-strucvars ---------------------------------
+
+
+rule varfish_export_mehari_annotate_strucvars:
+    input:
+        unpack(wf.get_input_files("mehari", "annotate_strucvars")),
+    output:
+        **wf.get_output_files("mehari", "annotate_strucvars"),
+    threads: wf.get_resource("mehari", "annotate_strucvars", "threads")
+    resources:
+        time=wf.get_resource("mehari", "annotate_strucvars", "time"),
+        memory=wf.get_resource("mehari", "annotate_strucvars", "memory"),
+        partition=wf.get_resource("mehari", "annotate_strucvars", "partition"),
+        tmpdir=wf.get_resource("mehari", "annotate_strucvars", "tmpdir"),
+    log:
+        **wf.get_log_file("mehari", "annotate_strucvars"),
+    params:
+        **{"args": wf.get_params("mehari", "annotate_strucvars")},
+    wrapper:
+        wf.wrapper_path("mehari/annotate_strucvars")
+
+
+# Gather statistics about the alignment ---------------------------------------
+
+
+# rule varfish_export_mehari_bam_qc:
+#     input:
+#         unpack(wf.get_input_files("mehari", "bam_qc")),
+#     output:
+#         **wf.get_output_files("mehari", "bam_qc"),
+#     threads: wf.get_resource("mehari", "bam_qc", "threads")
+#     resources:
+#         time=wf.get_resource("mehari", "bam_qc", "time"),
+#         memory=wf.get_resource("mehari", "bam_qc", "memory"),
+#         partition=wf.get_resource("mehari", "bam_qc", "partition"),
+#         tmpdir=wf.get_resource("mehari", "bam_qc", "tmpdir"),
+#     log:
+#         **wf.get_log_file("mehari", "bam_qc"),
+#     params:
+#         **{"args": wf.get_params("mehari", "bam_qc")},
+#     wrapper:
+#         wf.wrapper_path("mehari/bam_qc")