create Specialized Handler Classes for allele, glstring, mac etc.

Author: pbashyal-nmdp

pyard/ard_refactored.py

@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+
+from .allele_reducer import AlleleReducer
+from .gl_string_processor import GLStringProcessor
+from .mac_handler import MACHandler
+from .serology_handler import SerologyHandler
+from .v2_handler import V2Handler
+from .xx_handler import XXHandler
+from .shortnull_handler import ShortNullHandler
+
+__all__ = [
+    "AlleleReducer",
+    "GLStringProcessor",
+    "MACHandler",
+    "SerologyHandler",
+    "V2Handler",
+    "XXHandler",
+    "ShortNullHandler",
+]

pyard/handlers/__init__.py

@@ -0,0 +1,147 @@
+# -*- 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
+
+if TYPE_CHECKING:
+    from ..ard import ARD
+
+
+class AlleleReducer:
+    """Handles core allele reduction logic"""
+
+    def __init__(self, ard_instance: "ARD"):
+        self.ard = ard_instance
+
+    def reduce_allele(
+        self, allele: str, redux_type: VALID_REDUCTION_TYPES, 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)
+
+    def _add_lg_suffix(self, redux_allele):
+        """Add lg suffix to reduced allele"""
+        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/allele_reducer.py

@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+
+import functools
+from typing import List, TYPE_CHECKING
+
+from ..constants import VALID_REDUCTION_TYPES
+from ..misc import validate_reduction_type
+
+if TYPE_CHECKING:
+    from ..ard import ARD
+
+
+class GLStringProcessor:
+    """Handles GL string parsing, validation and processing"""
+
+    def __init__(self, ard_instance: "ARD"):
+        self.ard = ard_instance
+
+    def process_gl_string(
+        self, glstring: str, redux_type: VALID_REDUCTION_TYPES = "lgx"
+    ) -> str:
+        """Main GL string processing logic extracted from redux method"""
+        validate_reduction_type(redux_type)
+
+        if self.ard._config["strict"]:
+            self.validate_gl_string(glstring)
+
+        # Handle GL string delimiters
+        if "^" in glstring:
+            return self._sorted_unique_gl(
+                [self.ard.redux(a, redux_type) for a in glstring.split("^")], "^"
+            )
+
+        if "|" in glstring:
+            return self._sorted_unique_gl(
+                [self.ard.redux(a, redux_type) for a in glstring.split("|")], "|"
+            )
+
+        if "+" in glstring:
+            return self._sorted_unique_gl(
+                [self.ard.redux(a, redux_type) for a in glstring.split("+")], "+"
+            )
+
+        if "~" in glstring:
+            return self._sorted_unique_gl(
+                [self.ard.redux(a, redux_type) for a in glstring.split("~")], "~"
+            )
+
+        if "/" in glstring:
+            return self._sorted_unique_gl(
+                [self.ard.redux(a, redux_type) for a in glstring.split("/")], "/"
+            )
+
+        return glstring
+
+    def _sorted_unique_gl(self, gls: List[str], delim: str) -> str:
+        """Make a list of sorted unique GL Strings separated by delim"""
+        if delim == "~":
+            return delim.join(gls)
+
+        if delim == "+":
+            non_empty_gls = filter(lambda s: s != "", gls)
+            return delim.join(
+                sorted(
+                    non_empty_gls,
+                    key=functools.cmp_to_key(
+                        lambda a, b: self.ard.smart_sort_comparator(
+                            a, b, self.ard._config["ignore_allele_with_suffixes"]
+                        )
+                    ),
+                )
+            )
+
+        all_gls = []
+        for gl in gls:
+            all_gls += gl.split(delim)
+        unique_gls = filter(lambda s: s != "", set(all_gls))
+        return delim.join(
+            sorted(
+                unique_gls,
+                key=functools.cmp_to_key(
+                    lambda a, b: self.ard.smart_sort_comparator(
+                        a, b, self.ard._config["ignore_allele_with_suffixes"]
+                    )
+                ),
+            )
+        )
+
+    def validate_gl_string(self, glstring: str) -> bool:
+        """Validate GL string structure and components"""
+        if "^" in glstring:
+            return all(map(self.validate_gl_string, glstring.split("^")))
+        if "|" in glstring:
+            return all(map(self.validate_gl_string, glstring.split("|")))
+        if "+" in glstring:
+            return all(map(self.validate_gl_string, glstring.split("+")))
+        if "~" in glstring:
+            return all(map(self.validate_gl_string, glstring.split("~")))
+        if "/" in glstring:
+            return all(map(self.validate_gl_string, glstring.split("/")))
+
+        # what falls through here is an allele
+        is_valid_allele = self.ard._is_valid(glstring)
+        if not is_valid_allele:
+            from ..exceptions import InvalidAlleleError
+
+            raise InvalidAlleleError(f"{glstring} is not a valid Allele")
+        return is_valid_allele

