Finish typing sunbeamlib

src/sunbeamlib/post.py

@@ -3,7 +3,7 @@
 import os
 
 
-def compile_benchmarks(benchmark_fp: str, stats_fp: str):
+def compile_benchmarks(benchmark_fp: str, stats_fp: str) -> None:
     """Aggregate all the benchmark files into one and put it in stats_fp"""
     benchmarks = []
     try:

src/sunbeamlib/qc.py

@@ -3,10 +3,12 @@
 """
 
 import gzip
+from pathlib import Path
 from sunbeamlib.parse import parse_fastq, write_many_fastq
+from typing import List, TextIO
 
 
-def filter_ids(fp_in, fp_out, ids, log):
+def filter_ids(fp_in: Path, fp_out: Path, ids: List[str], log: TextIO) -> None:
     """Remove ids from FASTQ file.
 
     fp_in: path to input FASTQ
@@ -31,7 +33,7 @@ def filter_ids(fp_in, fp_out, ids, log):
         write_many_fastq(records, f_out)
 
 
-def remove_pair_id(id, log):
+def remove_pair_id(id: str, log: TextIO) -> str:
     """Remove the 1 or 2 from a paired read ID
 
     id: id string

src/sunbeamlib/reports.py

@@ -1,39 +1,41 @@
-from collections import OrderedDict
+import pandas
 import re
 import os
 import sys
-
-import pandas
+from collections import OrderedDict
 from io import StringIO
+from typing import TextIO
 
 
-def parse_trim_summary_paired(f):
+def parse_trim_summary_paired(f: TextIO) -> OrderedDict[str, str]:
     for line in f.readlines():
         if line.startswith("Input Read"):
             vals = re.findall("\D+\: (\d+)", line)
             keys = ("input", "both_kept", "fwd_only", "rev_only", "dropped")
             return OrderedDict(zip(keys, vals))
 
 
-def parse_trim_summary_single(f):
+def parse_trim_summary_single(f: TextIO) -> OrderedDict[str, str]:
     for line in f:
         if line.startswith("Input Read"):
             vals = re.findall("\D+\: (\d+)", line)
             keys = ("input", "kept", "dropped")
             return OrderedDict(zip(keys, vals))
 
 
-def parse_decontam_log(f):
+def parse_decontam_log(f: TextIO) -> OrderedDict[str, str]:
     keys = f.readline().rstrip().split("\t")
     vals = f.readline().rstrip().split("\t")
     return OrderedDict(zip(keys, vals))
 
 
-def parse_komplexity_log(f):
-    return OrderedDict([("komplexity", len(f.readlines()))])
+def parse_komplexity_log(f: TextIO) -> OrderedDict[str, str]:
+    return OrderedDict([("komplexity", str(len(f.readlines())))])
 
 
-def summarize_qual_decontam(tfile, dfile, kfile, paired_end):
+def summarize_qual_decontam(
+    tfile: str, dfile: str, kfile: str, paired_end: bool
+) -> pandas.DataFrame:
     """Return a dataframe for summary information for trimmomatic and decontam rule"""
     tname = os.path.basename(tfile).split(".out")[0]
     with open(tfile) as tf:
@@ -56,7 +58,7 @@ def summarize_qual_decontam(tfile, dfile, kfile, paired_end):
     )
 
 
-def parse_fastqc_quality(filename):
+def parse_fastqc_quality(filename: str) -> pandas.DataFrame:
     with open(filename) as f:
         report = f.read()
     try: