WIP: a first pass of updating for the changes in the EasyHLA object.

Tests still need to be fixed!

Author: rhliang

src/easyhla/bblab.py

@@ -70,6 +70,7 @@ def report_unmatched_sequences(
 
 def process_from_file_to_files(
     hla_alg: EasyHLA,
+    locus: HLALocus,
     filename: str,
     output_filename: str,
     mismatches_filename: str,
@@ -97,7 +98,7 @@ def process_from_file_to_files(
     with open(filename, "r", encoding="utf-8") as f:
         matched_sequences, unmatched = pair_exons(
             Bio.SeqIO.parse(f, "fasta"),
-            hla_alg.locus,
+            locus.value,
             list(hla_alg.standards.values())[0],
         )
 
@@ -234,10 +235,11 @@ def main(
 ) -> None:
     min_log_level = max(min(40, (4 - log_level) * 10), 50)
     logger.setLevel(min_log_level)
-    easyhla = EasyHLA(locus=locus.value)
+    easyhla = EasyHLA()
 
     process_from_file_to_files(
         easyhla,
+        locus,
         sequence_file.as_posix(),
         output_file.as_posix(),
         mismatch_file.as_posix(),

src/easyhla/clinical_hla.py

@@ -23,9 +23,7 @@
 from .easyhla import EasyHLA
 from .models import (
     HLAInterpretation,
-    HLAProteinPair,
     HLASequence,
-    HLAStandard,
 )
 from .utils import HLA_LOCUS
 
@@ -49,42 +47,10 @@
 # These are the "configuration files" that the algorithm uses; these are or may
 # be updated, in which case you specify the path to the new version in the
 # environment.
-HLA_STANDARDS: Final[dict[HLA_LOCUS, Optional[str]]] = {
-    "A": os.environ.get("HLA_STANDARDS_A"),
-    "B": os.environ.get("HLA_STANDARDS_B"),
-    "C": os.environ.get("HLA_STANDARDS_C"),
-}
+HLA_STANDARDS: Final[str] = os.environ.get("HLA_STANDARDS")
 HLA_FREQUENCIES: Final[str] = os.environ.get("HLA_FREQUENCIES")
 
 
-def interpret_sequences(
-    sequences: list[HLASequence],
-    locus: HLA_LOCUS,
-    standards_file: Optional[str] = None,
-    frequencies_file: Optional[str] = None,
-) -> tuple[list[HLAInterpretation], dict[HLAProteinPair, int]]:
-    curr_standards: Optional[dict[str, HLAStandard]] = None
-    curr_frequencies: Optional[dict[HLAProteinPair, int]] = None
-    if frequencies_file is not None:
-        with open(frequencies_file) as f:
-            curr_frequencies = EasyHLA.read_hla_frequencies(locus, f)
-    if standards_file is not None:
-        with open(standards_file) as f:
-            curr_standards = EasyHLA.read_hla_standards(f)
-    easyhla: EasyHLA = EasyHLA(
-        locus,
-        hla_standards=curr_standards,
-        hla_frequencies=curr_frequencies,
-    )
-    interpretations: list[HLAInterpretation] = []
-    for sequence in sequences:
-        try:
-            interpretations.append(easyhla.interpret(sequence))
-        except EasyHLA.NoMatchingStandards:
-            pass
-    return interpretations, easyhla.hla_frequencies
-
-
 def prepare_interpretation_for_serialization(
     interpretation: HLAInterpretation,
     locus: HLA_LOCUS,
@@ -110,16 +76,27 @@ class SequencesByLocus(TypedDict):
     C: list[HLASequenceC]
 
 
+def interpret_sequences(
+    hla_alg: EasyHLA,
+    sequences: list[HLASequence],
+) -> list[HLAInterpretation]:
+    interpretations: list[HLAInterpretation] = []
+    for sequence in sequences:
+        try:
+            interpretations.append(hla_alg.interpret(sequence))
+        except EasyHLA.NoMatchingStandards:
+            pass
+    return interpretations
+
+
 def clinical_hla_driver(
     input_dir: str,
     db_engine: Optional[Engine] = None,
-    hla_a_standards: Optional[str] = None,
     hla_a_results: Optional[str] = None,
-    hla_b_standards: Optional[str] = None,
     hla_b_results: Optional[str] = None,
-    hla_c_standards: Optional[str] = None,
     hla_c_results: Optional[str] = None,
-    hla_frequencies: Optional[str] = None,
+    standards_path: Optional[str] = None,
+    frequencies_path: Optional[str] = None,
 ) -> None:
     # Read in the sequences:
     sequences: dict[HLA_LOCUS, list[HLASequence]] = {
@@ -131,35 +108,15 @@ def clinical_hla_driver(
         sequences[locus] = read_bc_sequences(input_dir, locus, logger)
 
     # Perform interpretations:
-    standards_files: dict[HLA_LOCUS, Optional[str]] = {
-        "A": hla_a_standards,
-        "B": hla_b_standards,
-        "C": hla_c_standards,
-    }
     interpretations: dict[HLA_LOCUS, list[HLAInterpretation]] = {
         "A": [],
         "B": [],
         "C": [],
     }
-    frequencies: dict[HLA_LOCUS, dict[HLAProteinPair, int]] = {
-        "A": {},
-        "B": {},
-        "C": {},
-    }
-
     processing_datetime: datetime = datetime.now()
-
+    easyhla: EasyHLA = EasyHLA.use_config(standards_path, frequencies_path)
     for locus in ("A", "B", "C"):
-        curr_interps: list[HLAInterpretation]
-        curr_freqs: dict[HLAProteinPair, int]
-        curr_interps, curr_freqs = interpret_sequences(
-            sequences[locus],
-            locus,
-            standards_files[locus],
-            hla_frequencies,
-        )
-        interpretations[locus] = curr_interps
-        frequencies[locus] = curr_freqs
+        interpretations[locus] = interpret_sequences(easyhla, sequences[locus])
 
     # Prepare the interpretations for output:
     seqs_for_db: SequencesByLocus = {
@@ -218,18 +175,18 @@ def main():
         default=DEFAULT_INPUT_DIR,
     )
     for locus in ("A", "B", "C"):
-        parser.add_argument(
-            f"--hla_{locus.lower()}_standards",
-            help=f"CSV file containing the (reduced) HLA-{locus} standards to use",
-            type=str,
-            default=None,
-        )
         parser.add_argument(
             f"--hla_{locus.lower()}_results",
             help=f"CSV file containing the HLA-{locus} results",
             type=str,
             default=f"hla_{locus.lower()}_seq.csv",
         )
+    parser.add_argument(
+        "--hla_standards",
+        help="YAML file containing the HLA standards to use",
+        type=str,
+        default=None,
+    )
     parser.add_argument(
         "--hla_frequencies",
         help=(
@@ -253,7 +210,6 @@ def main():
         help="If set, skip connecting to the database entirely (overrules --sqlite)",
         action="store_true",
     )
-    # FIXME what to do about "last modified"?
     args: argparse.Namespace = parser.parse_args()
 
     # Connect to the database:
@@ -291,12 +247,10 @@ def schema_workaround(dbapi_connection, _):
     clinical_hla_driver(
         args.input_dir,
         db_engine,
-        args.hla_a_standards,
         args.hla_a_results,
-        args.hla_b_standards,
         args.hla_b_results,
-        args.hla_c_standards,
         args.hla_c_results,
+        args.hla_standards,
         args.hla_frequencies,
     )
 

src/easyhla/easyhla.py

@@ -30,7 +30,7 @@
 DATE_FORMAT = "%a %b %d %H:%M:%S %Z %Y"
 
 
-class ProcessedStoredStandards(TypedDict):
+class LoadedStandards(TypedDict):
     tag: str
     last_modified: datetime
     standards: dict[HLA_LOCUS, dict[str, HLAStandard]]
@@ -49,7 +49,7 @@ class EasyHLA:
 
     def __init__(
         self,
-        hla_standards: Optional[ProcessedStoredStandards] = None,
+        loaded_standards: Optional[LoadedStandards] = None,
         hla_frequencies: Optional[dict[HLA_LOCUS, dict[HLAProteinPair, int]]] = None,
     ):
         """
@@ -58,14 +58,14 @@ def __init__(
         :param logger: Python logger object, defaults to None
         :type logger: Optional[logging.Logger], optional
         """
-        if hla_standards is None:
-            hla_standards = self.load_default_hla_standards()
+        if loaded_standards is None:
+            loaded_standards = self.load_default_hla_standards()
 
-        self.hla_standards: dict[HLA_LOCUS, dict[str, HLAStandard]] = hla_standards[
+        self.hla_standards: dict[HLA_LOCUS, dict[str, HLAStandard]] = loaded_standards[
             "standards"
         ]
-        self.last_modified: datetime = hla_standards["last_modified"]
-        self.tag: str = hla_standards["tag"]
+        self.last_modified: datetime = loaded_standards["last_modified"]
+        self.tag: str = loaded_standards["tag"]
 
         self.hla_frequencies: dict[HLA_LOCUS, dict[HLAProteinPair, int]]
         if hla_frequencies is not None:
@@ -82,7 +82,7 @@ def use_config(
         """
         An alternate constructor that accepts file paths for the configuration.
         """
-        processed_stds: Optional[ProcessedStoredStandards] = None
+        processed_stds: Optional[LoadedStandards] = None
         frequencies: Optional[dict[HLA_LOCUS, dict[HLAProteinPair, int]]] = None
 
         if standards_path is not None:
@@ -96,9 +96,7 @@ def use_config(
         return cls(processed_stds, frequencies)
 
     @staticmethod
-    def read_hla_standards(
-        standards_io: TextIOBase,
-    ) -> ProcessedStoredStandards:
+    def read_hla_standards(standards_io: TextIOBase) -> LoadedStandards:
         """
         Read HLA standards from a specified file-like object.
 
@@ -133,7 +131,8 @@ def read_hla_standards(
             "standards": hla_stds,
         }
 
-    def load_default_hla_standards(self) -> dict[str, HLAStandard]:
+    @staticmethod
+    def load_default_hla_standards() -> dict[str, HLAStandard]:
         """
         Load HLA Standards from reference file.
 
@@ -146,7 +145,7 @@ def load_default_hla_standards(self) -> dict[str, HLAStandard]:
             "hla_standards.yaml",
         )
         with open(standards_filename) as standards_file:
-            return self.read_hla_standards(standards_file)
+            return EasyHLA.read_hla_standards(standards_file)
 
     @staticmethod
     def read_hla_frequencies(
@@ -184,9 +183,8 @@ def read_hla_frequencies(
                 hla_freqs[locus][protein_pair] += 1
         return hla_freqs
 
-    def load_default_hla_frequencies(
-        self,
-    ) -> dict[HLA_LOCUS, dict[HLAProteinPair, int]]:
+    @staticmethod
+    def load_default_hla_frequencies() -> dict[HLA_LOCUS, dict[HLAProteinPair, int]]:
         """
         Load HLA frequencies from reference file.
 
@@ -205,7 +203,7 @@ def load_default_hla_frequencies(
             "hla_frequencies.csv",
         )
         with open(default_frequencies_filename, "r") as f:
-            hla_freqs = self.read_hla_frequencies(f)
+            hla_freqs = EasyHLA.read_hla_frequencies(f)
         return hla_freqs
 
     @staticmethod

src/easyhla/interpret_from_json.py

@@ -39,22 +39,14 @@ def main():
         error_result: HLAResult = HLAResult(errors=errors)
         print(error_result.model_dump_json())
     else:
-        curr_standards: Optional[dict[str, HLAStandard]] = None
-        if hla_input.hla_std_path is not None:
-            with open(hla_input.hla_std_path) as f:
-                curr_standards = EasyHLA.read_hla_standards(f)
-
-        curr_frequencies: Optional[dict[HLAProteinPair, int]] = None
-        if hla_input.hla_freq_path is not None:
-            with open(hla_input.hla_freq_path) as f:
-                curr_frequencies = EasyHLA.read_hla_frequencies(hla_input.locus, f)
-
-        easyhla: EasyHLA = EasyHLA(
-            locus=hla_input.locus,
-            hla_standards=curr_standards,
-            hla_frequencies=curr_frequencies,
+        easyhla: EasyHLA = EasyHLA.use_config(
+            hla_input.hla_std_path,
+            hla_input.hla_freq_path,
+        )
+        interp: HLAInterpretation = easyhla.interpret(
+            hla_input.hla_sequence(),
+            hla_input.locus,
         )
-        interp: HLAInterpretation = easyhla.interpret(hla_input.hla_sequence())
         print(HLAResult.build_from_interpretation(interp).model_dump_json())