pyard/handlers/gl_string_processor.py

@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+
+import functools
+import sqlite3
+from collections import Counter
+from typing import Iterable, TYPE_CHECKING
+
+from ..constants import HLA_regex, DEFAULT_CACHE_SIZE
+from ..exceptions import InvalidMACError
+from .. import db
+
+if TYPE_CHECKING:
+    from ..ard import ARD
+
+
+class MACHandler:
+    """Handles MAC (Multiple Allele Code) operations"""
+
+    def __init__(self, ard_instance: "ARD"):
+        self.ard = ard_instance
+
+    @functools.lru_cache(maxsize=DEFAULT_CACHE_SIZE)
+    def is_mac(self, allele: str) -> bool:
+        """Check if allele is a valid MAC code"""
+        if ":" in allele:
+            allele_split = allele.split(":")
+            if len(allele_split) == 2:
+                locus_antigen, code = allele_split
+                if code.isalpha():
+                    try:
+                        alleles = db.mac_code_to_alleles(self.ard.db_connection, code)
+                        if alleles:
+                            if any(map(lambda a: ":" in a, alleles)):
+                                antigen_groups = map(lambda a: a.split(":")[0], alleles)
+                                antigen_counts = Counter(antigen_groups)
+                                valid_antigen = antigen_counts.most_common(1).pop()[0]
+                                provided_antigen = locus_antigen.split("*").pop()
+                                return provided_antigen == valid_antigen
+                            return True
+                    except sqlite3.OperationalError as e:
+                        print("Error: ", e)
+        return False
+
+    def expand_mac(self, mac_code: str) -> str:
+        """Expand MAC code into GL string of alleles"""
+        if self.is_mac(mac_code):
+            locus_antigen, code = mac_code.split(":")
+            if HLA_regex.search(mac_code):
+                locus_antigen = locus_antigen.split("-")[1]
+                return "/".join(
+                    ["HLA-" + a for a in self._get_alleles(code, locus_antigen)]
+                )
+            else:
+                return "/".join(self._get_alleles(code, locus_antigen))
+        raise InvalidMACError(f"{mac_code} is an invalid MAC.")
+
+    def lookup_mac(self, allelelist_gl: str) -> str:
+        """Find MAC code corresponding to allele list"""
+        alleles = allelelist_gl.split("/")
+        allele_fields = [allele.split("*")[1] for allele in alleles]
+        antigen_groups = sorted({allele.split(":")[0] for allele in allele_fields})
+
+        if len(antigen_groups) == 1:
+            mac_expansion = "/".join(
+                sorted({allele.split(":")[1] for allele in allele_fields})
+            )
+            mac_code = db.alleles_to_mac_code(self.ard.db_connection, mac_expansion)
+            if mac_code:
+                locus = allelelist_gl.split("*")[0]
+                return f"{locus}*{antigen_groups[0]}:{mac_code}"
+
+        # Try given list order
+        mac_expansion = "/".join(allele_fields)
+        mac_code = db.alleles_to_mac_code(self.ard.db_connection, mac_expansion)
+        if mac_code:
+            locus = allelelist_gl.split("*")[0]
+            return f"{locus}*{antigen_groups[0]}:{mac_code}"
+
+        # Try sorted list
+        mac_expansion = "/".join(
+            sorted(
+                allele_fields, key=functools.cmp_to_key(self.ard.smart_sort_comparator)
+            )
+        )
+        mac_code = db.alleles_to_mac_code(self.ard.db_connection, mac_expansion)
+        if mac_code:
+            locus = allelelist_gl.split("*")[0]
+            return f"{locus}*{antigen_groups[0]}:{mac_code}"
+
+        raise InvalidMACError(f"{allelelist_gl} does not have a MAC.")
+
+    def _get_alleles(self, code, locus_antigen) -> Iterable[str]:
+        """Get alleles for MAC code"""
+        alleles = db.mac_code_to_alleles(self.ard.db_connection, code)
+
+        is_allelic_expansion = any([":" in allele for allele in alleles])
+        if is_allelic_expansion:
+            locus = locus_antigen.split("*")[0]
+            alleles = [f"{locus}*{a}" for a in alleles]
+        else:
+            alleles = [f"{locus_antigen}:{a}" for a in alleles]
+
+        return list(filter(self.ard._is_allele_in_db, alleles))