Update scripts

Author: Ulthran

sbx_cenote_taker.smk

@@ -9,6 +9,7 @@ VIRUS_FP = output_subdir(Cfg, "virus")
 def get_extension_path() -> Path:
     return Path(__file__).parent.resolve()
 
+
 def cenote_output() -> Path:
     return VIRUS_FP / "cenote_taker" / "{sample}.fasta"
 
@@ -30,8 +31,8 @@ rule cenote_taker:
     input:
         contigs=ASSEMBLY_FP / "megahit" / "{sample}_asm" / "final.contigs.fa",
     output:
-        VIRUS_FP / "cenote_taker" / "{sample}" / "final.contigs.fasta",
-        VIRUS_FP
+        contigs=VIRUS_FP / "cenote_taker" / "{sample}" / "final.contigs.fasta",
+        summary=VIRUS_FP
         / "cenote_taker"
         / "{sample}"
         / "{sample}"
@@ -65,7 +66,8 @@ rule cenote_taker:
             echo "Contigs file exists and is not empty" >> {log}
         else
             echo "Contigs file is empty" >> {log}
-            exit 1
+            touch {output.contigs} {output.summary}
+            exit 0
         fi
 
         if [ ! -d {params.db_fp} ] || [ ! "$(ls -A {params.db_fp})" ]; then

scripts/filter_cenote_contigs.py

@@ -1,17 +1,50 @@
 import csv
-from sunbeamlib.parse import parse_fasta, write_fasta
+import os
+from typing import Generator, TextIO
 
 
-with open(snakemake.input.summary) as f_summary, open(
-    snakemake.input.contigs
-) as f_contigs, open(snakemake.output[0], "w") as f_out:
+def parse_fasta(f: TextIO) -> Generator[tuple[str, str], None, None]:
+    header_str = ""
+    seq_str = ""
+    for line in f.readlines():
+        line = line.strip()
+        if line.startswith(">"):
+            if header_str:
+                yield header_str, seq_str
+            header_str = line
+            seq_str = ""
+        else:
+            seq_str += line
+    if header_str:
+        yield header_str, seq_str
+
+
+def write_fasta(record: list[str], f: TextIO) -> None:
+    f.write(f"{record[0]}\n")
+    f.write(f"{record[1]}\n")
+
+
+summary = snakemake.input.summary  # type: ignore
+contigs = snakemake.input.contigs  # type: ignore
+output_fp = snakemake.output[0]  # type: ignore
+include_phages = snakemake.params["include_phages"]  # type: ignore
+
+# Empty output if empty contigs
+if os.path.getsize(contigs) == 0:
+    with open(output_fp, "w") as f_out:
+        pass
+    exit(0)
+
+with open(summary) as f_summary, open(contigs) as f_contigs, open(
+    output_fp, "w"
+) as f_out:
     dr = csv.DictReader(f_summary, delimiter="\t")
     cd = {}
     phages = ["phage", "siphoviridae", "conjugative transposon"]
     for line in dr:
         if (
             all([x not in line["ORGANISM_NAME"].lower() for x in phages])
-            or snakemake.params["include_phages"]
+            or include_phages
         ) and int(line["NUM_HALLMARKS"]) > 0:
             cd[line["ORIGINAL_NAME"]] = 1
 

scripts/filter_virus_coverage.py

@@ -1,9 +1,36 @@
 import csv
-from sunbeamlib.parse import parse_fasta, write_fasta
+import os
+from typing import Generator, TextIO
 
+
+def parse_fasta(f: TextIO) -> Generator[tuple[str, str], None, None]:
+    header_str = ""
+    seq_str = ""
+    for line in f.readlines():
+        line = line.strip()
+        if line.startswith(">"):
+            if header_str:
+                yield header_str, seq_str
+            header_str = line
+            seq_str = ""
+        else:
+            seq_str += line
+    if header_str:
+        yield header_str, seq_str
+
+
+def write_fasta(record: list[str], f: TextIO) -> None:
+    f.write(f"{record[0]}\n")
+    f.write(f"{record[1]}\n")
+
+
+idx = snakemake.input.idx  # type: ignore
+fa = snakemake.input.fa  # type: ignore
+output_fp = snakemake.output[0]  # type: ignore
+log_fp = snakemake.log[0]  # type: ignore
 contigs = {}
 
-with open(snakemake.input.idx) as f_idx:
+with open(idx) as f_idx:
     rd = csv.reader(f_idx, delimiter="\t", quotechar='"')
     for row in rd:
         if row[0] != "*":
@@ -12,11 +39,11 @@
             else:
                 contigs[row[0]] = 0
 
-with open(snakemake.log[0], "w") as f_log:
+with open(log_fp, "w") as f_log:
     f_log.write(f"Contigs: {contigs}")
 
-with open(snakemake.input.fa) as f_fa, open(snakemake.output[0], "w") as f_out:
+with open(fa) as f_fa, open(output_fp, "w") as f_out:
     for header, seq in parse_fasta(f_fa):
         contig_name = header.split(" ")[0]
         if contigs[contig_name]:
-            write_fasta((contig_name, seq), f_out)
+            write_fasta([contig_name, seq], f_out)