fix: use more newlines for the Bioconda bwa-aln-interactive

prymer/offtarget/bwa.py

@@ -210,6 +210,7 @@ class BwaAlnInteractive(ExecutableRunner):
         reverse_complement: reverse complement each query sequence before alignment.
         include_alt_hits: if True include hits to references with names ending in _alt, otherwise
                           do not include them.
+        header: the SAM alignment header.
     """
 
     def __init__(
@@ -293,7 +294,7 @@ def __init__(
 
         super().__init__(command=command)
 
-        # Send in a single record to be aligned so we get bwa to output a SAM header.
+        # Send a sentinel record to be aligned so we know when bwa is done outputting a header.
         self._subprocess.stdin.write(Query(id="ignore", bases="A").to_fastq())
         self.__signal_bwa()  # forces the input to be sent to the underlying process.
 
@@ -304,14 +305,13 @@ def __init__(
             if line.startswith("ignore"):
                 break
 
-        self._header = AlignmentHeader.from_text("".join(header))
+        self.header = AlignmentHeader.from_text("".join(header))
 
     def __signal_bwa(self) -> None:
         """Signals BWA to process the queries"""
-        for _ in range(3):
-            self._subprocess.stdin.flush()
-            self._subprocess.stdin.write("\n\n")
-            self._subprocess.stdin.flush()
+        self._subprocess.stdin.flush()
+        self._subprocess.stdin.write("\n" * 16)
+        self._subprocess.stdin.flush()
 
     def map_one(self, query: str, id: str = "unknown") -> BwaResult:
         """Maps a single query to the genome and returns the result.
@@ -348,7 +348,7 @@ def map_all(self, queries: list[Query]) -> list[BwaResult]:
         for query in queries:
             # get the next alignment and convert to a result
             line: str = next(self._subprocess.stdout).strip()
-            alignment = AlignedSegment.fromstring(line, self._header)
+            alignment = AlignedSegment.fromstring(line, self.header)
             results.append(self._to_result(query=query, rec=alignment))
 
         return results

prymer/offtarget/offtarget_detector.py

@@ -75,6 +75,7 @@
 """  # noqa: E501
 
 import itertools
+from contextlib import AbstractContextManager
 from dataclasses import dataclass
 from dataclasses import field
 from dataclasses import replace
@@ -121,7 +122,7 @@ class OffTargetResult:
     right_primer_spans: list[Span] = field(default_factory=list)
 
 
-class OffTargetDetector:
+class OffTargetDetector(AbstractContextManager):
     """A class for detecting off-target mappings of primers and primer pairs that uses a custom
     version of "bwa aln" named "bwa-aln-interactive".
 

tests/api/test_picking.py

@@ -573,31 +573,28 @@ def _pick_top_primer_pairs(
     max_primer_pair_hits: int,
     min_difference: int = 1,
 ) -> list[PrimerPair]:
-    offtarget_detector = OffTargetDetector(
-        ref=picking_ref,
-        max_primer_hits=max_primer_hits,
-        max_primer_pair_hits=max_primer_pair_hits,
-        three_prime_region_length=5,
-        max_mismatches_in_three_prime_region=0,
-        max_mismatches=0,
-        max_amplicon_size=params.amplicon_sizes.max,
-    )
-    dimer_checker = NtThermoAlign()
-
-    picked = pick_top_primer_pairs(
-        primer_pairs=primer_pairs,
-        num_primers=len(primer_pairs),
-        min_difference=min_difference,
-        params=params,
-        offtarget_detector=offtarget_detector,
-        is_dimer_tm_ok=lambda s1, s2: (
-            dimer_checker.duplex_tm(s1=s1, s2=s2) <= params.max_dimer_tm
-        ),
-    )
-    offtarget_detector.close()
-    dimer_checker.close()
-
-    return picked
+    with (
+        OffTargetDetector(
+            ref=picking_ref,
+            max_primer_hits=max_primer_hits,
+            max_primer_pair_hits=max_primer_pair_hits,
+            three_prime_region_length=5,
+            max_mismatches_in_three_prime_region=0,
+            max_mismatches=0,
+            max_amplicon_size=params.amplicon_sizes.max,
+        ) as offtarget_detector,
+        NtThermoAlign() as dimer_checker,
+    ):
+        return pick_top_primer_pairs(
+            primer_pairs=primer_pairs,
+            num_primers=len(primer_pairs),
+            min_difference=min_difference,
+            params=params,
+            offtarget_detector=offtarget_detector,
+            is_dimer_tm_ok=lambda s1, s2: (
+                dimer_checker.duplex_tm(s1=s1, s2=s2) <= params.max_dimer_tm
+            ),
+        )
 
 
 _PARAMS: FilteringParams = _zero_score_filtering_params(_score_input())