Removed warnings, fixed Codacy style issues.

snappy_pipeline/workflows/somatic_targeted_seq_cnv_calling/__init__.py

@@ -258,7 +258,9 @@ def get_output_files(self, action):
 
     @dictify
     def _get_output_files_coverage(self):
-        """The "coverage" action creates a BCF file (CSI+MD5 files) with an entry for each target."""
+        """The "coverage" action creates a BCF file (CSI+MD5 files) with an
+        entry for each target.
+        """
         name_pattern = self.name_pattern.format(action="coverage")
         for key, ext in BCF_KEY_EXTS:
             yield key, os.path.join("work", name_pattern, "out", name_pattern + ext)
@@ -449,31 +451,32 @@ def _get_input_files_fix(self, wildcards):
         return input_files
 
     def _get_input_files_segment(self, wildcards):
-        input_files = {
-            "cnr": "work/{mapper}.cnvkit.fix.{library_name}/out/{mapper}.cnvkit.fix.{library_name}.cnr".format(
-                **wildcards
-            )
-        }
+        cnr_pattern = (
+            "work/{mapper}.cnvkit.fix.{library_name}/out/{mapper}.cnvkit.fix.{library_name}.cnr"
+        )
+        input_files = {"cnr": cnr_pattern.format(**wildcards)}
         return input_files
 
     def _get_input_files_call(self, wildcards):
-        input_files = {
-            "segment": "work/{mapper}.cnvkit.segment.{library_name}/out/{mapper}.cnvkit.segment.{library_name}.cns".format(
-                **wildcards
-            )
-        }
+        segment_pattern = (
+            "work/{mapper}.cnvkit.segment.{library_name}/out/"
+            "{mapper}.cnvkit.segment.{library_name}.cns"
+        )
+        input_files = {"segment": segment_pattern.format(**wildcards)}
         return input_files
 
     def _get_input_files_export(self, wildcards):
-        input_files = {
-            "cns": "work/{mapper}.cnvkit.call.{library_name}/out/{mapper}.cnvkit.call.{library_name}.cns".format(
-                **wildcards
-            )
-        }
+        cns_pattern = (
+            "work/{mapper}.cnvkit.call.{library_name}/out/{mapper}.cnvkit.call.{library_name}.cns"
+        )
+        input_files = {"cns": cns_pattern.format(**wildcards)}
         return input_files
 
     def _get_input_files_plot(self, wildcards):
-        tpl = "work/{mapper}.cnvkit.{substep}.{library_name}/out/{mapper}.cnvkit.{substep}.{library_name}.{ext}"
+        tpl = (
+            "work/{mapper}.cnvkit.{substep}.{library_name}/out/"
+            "{mapper}.cnvkit.{substep}.{library_name}.{ext}"
+        )
         input_files = {
             "cnr": tpl.format(substep="fix", ext="cnr", **wildcards),
             "cns": tpl.format(substep="segment", ext="cns", **wildcards),
@@ -603,7 +606,10 @@ def get_log_file(self, action):
             "plot",
             "report",
         ):
-            prefix = "work/{{mapper}}.cnvkit.{action}.{{library_name}}/log/{{mapper}}.cnvkit.{action}.{{library_name}}"
+            prefix = (
+                "work/{{mapper}}.cnvkit.{action}.{{library_name}}/log/"
+                "{{mapper}}.cnvkit.{action}.{{library_name}}"
+            )
         else:
             raise ValueError("Unknown action {}".format(action))
         prefix = prefix.format(action=action)
