Construct VRS parsable HGVS strings for transcript identical (p.=) HGVS protein strings

Author: bencap

src/dcd_mapping/vrs_map.py

@@ -18,7 +18,7 @@
 )
 from ga4gh.vrs.normalize import normalize
 from mavehgvs.util import parse_variant_strings
-from mavehgvs.variant import Variant
+from mavehgvs.variant import AA_CODES, Variant
 
 from dcd_mapping.lookup import (
     cdot_rest,
@@ -112,6 +112,9 @@ def _create_pre_mapped_hgvs_strings(
         msg = f"Alignment result or accession ID must be provided for {layer} annotations (Alignment was `{alignment}`)."
         raise ValueError(msg)
 
+    if raw_description == "p.=" and tx is not None:
+        raw_description = _construct_transcript_identical_hgvs_string(tx)
+
     raw_variant_strings = _parse_raw_variant_str(raw_description)
     variants, errors = parse_variant_strings(raw_variant_strings)
 
@@ -123,7 +126,7 @@ def _create_pre_mapped_hgvs_strings(
 
         # ga4gh hgvs_tools does not support intronic variants, so they will err out when vrs allele translator is called
         # therefore skip them there
-        if is_intronic_variant(variant):
+        if layer != AnnotationLayer.PROTEIN and is_intronic_variant(variant):
             msg = f"Variant is intronic and cannot be processed: {variant}"
             raise ValueError(msg)
 
@@ -174,6 +177,9 @@ def _create_post_mapped_hgvs_strings(
         msg = f"Alignment result must be provided for {layer} annotations (Alignment was `{alignment}`)."
         raise ValueError(msg)
 
+    if raw_description == "p.=" and tx is not None:
+        raw_description = _construct_transcript_identical_hgvs_string(tx)
+
     raw_variants = _parse_raw_variant_str(raw_description)
     variants, errors = parse_variant_strings(raw_variants)
 
@@ -185,7 +191,7 @@ def _create_post_mapped_hgvs_strings(
 
         # ga4gh hgvs_tools does not support intronic variants, so they will err out when vrs allele translator is called
         # therefore skip them there
-        if is_intronic_variant(variant):
+        if layer != AnnotationLayer.PROTEIN and is_intronic_variant(variant):
             msg = f"Variant is intronic and cannot be processed: {variant}"
             raise ValueError(msg)
 
@@ -326,6 +332,23 @@ def _parse_raw_variant_str(raw_description: str) -> list[str]:
     return [raw_description]
 
 
+def _construct_transcript_identical_hgvs_string(tx: TxSelectResult) -> str:
+    """Construct an HGVS string for a transcript identical variant.
+
+    :param tx: TxSelectResult object containing the transcript sequence and start position
+    :return: The constructed HGVS string
+    """
+    description = "p.["
+
+    for i, code in enumerate(tx.sequence, start=1):
+        if i == 1:
+            description += f"{AA_CODES[code]}{tx.start + i}="
+        else:
+            description += f";{AA_CODES[code]}{tx.start + i}="
+
+    return description + "]"
+
+
 def _map_protein_coding_pro(
     row: ScoreRow,
     sequence_id: str,
@@ -340,7 +363,7 @@ def _map_protein_coding_pro(
     :param transcript: The transcript selection information for a score set
     :return: VRS mapping object if mapping succeeds
     """
-    if row.hgvs_pro in {"_wt", "_sy", "NA"} or len(row.hgvs_pro) == 3:
+    if row.hgvs_pro in {"_wt", "_sy", "NA"}:
         _logger.warning(
             "Can't process variant syntax %s for %s", row.hgvs_pro, row.accession
         )
@@ -677,11 +700,7 @@ def _hgvs_pro_is_valid(hgvs_pro: str) -> bool:
     :param hgvs_nt: MAVE_HGVS protein expression
     :return: True if expression appears populated and valid
     """
-    return (
-        (hgvs_pro not in {"_wt", "_sy", "NA"})
-        and (len(hgvs_pro) != 3)
-        and ("fs" not in hgvs_pro)
-    )
+    return (hgvs_pro not in {"_wt", "_sy", "NA"}) and ("fs" not in hgvs_pro)
 
 
 def _map_protein_coding(
@@ -815,14 +834,16 @@ def _map_accession(
             hgvs_nt_mappings = _map_genomic(row, sequence_id, align_result)
             variations.append(hgvs_nt_mappings)
     else:
-        [
-            MappedScore(
-                accession_id=row.accession,
-                score=row.score,
-                error_message=f"Unrecognized accession prefix for accession id {metadata.target_accession_id}",
-            )
-            for row in records
-        ]
+        variations.append(
+            [
+                MappedScore(
+                    accession_id=row.accession,
+                    score=row.score,
+                    error_message=f"Unrecognized accession prefix for accession id {metadata.target_accession_id}",
+                )
+                for row in records
+            ]
+        )
 
     return variations