Update package to include entrypoint, update pyproject to build a package successfully.

Author: David Rickett

pyproject.toml

@@ -39,21 +39,31 @@ Documentation = "https://github.com/unknown/easyhla#readme"
 Issues = "https://github.com/unknown/easyhla/issues"
 Source = "https://github.com/unknown/easyhla"
 
+[project.scripts]
+easyhla = "easyhla:run"
+
 [tool.hatch.version]
 path = "src/easyhla/__about__.py"
 
 [tool.hatch.build]
 include = [
+  "src/entrypoint.py",
   "src/easyhla/*.py",
   "src/easyhla/*.csv",
   "src/easyhla/hla_nuc.fasta.mtime",
 ]
 exclude = [
   "src/easyhla/check_date_modified.py",
+  "tools",
+  "tests/output",
+  "tests/input"
 ]
 skip-excluded-dirs = true
 directory = "output"
 
+[tool.hatch.build.targets.wheel]
+packages = ["src/easyhla"]
+
 [tool.hatch.envs.default]
 dependencies = [
   "biopython",
@@ -113,6 +123,7 @@ omit = [
   "src/easyhla/__about__.py",
   "src/easyhla/check_date_modified.py",
   "tests/__init__.py",
+  "src/easyhla/entrypoint.py",
 ]
 
 [tool.coverage.report]

src/debug_entrypoint.py

@@ -1 +1,2 @@
-from .easyhla import EasyHLA
+from .easyhla import EasyHLA, HLAType
+from .entrypoint import run

src/easyhla/__init__.py

@@ -1,13 +1,14 @@
 import os
 import re
-import typer
 import logging
+import typer
 import numpy as np
+from pathlib import Path
 from datetime import datetime
 from typing import List, Optional, Dict, Tuple, Any, Final, Literal
-from operator import itemgetter, attrgetter
-
 import Bio.SeqIO
+from enum import Enum
+
 
 from .models import (
     HLAStandard,
@@ -17,6 +18,13 @@
     HLAResultRow,
 )
 
+
+class HLAType(str, Enum):
+    A = "A"
+    B = "B"
+    C = "C"
+
+
 DATE_FORMAT = "%a %b %d %H:%M:%S %Z %Y"
 
 HLA_TYPES = Final[Literal["A", "B", "C"]]
@@ -86,12 +94,13 @@ class EasyHLA:
 
     COLUMN_IDS: Final[Dict[str, int]] = {"A": 0, "B": 2, "C": 4}
 
-    def __init__(self, letter: HLA_TYPES):
+    def __init__(self, letter: HLA_TYPES, logger: Optional[logging.Logger] = None):
         if letter.upper() not in ["A", "B", "C"]:
             raise ValueError("Invalid HLA Type!")
         self.letter: str = letter.upper()
         self.hla_stds: List[HLAStandard] = self.load_hla_stds(letter=self.letter)
         self.hla_freqs: Dict[str, int] = self.load_hla_frequencies(letter=self.letter)
+        self.log = logger or logging.Logger(__name__, logging.ERROR)
 
     def check_length(self, letter: HLA_TYPES, seq: str, name: str) -> bool:
         error_condition: bool = False
@@ -121,6 +130,16 @@ def check_length(self, letter: HLA_TYPES, seq: str, name: str) -> bool:
             )
         return True
 
+    def print(
+        self,
+        message: Any,
+        log_level: int = logging.INFO,
+        to_stdout: Optional[bool] = None,
+    ) -> None:
+        self.log.log(level=log_level, msg=message)
+        if to_stdout:
+            print(message)
+
     def check_bases(self, seq: str, name: str) -> bool:
         if not re.match(r"^[ATGCRYKMSWNBDHV]+$", seq):
             raise ValueError(f"Sequence {name} has invalid characters")
@@ -316,6 +335,7 @@ def interpret(
         entry: Bio.SeqIO.SeqRecord,
         unmatched: List[List[Bio.SeqIO.SeqRecord]],
         threshold: Optional[int] = None,
+        to_stdout: Optional[bool] = None,
     ) -> Optional[HLAResult]:
         samp = entry.description
 
