feat: add PickHybProbeOnly logic to Primer3Task and Primer3Input

prymer/api/probe.py

@@ -4,6 +4,7 @@
 
 from prymer.api.primer import Primer
 
+
 @dataclass(frozen=True, init=True, kw_only=True, slots=True)
 class Probe(Primer, Metric["Probe"]):
     """Stores the properties of the designed Probe. Inherits `tm`, `penalty`,

prymer/primer3/primer3.py

@@ -140,9 +140,9 @@
 from fgpyo.util.metric import Metric
 
 from prymer.api.primer import Primer
-from prymer.api.probe import Probe
 from prymer.api.primer_like import PrimerLike
 from prymer.api.primer_pair import PrimerPair
+from prymer.api.probe import Probe
 from prymer.api.span import Span
 from prymer.api.span import Strand
 from prymer.api.variant_lookup import SimpleVariant
@@ -366,11 +366,12 @@ def design_oligos(self, design_input: Primer3Input) -> Primer3Result:  # noqa: C
                 f"Error, trying to use a subprocess that has already been "
                 f"terminated, return code {self._subprocess.returncode}"
             )
+        design_region: Span
         match design_input.task:
             case PickHybProbeOnly():
-                design_region: Span = design_input.target
+                design_region = design_input.target
             case _:
-                design_region: Span = self._create_design_region(
+                design_region = self._create_design_region(
                     target_region=design_input.target,
                     max_amplicon_length=design_input.primer_and_amplicon_params.max_amplicon_length,
                     min_primer_length=design_input.primer_and_amplicon_params.min_primer_length,
@@ -438,12 +439,13 @@ def primer3_error(message: str) -> None:
             primer3_error("Primer3 failed")
 
         match design_input.task:
-            case PickHybProbeOnly(): # Probe design
+            case PickHybProbeOnly():  # Probe design
                 all_probe_results: list[Probe] = Primer3._build_probes(
                     design_input=design_input,
                     design_results=primer3_results,
                     design_region=design_region,
-                    unmasked_design_seq=soft_masked)
+                    unmasked_design_seq=soft_masked,
+                )
 
                 return Primer3._assemble_single_designs(
                     design_input=design_input,
@@ -483,10 +485,10 @@ def primer3_error(message: str) -> None:
 
     @staticmethod
     def _build_probes(
-            design_input: Primer3Input,
-            design_results: dict[str, str],
-            design_region: Span,
-            unmasked_design_seq: str,
+        design_input: Primer3Input,
+        design_results: dict[str, str],
+        design_region: Span,
+        unmasked_design_seq: str,
     ) -> list[Probe]:
         count: int = _check_design_results(design_input, design_results)
         task_key = design_input.task.task_type
@@ -585,9 +587,13 @@ def _build_primers(
         return primers
 
     @staticmethod
-    def _assemble_single_designs(design_input: Primer3Input, design_results: dict[str, str], unfiltered_designs: Union[list[Primer], list[Probe]]
+    def _assemble_single_designs(
+        design_input: Primer3Input,
+        design_results: dict[str, str],
+        unfiltered_designs: Union[list[Primer], list[Probe]],
     ) -> Primer3Result:
-        """Screens oligo designs (primers or probes) emitted by Primer3 for acceptable dinucleotide runs and extracts failure reasons for failed designs."""
+        """Screens oligo designs (primers or probes) emitted by Primer3 for acceptable
+        dinucleotide runs and extracts failure reasons for failed designs."""
 
         valid_oligo_designs = [
             design
@@ -607,7 +613,6 @@ def _assemble_single_designs(design_input: Primer3Input, design_results: dict[st
         )
         return design_candidates
 
-
     @staticmethod
     def _build_primer_pairs(
         design_input: Primer3Input,
@@ -789,8 +794,9 @@ def _create_design_region(
 
         return design_region
 
+
 def _check_design_results(design_input: Primer3Input, design_results: dict[str, str]) -> int:
-    """Checks for any additional Primer3 errors and reports out the count of designs emitted by Primer3."""
+    """Checks for any additional Primer3 errors and reports out the count of emitted designs."""
     count_tag = design_input.task.count_tag
     maybe_count: Optional[str] = design_results.get(count_tag)
     if maybe_count is None:  # no count tag was found
@@ -803,13 +809,14 @@ def _check_design_results(design_input: Primer3Input, design_results: dict[str,
 
     return count
 
-def _has_acceptable_dinuc_run(design_input: Primer3Input, oligo_design: Union[Primer, Probe]) -> bool:
+
+def _has_acceptable_dinuc_run(
+    design_input: Primer3Input, oligo_design: Union[Primer, Probe]
+) -> bool:
+    max_dinuc_bases: int
     if type(oligo_design) is Primer:
-        max_dinuc_bases: int = design_input.primer_and_amplicon_params.primer_max_dinuc_bases
+        max_dinuc_bases = design_input.primer_and_amplicon_params.primer_max_dinuc_bases
     elif type(oligo_design) is Probe:
-        max_dinuc_bases: int = design_input.probe_params.probe_max_dinuc_bases
+        max_dinuc_bases = design_input.probe_params.probe_max_dinuc_bases
 
-    return (
-            oligo_design.longest_dinucleotide_run_length()
-            <= max_dinuc_bases
-    )
+    return oligo_design.longest_dinucleotide_run_length() <= max_dinuc_bases

prymer/primer3/primer3_task.py

@@ -194,7 +194,7 @@ def _to_input_tags(cls, target: Span, design_region: Span) -> dict[Primer3InputT
             Primer3InputTag.PRIMER_PICK_RIGHT_PRIMER: 1,
             Primer3InputTag.PRIMER_PICK_INTERNAL_OLIGO: 0,
             Primer3InputTag.SEQUENCE_TARGET: f"{target.start - design_region.start + 1},"
-                                             f"{target.length}",
+            f"{target.length}",
         }
 
     @property

tests/primer3/test_primer3.py

@@ -12,9 +12,10 @@
 from prymer.api.span import Span
 from prymer.api.span import Strand
 from prymer.api.variant_lookup import cached
-from prymer.primer3.primer3 import Primer3, _has_acceptable_dinuc_run
+from prymer.primer3.primer3 import Primer3
 from prymer.primer3.primer3 import Primer3Failure
 from prymer.primer3.primer3 import Primer3Result
+from prymer.primer3.primer3 import _has_acceptable_dinuc_run
 from prymer.primer3.primer3_input import Primer3Input
 from prymer.primer3.primer3_parameters import PrimerAndAmpliconParameters
 from prymer.primer3.primer3_parameters import ProbeParameters
@@ -167,7 +168,7 @@ def test_internal_probe_valid_designs(
     genome_ref: Path,
     valid_probe_params_no_exclude: ProbeParameters,
 ) -> None:
-    """Test that left primer designs are within the specified design specifications."""
+    """Test that internal probe designs are within the specified design specifications."""
     target = Span(refname="chr1", start=201, end=250, strand=Strand.POSITIVE)
     assert valid_probe_params_no_exclude is not None
     design_input = Primer3Input(
@@ -176,9 +177,19 @@ def test_internal_probe_valid_designs(
         task=PickHybProbeOnly(),
     )
     with Primer3(genome_fasta=genome_ref) as designer:
-        print(designer.get_design_sequences(target))
-        valid_probes = designer.design_oligos(design_input=design_input)
-    print(valid_probes)
+        primer3_result = designer.design_oligos(design_input=design_input)
+    assert len(primer3_result.filtered_designs) == 5
+    for probe_design in primer3_result.filtered_designs:
+        assert probe_design.self_any_th < valid_probe_params_no_exclude.probe_max_self_any_thermo
+        assert probe_design.self_end_th < valid_probe_params_no_exclude.probe_max_self_end_thermo
+        assert probe_design.hairpin_th < valid_probe_params_no_exclude.probe_max_hairpin_thermo
+        assert (
+            probe_design.longest_dinucleotide_run_length()
+            <= valid_probe_params_no_exclude.probe_max_dinuc_bases
+        )
+        assert probe_design.span.start >= target.start
+        assert probe_design.span.end <= target.end
+
 
 def test_left_primer_valid_designs(
     genome_ref: Path,

tests/primer3/test_primer3_weights.py

@@ -20,15 +20,7 @@ def test_primer_weights_valid() -> None:
     assert test_dict[Primer3InputTag.PRIMER_WT_SIZE_GT] == 0.1
     assert test_dict[Primer3InputTag.PRIMER_WT_TM_LT] == 1.0
     assert test_dict[Primer3InputTag.PRIMER_WT_TM_GT] == 1.0
-    assert test_dict[Primer3InputTag.PRIMER_INTERNAL_WT_SIZE_LT] == 0.25
-    assert test_dict[Primer3InputTag.PRIMER_INTERNAL_WT_SIZE_GT] == 0.25
-    assert test_dict[Primer3InputTag.PRIMER_INTERNAL_WT_TM_LT] == 1.0
-    assert test_dict[Primer3InputTag.PRIMER_INTERNAL_WT_TM_GT] == 1.0
-    assert test_dict[Primer3InputTag.PRIMER_INTERNAL_WT_GC_PERCENT_LT] == 0.5
-    assert test_dict[Primer3InputTag.PRIMER_INTERNAL_WT_GC_PERCENT_GT] == 0.5
-    assert test_dict[Primer3InputTag.PRIMER_INTERNAL_WT_SELF_ANY] == 1.0
-    assert test_dict[Primer3InputTag.PRIMER_INTERNAL_WT_SELF_END] == 1.0
-    assert len((test_dict.values())) == 21
+    assert len((test_dict.values())) == 13
 
 
 def test_probe_weights_valid() -> None: