Strategies

 - Refactored so that each reduction mode is a strategy.

Author: pbashyal-nmdp

pyard/__init__.py

@@ -36,7 +36,7 @@ def init(
     cache_size: int = DEFAULT_CACHE_SIZE,
     config: dict = None,
 ):
-    from .ard import ARD
+    from .ard_refactored import ARD
 
     ard = ARD(
         imgt_version=imgt_version,

pyard/ard_refactored.py

@@ -9,9 +9,9 @@
 from . import smart_sort
 from .constants import (
     HLA_regex,
-    VALID_REDUCTION_TYPES,
     DEFAULT_CACHE_SIZE,
     G_GROUP_LOCI,
+    VALID_REDUCTION_TYPE,
 )
 from .exceptions import InvalidMACError, InvalidTypingError
 from .handlers import (
@@ -156,7 +156,7 @@ def __del__(self):
 
     @functools.lru_cache(maxsize=DEFAULT_CACHE_SIZE)
     def _redux_allele(
-        self, allele: str, redux_type: VALID_REDUCTION_TYPES, re_ping=True
+        self, allele: str, redux_type: VALID_REDUCTION_TYPE, re_ping=True
     ) -> str:
         """Core allele reduction with ping logic"""
         # Handle HLA- prefix
@@ -210,7 +210,7 @@ def _redux_allele(
         return self.allele_reducer.reduce_allele(allele, redux_type, re_ping)
 
     @functools.lru_cache(maxsize=DEFAULT_CACHE_SIZE)
-    def redux(self, glstring: str, redux_type: VALID_REDUCTION_TYPES = "lgx") -> str:
+    def redux(self, glstring: str, redux_type: VALID_REDUCTION_TYPE = "lgx") -> str:
         """Main redux method using specialized handlers"""
         # Handle GL string delimiters first
         processed_gl = self.gl_processor.process_gl_string(glstring, redux_type)

pyard/constants.py

@@ -20,12 +20,15 @@
 #    > http://www.opensource.org/licenses/lgpl-license.php
 #
 import re
+import typing
 
 DEFAULT_CACHE_SIZE = 1_000
 
 HLA_regex = re.compile("^HLA-")
 
-VALID_REDUCTION_TYPES = ("G", "P", "lg", "lgx", "W", "exon", "U2", "S")
+VALID_REDUCTION_MODES = ("G", "P", "lg", "lgx", "W", "exon", "U2", "S")
+VALID_REDUCTION_TYPE = typing.Literal[VALID_REDUCTION_MODES]
+
 expression_chars = ("N", "Q", "L", "S")
 # List of P and G characters
 P_and_G_chars = ("P", "G")

pyard/handlers/allele_reducer.py

@@ -1,147 +1,34 @@
 # -*- coding: utf-8 -*-
 
-import functools
 from typing import TYPE_CHECKING
 
-from ..constants import VALID_REDUCTION_TYPES, expression_chars
-from ..exceptions import InvalidAlleleError
-from ..misc import get_n_field_allele
+from ..constants import VALID_REDUCTION_TYPE
+from ..strategies.strategy_factory import StrategyFactory
 
 if TYPE_CHECKING:
     from ..ard import ARD
 
 
 class AlleleReducer:
-    """Handles core allele reduction logic"""
+    """Handles core allele reduction logic using Strategy Pattern"""
 
     def __init__(self, ard_instance: "ARD"):
         self.ard = ard_instance
+        self.strategy_factory = StrategyFactory(ard_instance)
 
     def reduce_allele(
-        self, allele: str, redux_type: VALID_REDUCTION_TYPES, re_ping=True
+        self, allele: str, redux_type: VALID_REDUCTION_TYPE, re_ping=True
     ) -> str:
-        """Core allele reduction logic extracted from _redux_allele"""
-
-        if redux_type == "G" and allele in self.ard.ars_mappings.g_group:
-            if allele in self.ard.ars_mappings.dup_g:
-                return self.ard.ars_mappings.dup_g[allele]
-            else:
-                return self.ard.ars_mappings.g_group[allele]
-
-        elif redux_type == "P" and allele in self.ard.ars_mappings.p_group:
-            return self.ard.ars_mappings.p_group[allele]
-
-        elif redux_type in ["lgx", "lg"]:
-            if allele in self.ard.ars_mappings.lgx_group:
-                redux_allele = self.ard.ars_mappings.lgx_group[allele]
-            else:
-                redux_allele = ":".join(allele.split(":")[0:2])
-            if redux_type == "lg":
-                return self._add_lg_suffix(redux_allele)
-            return redux_allele
-
-        elif redux_type == "W":
-            if self.ard._is_who_allele(allele):
-                return allele
-            if allele in self.ard.code_mappings.who_group:
-                return self.ard.redux(
-                    "/".join(self.ard.code_mappings.who_group[allele]), redux_type
-                )
-            else:
-                return allele
-
-        elif redux_type == "exon":
-            return self._handle_exon_reduction(allele)
-
-        elif redux_type == "U2":
-            return self._handle_u2_reduction(allele)
-
-        elif redux_type == "S":
-            return self._handle_serology_reduction(allele)
-
-        else:
-            return self._handle_default_reduction(allele)
+        """Core allele reduction logic using Strategy Pattern"""
+        strategy = self.strategy_factory.get_strategy(redux_type)
+        return strategy.reduce(allele)
 
     def _add_lg_suffix(self, redux_allele):
-        """Add lg suffix to reduced allele"""
+        """Add lg suffix to reduced allele - kept for backward compatibility"""
         if "/" in redux_allele:
             return "/".join(
                 [self._add_lg_suffix(allele) for allele in redux_allele.split("/")]
             )
         if self.ard._config["ARS_as_lg"]:
             return redux_allele + "ARS"
         return redux_allele + "g"
-
-    def _handle_exon_reduction(self, allele):
-        """Handle exon reduction type"""
-        if allele in self.ard.ars_mappings.exon_group:
-            exon_group_allele = self.ard.ars_mappings.exon_group[allele]
-            last_char = allele[-1]
-            if last_char in expression_chars:
-                exon_short_null_allele = exon_group_allele + last_char
-                if self.ard.is_shortnull(exon_short_null_allele):
-                    return exon_short_null_allele
-            return exon_group_allele
-        else:
-            w_redux = self.ard.redux(allele, "W")
-            if w_redux == allele or len(w_redux.split(":")) == 2:
-                return allele
-            else:
-                return self.ard.redux(w_redux, "exon")
-
-    def _handle_u2_reduction(self, allele):
-        """Handle U2 reduction type"""
-        allele_fields = allele.split(":")
-        if len(allele_fields) == 2:
-            return allele
-        allele_2_fields = get_n_field_allele(allele, 2, preserve_expression=True)
-        if self.ard._is_allele_in_db(allele_2_fields):
-            return allele_2_fields
-        else:
-            return self.reduce_allele(allele, "lgx")
-
-    def _handle_serology_reduction(self, allele):
-        """Handle serology reduction type"""
-        from .. import db
-        from ..misc import is_2_field_allele
-
-        if is_2_field_allele(allele):
-            allele = self.reduce_allele(allele, "lgx")
-            serology_mapping = db.find_serology_for_allele(
-                self.ard.db_connection, allele, "lgx_allele_list"
-            )
-        else:
-            serology_mapping = db.find_serology_for_allele(
-                self.ard.db_connection, allele
-            )
-
-        serology_set = set()
-        for serology, allele_list in serology_mapping.items():
-            if allele in allele_list.split("/"):
-                serology_set.add(serology)
-
-        if not serology_set and is_2_field_allele(allele):
-            for serology, allele_list in serology_mapping.items():
-                allele_list_lgx = self.ard.redux(allele_list, "lgx")
-                if allele in allele_list_lgx.split("/"):
-                    serology_set.add(serology)
-
-        return "/".join(
-            sorted(
-                serology_set, key=functools.cmp_to_key(self.ard.smart_sort_comparator)
-            )
-        )
-
-    def _handle_default_reduction(self, allele):
-        """Handle default reduction cases"""
-        if allele.endswith("P"):
-            if allele in self.ard.ars_mappings.p_group.values():
-                return allele
-        elif allele.endswith("G"):
-            if allele in self.ard.ars_mappings.g_group.values():
-                return allele
-
-        if self.ard._is_allele_in_db(allele):
-            return allele
-        else:
-            raise InvalidAlleleError(f"{allele} is an invalid allele.")

pyard/handlers/gl_string_processor.py

@@ -3,7 +3,7 @@
 import functools
 from typing import List, TYPE_CHECKING
 
-from ..constants import VALID_REDUCTION_TYPES
+from ..constants import VALID_REDUCTION_TYPE
 from ..misc import validate_reduction_type
 
 if TYPE_CHECKING:
@@ -17,7 +17,7 @@ def __init__(self, ard_instance: "ARD"):
         self.ard = ard_instance
 
     def process_gl_string(
-        self, glstring: str, redux_type: VALID_REDUCTION_TYPES = "lgx"
+        self, glstring: str, redux_type: VALID_REDUCTION_TYPE = "lgx"
     ) -> str:
         """Main GL string processing logic extracted from redux method"""
         validate_reduction_type(redux_type)

pyard/misc.py

@@ -24,7 +24,7 @@
 import tempfile
 from typing import List
 
-from pyard.constants import VALID_REDUCTION_TYPES, expression_chars, P_and_G_chars
+from pyard.constants import VALID_REDUCTION_MODES, expression_chars, P_and_G_chars
 
 
 def get_n_field_allele(allele: str, n: int, preserve_expression=False) -> str:
@@ -151,5 +151,5 @@ def get_default_db_directory():
 
 
 def validate_reduction_type(ars_type):
-    if ars_type not in VALID_REDUCTION_TYPES:
-        raise ValueError(f"Reduction type needs to be one of {VALID_REDUCTION_TYPES}")
+    if ars_type not in VALID_REDUCTION_MODES:
+        raise ValueError(f"Reduction type needs to be one of {VALID_REDUCTION_MODES}")

pyard/strategies/__init__.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+
+from .base_strategy import ReductionStrategy
+from .g_strategy import GGroupStrategy
+from .p_strategy import PGroupStrategy
+from .lg_strategy import LGStrategy, LGXStrategy
+from .w_strategy import WStrategy
+from .exon_strategy import ExonStrategy
+from .u2_strategy import U2Strategy
+from .s_strategy import SStrategy
+from .default_strategy import DefaultStrategy
+from .strategy_factory import StrategyFactory
+
+__all__ = [
+    "ReductionStrategy",
+    "GGroupStrategy",
+    "PGroupStrategy",
+    "LGStrategy",
+    "LGXStrategy",
+    "WStrategy",
+    "ExonStrategy",
+    "U2Strategy",
+    "SStrategy",
+    "DefaultStrategy",
+    "StrategyFactory",
+]

pyard/strategies/base_strategy.py

@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ..ard import ARD
+
+
+class ReductionStrategy(ABC):
+    """Base class for all reduction strategies"""
+
+    def __init__(self, ard_instance: "ARD"):
+        self.ard = ard_instance
+
+    @abstractmethod
+    def reduce(self, allele: str) -> str:
+        """Reduce allele according to this strategy"""
+        pass

pyard/strategies/default_strategy.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+from typing_extensions import override
+
+from .base_strategy import ReductionStrategy
+
+
+class DefaultStrategy(ReductionStrategy):
+    """Default strategy for handling P/G suffixes and validation"""
+
+    @override
+    def reduce(self, allele: str) -> str:
+        # Make this an explicit lookup to the g_group or p_group table
+        # for stringent validation
+        if allele.endswith("P"):
+            if allele in self.ard.ars_mappings.p_group.values():
+                return allele
+        elif allele.endswith("G"):
+            if allele in self.ard.ars_mappings.g_group.values():
+                return allele
+
+        if self.ard._is_allele_in_db(allele):
+            return allele
+        else:
+            from ..exceptions import InvalidAlleleError
+
+            raise InvalidAlleleError(f"{allele} is an invalid allele.")

pyard/strategies/exon_strategy.py

@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+from typing import override
+
+from .base_strategy import ReductionStrategy
+
+
+class ExonStrategy(ReductionStrategy):
+    """Strategy for exon reduction"""
+
+    @override
+    def reduce(self, allele: str) -> str:
+        if allele in self.ard.ars_mappings.exon_group:
+            exon_group_allele = self.ard.ars_mappings.exon_group[allele]
+            # Check if the 3 field exon allele has a 4 field alleles
+            # that all have the same expression characters
+            from ..constants import expression_chars
+
+            last_char = allele[-1]
+            if last_char in expression_chars:
+                exon_short_null_allele = exon_group_allele + last_char
+                if self.ard.is_shortnull(exon_short_null_allele):
+                    return exon_short_null_allele
+            return exon_group_allele
+        else:
+            # Expand to W level and then reduce to exon
+            w_redux = self.ard.redux(allele, "W")
+            # If the W redux produces 2 field allele or the same allele, don't recurse
+            if w_redux == allele or len(w_redux.split(":")) == 2:
+                return allele
+            else:
+                # recurse with the W fields
+                return self.ard.redux(w_redux, "exon")