@@ -852,8 +858,8 @@ def check_config(self):
         self.ensure_w_config(
             ("step_config", "somatic_targeted_seq_cnv_calling", "path_ngs_mapping"),
             (
-                "Path to somatic variant calling not configured but required for targeted sequencing "
-                "CNV calling"
+                "Path to somatic variant calling not configured but required for "
+                "targeted sequencing CNV calling"
             ),
         )
         self.ensure_w_config(

snappy_pipeline/workflows/somatic_variant_filtration/__init__.py

@@ -132,8 +132,8 @@ class SomaticVariantFiltrationStepPart(BaseStepPart):
     def __init__(self, parent):
         super().__init__(parent)
         self.log_path = (
-            "work/{mapper}.{var_caller}.jannovar_annotate_somatic_vcf."
-            "dkfz_bias_filter.{tumor_library,[^\.]+}/log/snakemake.dkfz_bias_filter.log"
+            r"work/{mapper}.{var_caller}.jannovar_annotate_somatic_vcf."
+            r"dkfz_bias_filter.{tumor_library,[^\.]+}/log/snakemake.dkfz_bias_filter.log"
         )
         # Build shortcut from cancer bio sample name to matched cancer sample
         self.tumor_ngs_library_to_sample_pair = OrderedDict()
@@ -171,7 +171,7 @@ def get_normal_lib_name(self, wildcards):
         pair = self.tumor_ngs_library_to_sample_pair[wildcards.tumor_library]
         return pair.normal_sample.dna_ngs_library.name
 
-    def get_params(self, action):
+    def get_params(self, _action):
         """Return arguments to pass down."""
 
         def params_function(wildcards):
@@ -208,33 +208,33 @@ def get_input_files(self, action):
             "output/{mapper}.{var_caller}.jannovar_annotate_somatic_vcf.{tumor_library}/out/"
             "{mapper}.{var_caller}.jannovar_annotate_somatic_vcf.{tumor_library}"
         )
-        KEY_EXT = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
+        key_ext = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
         variant_annotation = self.parent.sub_workflows["somatic_variant_annotation"]
-        for key, ext in KEY_EXT.items():
+        for key, ext in key_ext.items():
             yield key, variant_annotation(tpl + ext)
         # BAM file and index
         tpl = "output/{mapper}.{tumor_library}/out/{mapper}.{tumor_library}"
-        KEY_EXT = {"bam": ".bam", "bai": ".bam.bai"}
+        key_ext = {"bam": ".bam", "bai": ".bam.bai"}
         ngs_mapping = self.parent.sub_workflows["ngs_mapping"]
-        for key, ext in KEY_EXT.items():
+        for key, ext in key_ext.items():
             yield key, ngs_mapping(tpl + ext)
 
     @dictify
     def get_output_files(self, action):
         """Return output files for the filtration"""
         assert action == "run"
         prefix = (
-            "work/{mapper}.{var_caller}.jannovar_annotate_somatic_vcf."
-            "dkfz_bias_filter.{tumor_library,[^\.]+}/out/{mapper}.{var_caller}."
-            "jannovar_annotate_somatic_vcf.dkfz_bias_filter.{tumor_library}"
+            r"work/{mapper}.{var_caller}.jannovar_annotate_somatic_vcf."
+            r"dkfz_bias_filter.{tumor_library,[^\.]+}/out/{mapper}.{var_caller}."
+            r"jannovar_annotate_somatic_vcf.dkfz_bias_filter.{tumor_library}"
         )
-        KEY_EXT = {
+        key_ext = {
             "vcf": ".vcf.gz",
             "tbi": ".vcf.gz.tbi",
             "vcf_md5": ".vcf.gz.md5",
             "tbi_md5": ".vcf.gz.tbi.md5",
         }
-        for key, ext in KEY_EXT.items():
+        for key, ext in key_ext.items():
             yield key, prefix + ext
 
     @classmethod
@@ -273,14 +273,14 @@ def _get_input_files_run(self, wildcards):
             "jannovar_annotate_somatic_vcf.dkfz_bias_filter."
             "{tumor_library}"
         )
-        KEY_EXT = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
-        for key, ext in KEY_EXT.items():
+        key_ext = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
+        for key, ext in key_ext.items():
             yield key, tpl.format(**wildcards) + ext
         # BAM file and index
         tpl = "output/{mapper}.{tumor_library}/out/{mapper}.{tumor_library}"
-        KEY_EXT = {"bam": ".bam", "bai": ".bam.bai"}
+        key_ext = {"bam": ".bam", "bai": ".bam.bai"}
         ngs_mapping = self.parent.sub_workflows["ngs_mapping"]
-        for key, ext in KEY_EXT.items():
+        for key, ext in key_ext.items():
             yield key, ngs_mapping(tpl.format(**wildcards) + ext)
         # Panel of normals TXT file
         yield "txt", self._get_output_files_write_panel()["txt"].format(**wildcards)
@@ -297,18 +297,18 @@ def get_output_files(self, action):
     @dictify
     def _get_output_files_run(self):
         prefix = (
-            "work/{mapper}.{var_caller}.jannovar_annotate_somatic_vcf."
-            "dkfz_bias_filter.eb_filter.{tumor_library,[^\.]+}/out/"
-            "{mapper}.{var_caller}.jannovar_annotate_somatic_vcf."
-            "dkfz_bias_filter.eb_filter.{tumor_library}"
+            r"work/{mapper}.{var_caller}.jannovar_annotate_somatic_vcf."
+            r"dkfz_bias_filter.eb_filter.{tumor_library,[^\.]+}/out/"
+            r"{mapper}.{var_caller}.jannovar_annotate_somatic_vcf."
+            r"dkfz_bias_filter.eb_filter.{tumor_library}"
         )
-        KEY_EXT = {
+        key_ext = {
             "vcf": ".vcf.gz",
             "tbi": ".vcf.gz.tbi",
             "vcf_md5": ".vcf.gz.md5",
             "tbi_md5": ".vcf.gz.tbi.md5",
         }
-        for key, ext in KEY_EXT.items():
+        for key, ext in key_ext.items():
             yield key, prefix + ext
 
     @dictify
@@ -400,8 +400,8 @@ def get_input_files(self, action):
             "jannovar_annotate_somatic_vcf.dkfz_bias_filter.eb_filter."
             "{tumor_library}"
         )
-        KEY_EXT = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
-        for key, ext in KEY_EXT.items():
+        key_ext = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
+        for key, ext in key_ext.items():
             yield key, tpl + ext
 
     @dictify
@@ -551,5 +551,5 @@ def check_config(self):
         """Check that the path to the NGS mapping is present"""
         self.ensure_w_config(
             ("step_config", "somatic_variant_filtration", "path_somatic_variant_annotation"),
-            ("Path to variant calling not configured but required for somatic variant annotation"),
+            "Path to variant calling not configured but required for somatic variant annotation",
         )

snappy_pipeline/workflows/variant_denovo_filtration/__init__.py

@@ -242,7 +242,7 @@ class FilterDeNovosHardStepPart(FilterDeNovosBaseStepPart):
     def __init__(self, parent):
         super().__init__(parent)
         # Output and log paths
-        self.name_pattern = "{mapper}.{caller}.%sde_novos_hard.{index_library,[^\.]+}" % (
+        self.name_pattern = r"{mapper}.{caller}.%sde_novos_hard.{index_library,[^\.]+}" % (
             self.prev_token,
         )
         self.base_path_out = os.path.join(

snappy_pipeline/workflows/wgs_cnv_calling/__init__.py

@@ -284,7 +284,9 @@ def _get_output_files_segment(self):
         modifiers = (("segments_", ".segments"), ("windows_", ".windows"))
         for kmod, emod in modifiers:
             for key, ext in BCF_KEY_EXTS.items():
-                yield kmod + key, os.path.join("work", name_pattern, "out", name_pattern + emod + ext)
+                yield kmod + key, os.path.join(
+                    "work", name_pattern, "out", name_pattern + emod + ext
+                )
 
     @dictify
     def _get_output_files_merge_segments(self):
@@ -358,7 +360,7 @@ class ErdsStepPart(BaseStepPart):
     def __init__(self, parent):
         super().__init__(parent)
         self.base_path_out = (
-            "work/{{mapper}}.erds.{{library_name}}/out/" "{{mapper}}.erds.{{library_name}}{ext}"
+            "work/{{mapper}}.erds.{{library_name}}/out/{{mapper}}.erds.{{library_name}}{ext}"
         )
         # Build shortcut from index library name to donor
         self.index_ngs_library_to_donor = OrderedDict()

snappy_pipeline/workflows/wgs_mei_annotation/__init__.py

@@ -116,10 +116,10 @@ def get_input_files(self, action):
             "work/{mapper}.{caller}.{index_ngs_library}/out/"
             "{mapper}.{caller}.{index_ngs_library}"
         )
-        KEY_EXT = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
+        key_ext = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
         # SVs
         wgs_mei_calling = self.parent.sub_workflows["wgs_mei_calling"]
-        for key, ext in KEY_EXT.items():
+        for key, ext in key_ext.items():
             yield key, wgs_mei_calling(tpl.replace("{index_ngs_library}", "merge_vcf") + ext)
 
     @dictify
@@ -130,13 +130,13 @@ def get_output_files(self, action):
             "work/{mapper}.{caller}.annotated.{index_ngs_library}/out/"
             "{mapper}.{caller}.annotated.{index_ngs_library}"
         )
