Fix mypy warning + check counting and mapping types

meteor/counter.py

@@ -50,6 +50,9 @@ class Counter(Session):
     json_data: dict = field(default_factory=dict)
 
     def __post_init__(self) -> None:
+        if self.counting_type not in Counter.COUNTING_TYPES:
+            raise ValueError(f'{self.counting_type} is not a valid counting type')
+
         if self.meteor.tmp_path:
             self.meteor.tmp_path.mkdir(exist_ok=True)
         self.meteor.tmp_dir = Path(mkdtemp(dir=self.meteor.tmp_path))
@@ -133,6 +136,8 @@ def filter_alignments(
         # contains a list of alignment of each read
         reads: dict[str, list[AlignedSegment]] = {}
         for element in cramdesc:
+            assert element.query_name is not None and element.reference_name is not None
+
             # identity = (element.query_length - element.get_tag("NM")) / element.query_length
             # identity = 1.0 - (element.get_tag("NM") / element.query_alignment_length)
             ali = sum(self.get_aligned_nucleotides(element))

meteor/mapper.py

@@ -33,7 +33,7 @@ class Mapper(Session):
     """Run the bowtie"""
 
     DEFAULT_NUM_THREADS : ClassVar[int] = 1
-    MAPPING_TYPES: ClassVar[list[str]] = ['end_to_end', 'local']
+    MAPPING_TYPES: ClassVar[list[str]] = ['end-to-end', 'local']
     DEFAULT_MAPPING_TYPE: ClassVar[str] = 'end-to-end'
     DEFAULT_TRIM: ClassVar[int] = 80
     NO_TRIM: ClassVar[int] = 0
@@ -48,6 +48,10 @@ class Mapper(Session):
     counting_type: str
     identity_threshold: float
 
+    def __post_init__(self) -> None:
+        if self.mapping_type not in Mapper.MAPPING_TYPES:
+            raise ValueError(f'{self.mapping_type} is not a valid mapping type')
+
     def set_mapping_config(
         self,
         cram_file: Path,
@@ -147,6 +151,7 @@ def execute(self) -> None:
             stderr=PIPE,
         )
         # cramfile_unsorted = Path(mkstemp(dir=self.meteor.tmp_dir)[1])
+        assert mapping_exec.stdout is not None and mapping_exec.stderr is not None
         with pysam.AlignmentFile(
             mapping_exec.stdout,
             "r",

meteor/merging.py

@@ -19,7 +19,7 @@
 import logging
 import sys
 import numpy as np
-from biom.table import Table
+from biom.table import Table # type: ignore
 from typing import ClassVar
 
 
@@ -348,7 +348,7 @@ def execute(self) -> None:
                     / ref_json["annotation"]["taxonomy"]["filename"],
                     sep="\t",
                     header=0,
-                    usecols=self.ranks.keys(),
+                    usecols=list(self.ranks.keys()),
                 )
 
                 annotation = annotation[

meteor/meteor.py

@@ -782,8 +782,8 @@ def main() -> None:  # pragma: no cover
             args.min_msp_abundance,
             args.min_msp_occurrence,
             args.remove_sample_with_no_msp,
-            None,
-            None,
+            False,
+            "a",
             #args.output_mpa,
             #args.taxonomic_level,
             args.output_biom,
@@ -806,10 +806,10 @@ def main() -> None:  # pragma: no cover
             fastq_importer.execute()
             meteor.fastq_dir = Path(tmpdirname) / "test"
             meteor.ref_dir = meteor.ref_dir / "mock"
-            counter = Counter(meteor, "best", "end-to-end", 80, 0.97, 100, False, True)
+            counter = Counter(meteor, "total", "end-to-end", 80, 0.97, 100, False, True)
             counter.execute()
             meteor.fastq_dir = Path(tmpdirname) / "test2"
-            counter = Counter(meteor, "best", "end-to-end", 80, 0.97, 100, False, True)
+            counter = Counter(meteor, "total", "end-to-end", 80, 0.97, 100, False, True)
             counter.execute()
             # Remove the mapping directory and its contents
             shutil.rmtree(Path(tmpdirname) / "test")