WIP: examining the conversion from old frequency file format to new.

Also some untested work on converting EasyHLA's read method for this data.

Author: rhliang

src/easyhla/easyhla.py

@@ -1,3 +1,4 @@
+import csv
 import os
 from collections.abc import Iterable, Sequence
 from datetime import datetime
@@ -161,20 +162,45 @@ def read_hla_frequencies(
             "B": {},
             "C": {},
         }
-        for line in frequencies_io.readlines():
-            for locus in ("A", "B", "C"):
-                column_id = EasyHLA.COLUMN_IDS[locus]
-                line_array = line.strip().split(",")[column_id : column_id + 2]
-
-                protein_pair: HLAProteinPair = HLAProteinPair(
-                    first_field_1=line_array[0][0:2],
-                    first_field_2=line_array[0][2:4],
-                    second_field_1=line_array[1][0:2],
-                    second_field_2=line_array[1][2:4],
-                )
-                if hla_freqs[locus].get(protein_pair, None) is None:
-                    hla_freqs[locus][protein_pair] = 0
-                hla_freqs[locus][protein_pair] += 1
+
+        locus_columns: dict[HLA_LOCUS, tuple[str, str]] = {
+            "A": ("a_first", "a_second"),
+            "B": ("b_first", "b_second"),
+            "C": ("c_first", "c_second"),
+        }
+        with frequencies_io:
+            frequencies_csv: csv.DictReader = csv.DictReader(frequencies_io)
+
+            for row in frequencies_csv:
+                for locus in ("A", "B", "C"):
+                    curr_col1: str
+                    curr_col2: str
+                    curr_col1, curr_col2 = locus_columns[locus]
+                    raw_allele_1: str = row[curr_col1]
+                    raw_allele_2: str = row[curr_col2]
+
+                    if raw_allele_1 in ("unknown", "deprecated"):
+                        continue
+                    if raw_allele_2 in ("unknown", "deprecated"):
+                        continue
+
+                    first_field_1: str
+                    first_field_2: str
+                    second_field_1: str
+                    second_field_2: str
+
+                    first_field_1, first_field_2 = raw_allele_1.split(":")
+                    second_field_1, second_field_2 = raw_allele_2.split(":")
+
+                    protein_pair: HLAProteinPair = HLAProteinPair(
+                        first_field_1=first_field_1,
+                        first_field_2=first_field_2,
+                        second_field_1=second_field_1,
+                        second_field_2=second_field_2,
+                    )
+                    if hla_freqs[locus].get(protein_pair, None) is None:
+                        hla_freqs[locus][protein_pair] = 0
+                    hla_freqs[locus][protein_pair] += 1
         return hla_freqs
 
     @staticmethod

src/scripts/examine_old_frequencies.r

@@ -0,0 +1,26 @@
+old.frequencies <- read.csv(
+    "src/easyhla/default_data/hla_frequencies.csv",
+    header=FALSE,
+    col.names=c("a1", "a2", "b1", "b2", "c1", "c2"),
+    colClasses="character",
+)
+
+a1.7401 <- which(old.frequencies$a1 == "7401")
+a2.7401 <- which(old.frequencies$a2 == "7401")
+a1.7403 <- which(old.frequencies$a1 == "7403")
+a2.7403 <- which(old.frequencies$a2 == "7403")
+
+a1.74xx <- which(startsWith(old.frequencies$a1, "74") & old.frequencies$a1 != "7400")
+a2.74xx <- which(startsWith(old.frequencies$a2, "74") & old.frequencies$a2 != "7400")
+
+old.frequencies[sort(union(a1.74xx, a2.74xx)),]
+
+c1.17xx <- which(startsWith(old.frequencies$c1, "17") & old.frequencies$c1 != "1700")
+c2.17xx <- which(startsWith(old.frequencies$c2, "17") & old.frequencies$c2 != "1700")
+c1.18xx <- which(startsWith(old.frequencies$c1, "18") & old.frequencies$c1 != "1800")
+c2.18xx <- which(startsWith(old.frequencies$c2, "18") & old.frequencies$c2 != "1800")
+
+old.frequencies[sort(union(a1.74xx, a2.74xx)),]
+
+old.frequencies[sort(union(c1.17xx, c2.17xx)),]
+old.frequencies[sort(union(c1.18xx, c2.18xx)),]

src/scripts/update_frequency_file.py

@@ -4,6 +4,7 @@
 import csv
 import logging
 import re
+from collections import Counter
 from dataclasses import dataclass, fields
 from io import TextIOBase
 from typing import ClassVar, Final, Optional, Self, TypedDict
@@ -98,6 +99,8 @@ def parse_nomenclature(remapping_str: str) -> dict[OldName, NewName]:
         except OtherLocusException:
             continue
         new_name: NewName = NewName.from_string(new_name_str)
+        if new_name.locus is None:
+            logger.info(f"Allele {old_name_str} is deprecated.")
         if old_name in remapping:
             if new_name.locus is None:
                 logger.info(
@@ -168,7 +171,10 @@ def update_old_frequencies(
 ) -> list[FrequencyRow]:
     old_frequencies_csv: csv.reader = csv.reader(old_frequencies_file)
 
-    unmapped_alleles: set[tuple[HLA_LOCUS, str]] = set()
+    # Report to the user any frequencies that are either unmapped or
+    # deprecated.
+    unmapped_alleles: Counter[tuple[HLA_LOCUS, str]] = Counter()
+    deprecated_alleles_seen: Counter[tuple[HLA_LOCUS, str]] = Counter()
 
     updated_frequencies: list[FrequencyRow] = []
     for row in old_frequencies_csv:
@@ -180,13 +186,35 @@ def update_old_frequencies(
             old_name: OldName = OldName.from_old_frequency_format(locus, row[idx])
             new_name: Optional[NewName] = old_to_new.get(old_name)
 
-            if new_name is None and (locus, row[idx]) not in unmapped_alleles:
-                logger.info(f'No mapping found for HLA-{locus} allele "{row[idx]}".')
-                unmapped_alleles.add((locus, row[idx]))
+            if new_name is None:
+                unmapped_alleles[(locus, row[idx])] += 1
+            elif new_name.locus is None:
+                deprecated_alleles_seen[(locus, row[idx])] += 1
 
             updated.append(new_name)
 
         updated_frequencies.append(FrequencyRow(*updated))
+
+    if len(unmapped_alleles) > 1:
+        logger.info(
+            "Alleles present in the old frequencies that do not have a mapping "
+            "in the new nomenclature, and their numbers of occurrences:"
+        )
+        for locus, name in unmapped_alleles:
+            logger.info(
+                f'HLA-{locus} allele "{name}": {unmapped_alleles[(locus, name)]}'
+            )
+
+    if len(deprecated_alleles_seen) > 1:
+        logger.info(
+            "Alleles present in the old frequencies that are deprecated "
+            "in the new nomenclature, and their numbers of occurrences:"
+        )
+        for locus, name in deprecated_alleles_seen:
+            logger.info(
+                f'HLA-{locus} allele "{name}": {deprecated_alleles_seen[(locus, name)]}'
+            )
+
     return updated_frequencies