@@ -382,8 +402,16 @@ def interpret(
 
         matching_stds = self.get_matching_stds(seq, self.hla_stds)
         if len(matching_stds) == 0:
-            print(f"Sequence {samp} did not match any known alleles.")
-            print("Please check the locus and the orientation.")
+            self.print(
+                f"Sequence {samp} did not match any known alleles.",
+                log_level=logging.WARN,
+                to_stdout=to_stdout,
+            )
+            self.print(
+                "Please check the locus and the orientation.",
+                log_level=logging.WARN,
+                to_stdout=to_stdout,
+            )
             return None
 
         # Now, combine all the stds (pick up that can citizen!)
@@ -396,13 +424,24 @@ def interpret(
             for i, combos in all_combos_sorted:
                 if i > threshold:
                     if i == 0:
-                        print("No matches found below specified threshold.")
-                        print("Please heck the locus, orientation, and/or increase")
-                        print("number of mismatches.")
+                        self.print(
+                            "No matches found below specified threshold.",
+                            log_level=logging.WARN,
+                            to_stdout=to_stdout,
+                        )
+                        self.print(
+                            "Please heck the locus, orientation, and/or increase",
+                            log_level=logging.WARN,
+                            to_stdout=to_stdout,
+                        )
+                        self.print(
+                            "number of mismatches.",
+                            log_level=logging.WARN,
+                            to_stdout=to_stdout,
+                        )
                     break
                 for cons in combos:
                     for pair in cons.discrete_allele_names:
-                        # print(" - ".join(pair))
                         misstrings = []
                         _seq = [int(nuc) for nuc in cons.standard.split("-")]
                         for n in range(len(_seq)):
@@ -414,8 +453,16 @@ def interpret(
                                 else:
                                     dex = n + 1
                                 misstrings.append(f"{dex}:{base}->{correct_base}")
-                        # print(";".join(misstrings) + ",")
-                        # print(f"{exon2},{intron},{exon3}")
+                        self.print(
+                            ";".join(misstrings) + ",",
+                            log_level=logging.INFO,
+                            to_stdout=to_stdout,
+                        )
+                        self.print(
+                            f"{exon2},{intron},{exon3}",
+                            log_level=logging.INFO,
+                            to_stdout=to_stdout,
+                        )
 
         best_matches = all_combos_sorted[0][1]
         mismatch_count = all_combos_sorted[0][0]
@@ -488,24 +535,38 @@ def interpret(
         return HLAResult(result=row, num_pats=1, num_seqs=nseqs)
 
     def report_unmatched_sequences(
-        self, unmatched: List[List[Bio.SeqIO.SeqRecord]]
+        self,
+        unmatched: List[List[Bio.SeqIO.SeqRecord]],
+        to_stdout: Optional[bool] = None,
     ) -> None:
         for exon in [2, 3]:
             for entry in unmatched[exon % 2]:
-                print(f"No matching exon{3 - exon % 2} for {entry.description}")
+                self.print(
+                    f"No matching exon{3 - exon % 2} for {entry.description}",
+                    to_stdout=to_stdout,
+                )
 
     def run(
         self,
         letter: HLA_TYPES,
         filename: str,
         output_filename: str,
         threshold: Optional[int] = None,
+        to_stdout: Optional[bool] = None,
     ):
         rows = []
         npats = 0
         nseqs = 0
         time_start = datetime.now()
         unmatched: List[List[Bio.SeqIO.SeqRecord]] = [[], []]
+        self.print(
+            f"Run commencing {time_start.strftime(DATE_FORMAT)}. Allele definitions last updated {self.load_allele_definitions_last_modified_time().strftime(DATE_FORMAT)}.",
+            to_stdout=to_stdout,
+        )
+        self.print(
+            "ENUM,ALLELES_CLEAN,ALLELES,AMBIGUOUS,HOMOZYGOUS,MISMATCH_COUNT,MISMATCHES,EXON2,INTRON,EXON3",
+            to_stdout=to_stdout,
+        )
         with open(filename, "r", encoding="utf-8") as f:
             fasta = Bio.SeqIO.parse(f, "fasta")
             for i, entry in enumerate(fasta):
@@ -519,10 +580,16 @@ def run(
                     continue
                 else:
                     rows.append(result.result.get_result_as_str())
+                    self.print(result.result.get_result_as_str(), to_stdout=to_stdout)
                     npats += result.num_pats
                     nseqs += result.num_seqs
 
-        self.report_unmatched_sequences(unmatched)
+        self.report_unmatched_sequences(unmatched, to_stdout=to_stdout)
+        self.print(
+            f"{npats} patients, {nseqs} sequences processed.", to_stdout=to_stdout
+        )
+
+        self.log.info(f"% patients, % sequences processed.", npats, nseqs)
 
         with open(output_filename, "w", encoding="utf-8") as f:
             f.write(
@@ -564,8 +631,6 @@ def get_clean_alleles(self, all_alleles: List[List[str]]) -> str:
             [a[0].strip().split(":"), a[1].strip().split(":")] for a in all_alleles
         ]
 
-        print(all_alleles)
-
         clean_allele: List[str] = []
         for n in [0, 1]:
             for i in [4, 3, 2, 1]:
@@ -576,7 +641,6 @@ def get_clean_alleles(self, all_alleles: List[List[str]]) -> str:
                     break
 
         clean_allele_str: str = " - ".join(clean_allele)
-        print(clean_allele_str)
         return clean_allele_str
 
     def get_alleles(
@@ -660,17 +724,3 @@ def sort_allele(item: Tuple[str, int]):
                     alleles.remove(a)
 
         return ambig, alleles
-
-
-if __name__ == "__main__":
-    input_file = "tests/input/test.fasta"
-    output_file = "tests/output/test.csv"
-
-    easyhla = EasyHLA("A")
-
-    easyhla.run(
-        easyhla.letter,
-        input_file,
-        output_file,
-        0,
-    )

src/easyhla/easyhla.py

@@ -0,0 +1,66 @@
+import typer
+import logging
+from pathlib import Path
+from easyhla import EasyHLA, HLAType
+
+
+def main(
+    letter: HLAType = typer.Option(
+        HLAType.A.value, "--letter", "-l", help="", case_sensitive=False
+    ),
+    mismatch_threshold: int = typer.Option(
+        0,
+        "--threshold",
+        "-t",
+        help="Maximum allowed number of mismatches for a sequence to contain versus reference sequences.",
+        min=0,
+    ),
+    log_level: int = typer.Option(
+        0,
+        "-v",
+        count=True,
+        help="Logging level from [Error, Warn, Info, Debug] default Error. Repeat -v's to receive more verbose output",
+    ),
+    print_to_stdout: bool = typer.Option(
+        False,
+        "--print",
+        "-p",
+        help="Print to stdout as sequences are interpretted",
+        flag_value=True,
+        is_flag=True,
+    ),
+    sequence_file: Path = typer.Argument(
+        ...,
+        help="Sequence file in fasta format to be classified.",
+        dir_okay=False,
+        file_okay=True,
+        exists=True,
+        readable=True,
+        path_type=str,
+    ),
+    output_file: Path = typer.Argument(
+        "output.csv",
+        help="Output file in csv format.",
+        dir_okay=False,
+        file_okay=True,
+        exists=False,
+        writable=True,
+        allow_dash=False,
+        path_type=str,
+    ),
+) -> None:
+    min_log_level = max(min(40, (4 - log_level) * 10), 50)
+    logger = logging.Logger(__name__, min_log_level)
+    easyhla = EasyHLA(letter=letter.value, logger=logger)
+
+    easyhla.run(
+        easyhla.letter,
+        sequence_file,
+        output_file,
+        threshold=mismatch_threshold,
+        to_stdout=print_to_stdout,
+    )
+
+
+def run():
+    typer.run(main)