-        KEY_EXT = {
+        key_ext = {
             "vcf": ".vcf.gz",
             "tbi": ".vcf.gz.tbi",
             "vcf_md5": ".vcf.gz.md5",
             "tbi_md5": ".vcf.gz.tbi.md5",
         }
-        for key, ext in KEY_EXT.items():
+        for key, ext in key_ext.items():
             yield key, prefix + ext
 
     def get_log_file(self, action):
@@ -232,9 +232,9 @@ def check_config(self):
         """Check that the path to the NGS mapping is present"""
         self.ensure_w_config(
             ("step_config", "wgs_mei_annotation", "tools_ngs_mapping"),
-            ("NGS mapping tools not configured but required for WGS SV annotation"),
+            "NGS mapping tools not configured but required for WGS SV annotation",
         )
         self.ensure_w_config(
             ("step_config", "wgs_mei_annotation", "tools_wgs_mei_calling"),
-            ("WGS SV calling tools not configured but required for WGS SV annotation"),
+            "WGS SV calling tools not configured but required for WGS SV annotation",
         )

snappy_pipeline/workflows/wgs_mei_filtration/__init__.py

@@ -165,7 +165,9 @@ def __init__(self, parent):
         self.base_path_out = os.path.join(
             "work", name_pattern, "out", name_pattern.replace(r",[^\.]+", "") + "{ext}"
         )
-        self.path_log = os.path.join("work", name_pattern, "out", name_pattern.replace(r",[^\.]+", "") + ".log")
+        self.path_log = os.path.join(
+            "work", name_pattern, "out", name_pattern.replace(r",[^\.]+", "") + ".log"
+        )
 
     def update_cluster_config(self, cluster_config):
         cluster_config["wgs_mei_filtration_{}_run".format(self.name)] = {
@@ -322,5 +324,5 @@ def check_config(self):
         """Check that the path to the WGS MEI annotation step is present"""
         self.ensure_w_config(
             ("step_config", "wgs_mei_filtration", "path_wgs_mei_annotation"),
-            ("Path to wgs_mei_annotation not configured but must be for " "wgs_mei_filtration"),
+            "Path to wgs_mei_annotation not configured but must be for wgs_mei_filtration",
         )

snappy_pipeline/workflows/wgs_sv_export/__init__.py

@@ -106,9 +106,9 @@ def get_input_files(self, action):
             "output/{mapper}.{var_caller}.annotated.{index_ngs_library}/out/"
             "{mapper}.{var_caller}.annotated.{index_ngs_library}"
         )
-        KEY_EXT = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
+        key_ext = {"vcf": ".vcf.gz", "tbi": ".vcf.gz.tbi"}
         wgs_sv_annotation = self.parent.sub_workflows["wgs_sv_annotation"]
-        for key, ext in KEY_EXT.items():
+        for key, ext in key_ext.items():
             yield key, wgs_sv_annotation(tpl + ext)
 
     @dictify