diff --git a/src/ga4gh/vrs/extras/annotator/vcf.py b/src/ga4gh/vrs/extras/annotator/vcf.py index 3dd70dcf..44fc7793 100644 --- a/src/ga4gh/vrs/extras/annotator/vcf.py +++ b/src/ga4gh/vrs/extras/annotator/vcf.py @@ -382,9 +382,11 @@ def _get_vrs_object( :param require_validation: If `True` then validation checks must pass in order to return a VRS object. If `False` then VRS object will be returned even if validation checks fail. Defaults to `True`. + :param kwargs: Additional options passed to ``on_vrs_object``. + :raises Exception: If translation fails. The exception is logged and re-raised. """ try: - vrs_obj = self.tlr._from_gnomad( # noqa: SLF001 + vrs_obj = self.tlr.from_gnomad( vcf_coords, assembly_name=assembly, require_validation=require_validation, diff --git a/src/ga4gh/vrs/extras/translator.py b/src/ga4gh/vrs/extras/translator.py index 30dbc594..b7b21aab 100644 --- a/src/ga4gh/vrs/extras/translator.py +++ b/src/ga4gh/vrs/extras/translator.py @@ -9,6 +9,7 @@ import re from abc import ABC from collections.abc import Mapping +from enum import Enum from typing import Protocol from ga4gh.core import ga4gh_identify @@ -21,6 +22,21 @@ _logger = logging.getLogger(__name__) +class TranslationDefault(Enum): + """Mark options that use the translator instance setting. + + USE_INSTANCE_DEFAULT distinguishes an omitted option from an explicit None. + For rle_seq_limit, the default uses the instance limit, while None means no + limit. Enum members preserve their identity through copying and pickling. + """ + + INSTANCE = "instance" + + +USE_INSTANCE_DEFAULT = TranslationDefault.INSTANCE +"""Use the translator instance setting for an omitted option.""" + + class VariationToStrProtocol(Protocol): """Protocol for translating VRS objects to other string expressions. @@ -48,7 +64,7 @@ class _Translator(ABC): # noqa: B024 Translates various variation formats to and from GA4GH VRS models - Private `_from_*` parsers follow this pattern: + Format-specific `from_*` parsers follow this pattern: * If the argument does not appear to be an appropriate type, None is returned * Otherwise, the argument is expected to be of the correct type. If an error occurs during processing, an exception is raised. @@ -86,35 +102,47 @@ def __init__( self.from_vrs_translators: dict[str, VariationToStrProtocol] = {} def to_vrs( - self, var: str, fmt: str | None = None, **kwargs + self, var: str | dict, fmt: str | None = None, **kwargs ) -> models._VariationBase: - """Translate variation `var` to VRS object - - If `fmt` is None, guess the appropriate format and return the variant. - If `fmt` is specified, try only that format. - - See the class notes about private `_from_*` parsers. - - kwargs: - For CnvTranslator - copies(int): The number of copies to use. If provided will return a - CopyNumberCount - copy_change(models.CopyChange): Copy change. If not provided, default is - EFO:0030067 for deletions and EFO:0030070 for duplications - For AlleleTranslator - assembly_name (str): Assembly used for `var`. Defaults to the - `default_assembly_name`. Only used for gnomad. - require_validation (bool): If `True` then validation checks must pass in - order to return a VRS object. A `DataProxyValidationError` will be - raised if validation checks fail. If `False` then VRS object will be - returned even if validation checks fail. Defaults to `True`. - rle_seq_limit Optional(int): If RLE is set as the new state after - normalization, this sets the limit for the length of the `sequence`. - To exclude `sequence` from the response, set to 0. - For no limit, set to `None`. - Defaults value set in instance variable, `rle_seq_limit`. - do_normalize (bool): `True` if fully justified normalization should be - performed. `False` otherwise. Defaults to `True` + """Translate a variation expression or VRS JSON dictionary to a VRS object. + + If fmt is omitted, try the registered parsers in order. Format-specific + methods return None for unrecognized input; this method raises ValueError. + + :param var: Variation expression, or a VRS JSON dictionary with fmt="vrs". + :param fmt: Input format, or None to try the registered parsers. + :param kwargs: Options forwarded to the selected parser. Unused options + are ignored. + + For AlleleTranslator HGVS, SPDI, and gnomAD inputs: + + * ``do_normalize`` (bool): Whether to perform fully justified + normalization. Defaults to True. + * ``rle_seq_limit`` (int or None): Maximum sequence length to include + when normalization produces a ReferenceLengthExpression. Omission + or ``USE_INSTANCE_DEFAULT`` uses the instance's ``rle_seq_limit``. + Set to 0 to exclude the sequence, or None for no limit. + + Additional options for AlleleTranslator gnomAD inputs: + + * ``assembly_name`` (str or None): Assembly used for the input. + Omission or None uses the instance's ``default_assembly_name``. + * ``require_validation`` (bool): Whether reference mismatches raise + DataProxyValidationError. Defaults to True. False allows reference + mismatches; location bounds are still validated. + + For CnvTranslator HGVS inputs: + + * ``copies`` (int, models.Range, or None): Copy count or range, + including zero. If supplied, returns CopyNumberCount and takes + precedence over ``copy_change``. Defaults to None. + * ``copy_change`` (models.CopyChange or None): Copy change used when + ``copies`` is omitted or None. Defaults to LOSS (EFO:0030067) for + deletions and GAIN (EFO:0030070) for duplications. + :return: Translated VRS variation. + :raises NotImplementedError: If the requested format is unsupported. + :raises ValueError: If the input cannot be parsed or represented. + :raises DataProxyValidationError: If required reference or location validation fails. """ if fmt: try: @@ -139,12 +167,28 @@ def to_vrs( raise ValueError(msg) def from_vrs(self, vo: models._VariationBase, fmt: str, **kwargs) -> list[str]: - """Translate vrs object `vo` to named format `fmt` + """Translate a VRS object to expressions in the requested format. + + :param vo: VRS variation to translate. + :param fmt: Output format. + :param kwargs: Options forwarded to the selected exporter. Unused options + are ignored. + + For AlleleTranslator HGVS and SPDI exports: - kwargs: - ref_seq_limit Optional(int): - If vo.state is a ReferenceLengthExpression, and `ref_seq_limit` is specified, and `fmt` is `spdi`, the reference sequence is included in the SPDI expression if it is below the limit Otherwise only the length of the reference sequence is included. If the limit is None, the reference sequence is always included. In all cases, the alt sequence is included. Default is 0 (never include reference sequence). - :raise NotImplementedError: If `fmt` is not supported + * ``namespace`` (str or None): Sequence identifier namespace. + Defaults to "refseq". None includes all available aliases. + + Additional options for AlleleTranslator SPDI exports: + + * ``ref_seq_limit`` (int or None): Maximum reference sequence length + to include in the SPDI expression. Longer reference sequences are + represented by their length. Defaults to 0; None always includes + the reference sequence. The alternate sequence is always included, + including when derived from a ReferenceLengthExpression. + :return: Expressions in the requested format. + :raises NotImplementedError: If the requested format is unsupported. + :raises ValueError: If the object cannot be represented in the requested format. """ try: t = self.from_vrs_translators[fmt] @@ -154,18 +198,13 @@ def from_vrs(self, vo: models._VariationBase, fmt: str, **kwargs) -> list[str]: return t(vo, **kwargs) - ############################################################################ - # INTERNAL + def from_vrs_dict(self, var: dict, **kwargs) -> models._VariationBase | None: # noqa: ARG002 + """Parse a VRS JSON dictionary into a VRS variation object. - @lazy_property - def hgvs_tools(self) -> HgvsTools: - """Instantiate and return an HgvsTools instance""" - return HgvsTools(self.data_proxy) - - def _from_vrs(self, var: dict, **kwargs) -> models._VariationBase | None: # noqa: ARG002 - """Convert from dict representation of VRS JSON to VRS object - - :raise pydantic.ValidationError: If `var` is not a valid VRS variation + :param var: Dictionary containing a VRS variation and its ``type``. + :param kwargs: Unused options accepted for format dispatch. + :return: Parsed variation, or None for non-mappings or missing ``type``. + :raises pydantic.ValidationError: If the dictionary is not a valid variation. """ if not isinstance(var, Mapping): return None @@ -173,6 +212,14 @@ def _from_vrs(self, var: dict, **kwargs) -> models._VariationBase | None: # noq return None return models.Variation.model_validate(var).root + ############################################################################ + # INTERNAL + + @lazy_property + def hgvs_tools(self) -> HgvsTools: + """Instantiate and return an HgvsTools instance""" + return HgvsTools(self.data_proxy) + class AlleleTranslator(_Translator): """Class for translating formats to and from VRS Alleles""" @@ -183,37 +230,43 @@ def __init__( default_assembly_name: str = "GRCh38", identify: bool = True, ) -> None: - """Initialize AlleleTranslator class""" + """Initialize translation to and from VRS Alleles. + + :param data_proxy: Sequence data provider used during translation. + :param default_assembly_name: Default assembly for gnomAD inputs. + :param identify: Whether to assign GA4GH identifiers to translated objects. + """ super().__init__(data_proxy, default_assembly_name, identify) self.to_vrs_translators = { - "gnomad": self._from_gnomad, - "hgvs": self._from_hgvs, - "spdi": self._from_spdi, - "vrs": self._from_vrs, + "gnomad": self.from_gnomad, + "hgvs": self.from_hgvs, + "spdi": self.from_spdi, + "vrs": self.from_vrs_dict, } self.from_vrs_translators = { - "hgvs": self._to_hgvs, - "spdi": self._to_spdi, + "hgvs": self.to_hgvs, + "spdi": self.to_spdi, } - def _create_allele(self, values: dict, **kwargs) -> models.Allele: - """Create an allele object with the given parameters. - - Args: - values (dict): The values to use for creating the allele object. - 'sequence_id' (str): The sequence identifier from the input - expression, used to validate `start` and `end`. - 'refget_accession' (str): The accession ID of the reference genome. - 'start' (int): The start position of the allele. - 'end' (int): The end position of the allele. - literal_sequence' (str): The literal sequence for the allele. - **kwargs: Additional keyword arguments. - - Returns: - models.Allele: The created allele object. - + def _create_allele( + self, + values: dict, + *, + do_normalize: bool = True, + rle_seq_limit: int | None | TranslationDefault = USE_INSTANCE_DEFAULT, + ) -> models.Allele: + """Build and post-process an Allele from parsed coordinates and sequence. + + :param values: Parsed ``sequence_id``, ``refget_accession``, ``start``, + ``end``, and ``literal_sequence`` values. + :param do_normalize: Whether to perform fully justified normalization (default True). + :param rle_seq_limit: Maximum RLE sequence length to include. + USE_INSTANCE_DEFAULT uses the instance limit; 0 excludes the sequence + and None includes it without a limit. + :return: Constructed Allele, normalized and identified as configured. + :raises DataProxyValidationError: If the location is outside the reference sequence. """ self.data_proxy.validate_location_bounds( values["sequence_id"], values["start"], values["end"] @@ -224,24 +277,33 @@ def _create_allele(self, values: dict, **kwargs) -> models.Allele: ) state = models.LiteralSequenceExpression(sequence=values["literal_sequence"]) allele = models.Allele(location=location, state=state) - return self._post_process_imported_allele(allele, **kwargs) + return self._post_process_imported_allele( + allele, do_normalize=do_normalize, rle_seq_limit=rle_seq_limit + ) - def _from_gnomad(self, gnomad_expr: str, **kwargs) -> models.Allele | None: + def from_gnomad( + self, + gnomad_expr: str, + *, + assembly_name: str | None = None, + require_validation: bool = True, + do_normalize: bool = True, + rle_seq_limit: int | None | TranslationDefault = USE_INSTANCE_DEFAULT, + **kwargs, # noqa: ARG002 + ) -> models.Allele | None: """Parse gnomAD-style VCF expression into VRS Allele - kwargs: - assembly_name (str): Assembly used for `gnomad_expr`. - require_validation (bool): If `True` then validation checks must pass in - order to return a VRS object. A `DataProxyValidationError` will be - raised if validation checks fail. If `False` then VRS object will be - returned even if validation checks fail. Defaults to `True`. - rle_seq_limit Optional(int): If RLE is set as the new state after - normalization, this sets the limit for the length of the `sequence`. - To exclude `sequence` from the response, set to 0. - For no limit, set to `None`. - Defaults value set in instance variable, `rle_seq_limit`. - do_normalize (bool): `True` if fully justified normalization should be - performed. `False` otherwise. Defaults to `True` + :param gnomad_expr: Variant in chr-pos-ref-alt format. + :param assembly_name: Assembly used for the input, or None for the instance default. + :param require_validation: Whether reference mismatches raise an error (default True). + :param do_normalize: Whether to perform fully justified normalization (default True). + :param rle_seq_limit: Maximum RLE sequence length to include. + USE_INSTANCE_DEFAULT uses the instance limit; 0 excludes the sequence + and None includes it without a limit. + :param kwargs: Unused options accepted for format dispatch. + :return: Translated Allele, or None if the expression or accession is unrecognized. + :raises DataProxyValidationError: If reference validation is required and fails, + or the location is outside the reference sequence. #>>> a = tlr.from_gnomad("1-55516888-G-GA") #>>> a.model_dump() @@ -270,7 +332,8 @@ def _from_gnomad(self, gnomad_expr: str, **kwargs) -> models.Allele | None: return None g = m.groupdict() - assembly_name = kwargs.get("assembly_name", self.default_assembly_name) + if assembly_name is None: + assembly_name = self.default_assembly_name sequence = assembly_name + ":" + g["chr"] refget_accession = self.data_proxy.derive_refget_accession(sequence) if not refget_accession: @@ -288,7 +351,7 @@ def _from_gnomad(self, gnomad_expr: str, **kwargs) -> models.Allele | None: start, end, ref, - require_validation=kwargs.get("require_validation", True), + require_validation=require_validation, ) values = { @@ -298,14 +361,30 @@ def _from_gnomad(self, gnomad_expr: str, **kwargs) -> models.Allele | None: "end": end, "literal_sequence": ins_seq, } - return self._create_allele(values, **kwargs) - - def _from_hgvs(self, hgvs_expr: str, **kwargs) -> models.Allele | None: - # Uncertain-range expressions (e.g. g.(A_B)_(C_D)del) have no faithful - # Allele representation — their endpoints aren't single positions. - # Reject here rather than inside HgvsTools so that helper layer stays - # unaware of sibling translator classes. CnvTranslator handles the - # uncertain-range del/dup path. + return self._create_allele( + values, do_normalize=do_normalize, rle_seq_limit=rle_seq_limit + ) + + def from_hgvs( + self, + hgvs_expr: str, + *, + do_normalize: bool = True, + rle_seq_limit: int | None | TranslationDefault = USE_INSTANCE_DEFAULT, + **kwargs, # noqa: ARG002 + ) -> models.Allele | None: + """Parse an HGVS expression into a VRS Allele. + + :param hgvs_expr: HGVS variant expression. + :param do_normalize: Whether to perform fully justified normalization (default True). + :param rle_seq_limit: Maximum RLE sequence length to include. + USE_INSTANCE_DEFAULT uses the instance limit; 0 excludes the sequence + and None includes it without a limit. + :param kwargs: Unused options accepted for format dispatch. + :return: Translated Allele, or None if the expression or accession is unrecognized. + :raises ValueError: If HGVS positions or edits cannot be represented as an Allele. + :raises DataProxyValidationError: If the location is outside the reference sequence. + """ sv = self.hgvs_tools.parse(hgvs_expr) if sv is not None and self.hgvs_tools.has_uncertain_range(sv): msg = ( @@ -316,20 +395,29 @@ def _from_hgvs(self, hgvs_expr: str, **kwargs) -> models.Allele | None: allele_values = self.hgvs_tools.extract_allele_values(hgvs_expr) if allele_values: - return self._create_allele(allele_values, **kwargs) + return self._create_allele( + allele_values, do_normalize=do_normalize, rle_seq_limit=rle_seq_limit + ) return None - def _from_spdi(self, spdi_expr: str, **kwargs) -> models.Allele | None: + def from_spdi( + self, + spdi_expr: str, + *, + do_normalize: bool = True, + rle_seq_limit: int | None | TranslationDefault = USE_INSTANCE_DEFAULT, + **kwargs, # noqa: ARG002 + ) -> models.Allele | None: """Parse SPDI expression in to a GA4GH Allele - kwargs: - rle_seq_limit Optional(int): If RLE is set as the new state after - normalization, this sets the limit for the length of the `sequence`. - To exclude `sequence` from the response, set to 0. - For no limit, set to `None`. - Defaults value set in instance variable, `rle_seq_limit`. - do_normalize (bool): `True` if fully justified normalization should be - performed. `False` otherwise. Defaults to `True` + :param spdi_expr: Variant in accession:position:deletion:insertion format. + :param do_normalize: Whether to perform fully justified normalization (default True). + :param rle_seq_limit: Maximum RLE sequence length to include. + USE_INSTANCE_DEFAULT uses the instance limit; 0 excludes the sequence + and None includes it without a limit. + :param kwargs: Unused options accepted for format dispatch. + :return: Translated Allele, or None if the expression or accession is unrecognized. + :raises DataProxyValidationError: If the location is outside the reference sequence. #>>> a = tlr.from_spdi("NC_000013.11:32936731:1:C") #>>> a.model_dump() @@ -378,18 +466,33 @@ def _from_spdi(self, spdi_expr: str, **kwargs) -> models.Allele | None: "literal_sequence": ins_seq, } - return self._create_allele(values, **kwargs) + return self._create_allele( + values, do_normalize=do_normalize, rle_seq_limit=rle_seq_limit + ) - def _to_hgvs( + def to_hgvs( self, vo: models.Allele, namespace: str | None = "refseq", **kwargs, # noqa: ARG002 ) -> list[str]: + """Generate HGVS expressions for a VRS Allele. + + :param vo: Allele to translate. + :param namespace: Sequence identifier namespace, or None for all aliases. + :param kwargs: Unused options accepted for format dispatch. + :return: HGVS expressions, or an empty list if no aliases are available. + :raises ValueError: If the object cannot be represented as HGVS. + """ return self.hgvs_tools.from_allele(vo, namespace) - def _to_spdi( - self, vo: models.Allele, namespace: str | None = "refseq", **kwargs + def to_spdi( + self, + vo: models.Allele, + namespace: str | None = "refseq", + *, + ref_seq_limit: int | None = 0, + **kwargs, # noqa: ARG002 ) -> list[str]: """Generate a *list* of SPDI expressions for VRS Allele. @@ -413,6 +516,14 @@ def _to_spdi( SPDI and VRS use identical normalization. The incoming Allele is expected to be normalized per VRS spec. + + :param vo: Allele to translate. + :param namespace: Sequence identifier namespace, or None for all aliases. + :param ref_seq_limit: Maximum reference sequence length to include (default 0). + None always includes the reference sequence. + :param kwargs: Unused options accepted for format dispatch. + :return: SPDI expressions, or an empty list if no aliases are available. + :raises ValueError: If the object cannot be represented as SPDI. """ sequence = f"ga4gh:{vo.location.get_refget_accession()}" aliases = self.data_proxy.translate_sequence_identifier(sequence, namespace) @@ -446,7 +557,6 @@ def _to_spdi( # Optionally allow using the length of the reference sequence # instead of the sequence itself. - ref_seq_limit = kwargs.get("ref_seq_limit", 0) if ref_seq_limit is not None and len(ref_seq) > int(ref_seq_limit): ref_seq = len(ref_seq) @@ -456,25 +566,29 @@ def _to_spdi( return spdi_exprs def _post_process_imported_allele( - self, allele: models.Allele, **kwargs + self, + allele: models.Allele, + *, + do_normalize: bool = True, + rle_seq_limit: int | None | TranslationDefault = USE_INSTANCE_DEFAULT, ) -> models.Allele: - """Provide common post-processing for imported Alleles IN-PLACE. - - :param allele: VRS Allele object - - kwargs: - rle_seq_limit Optional(int): If RLE is set as the new state after - normalization, this sets the limit for the length of the `sequence`. - To exclude `sequence` from the response, set to 0. - For no limit, set to `None`. - do_normalize (bool): `True` if fully justified normalization should be - performed. `False` otherwise. Defaults to `True` + """Normalize an imported Allele and assign identifiers as configured. + + :param allele: Allele to process. Identifiers are assigned in place. + :param do_normalize: Whether to perform fully justified normalization (default True). + :param rle_seq_limit: Maximum RLE sequence length to include. + USE_INSTANCE_DEFAULT uses the instance limit; 0 excludes the sequence + and None includes it without a limit. + :return: Processed Allele. Normalization may return a new object. """ - if kwargs.get("do_normalize", True): + if rle_seq_limit is USE_INSTANCE_DEFAULT: + rle_seq_limit = self.rle_seq_limit + + if do_normalize: allele = normalize( allele, self.data_proxy, - rle_seq_limit=kwargs.get("rle_seq_limit", self.rle_seq_limit), + rle_seq_limit=rle_seq_limit, ) if self.identify: @@ -493,24 +607,39 @@ def __init__( default_assembly_name: str = "GRCh38", identify: bool = True, ) -> None: - """Initialize CnvTranslator class""" + """Initialize translation from HGVS to VRS copy number objects. + + :param data_proxy: Sequence data provider used during translation. + :param default_assembly_name: Default assembly name. + :param identify: Whether to assign GA4GH identifiers to translated objects. + """ super().__init__(data_proxy, default_assembly_name, identify) self.to_vrs_translators = { - "hgvs": self._from_hgvs, + "hgvs": self.from_hgvs, } - def _from_hgvs( - self, hgvs_dup_del_expr: str, **kwargs + def from_hgvs( + self, + hgvs_dup_del_expr: str, + *, + copies: int | models.Range | None = None, + copy_change: models.CopyChange | None = None, + **kwargs, # noqa: ARG002 ) -> models.CopyNumberChange | models.CopyNumberCount | None: - """Parse hgvs into a VRS CNV Object - - kwargs: - copies: The number of copies to use. If provided will return a - CopyNumberCount - copy_change: Copy change. If not provided, default is EFO:0030067 for - deletions and EFO:0030070 for duplications + """Parse an HGVS deletion or duplication into a VRS copy number object. + + :param hgvs_dup_del_expr: HGVS deletion or duplication expression. + :param copies: Copy count or range, including zero. If supplied, returns + CopyNumberCount and takes precedence over ``copy_change``. + :param copy_change: Copy change, defaulting to LOSS for deletions and + GAIN for duplications when ``copies`` is omitted. + :param kwargs: Unused options accepted for format dispatch. + :return: CopyNumberCount if copies is supplied, otherwise CopyNumberChange, + or None if the expression or accession is unrecognized. + :raises ValueError: If the edit is not a deletion or duplication, or its + positions cannot be represented as a copy number location. + :raises DataProxyValidationError: If the location is outside the reference sequence. """ - # sv = self._get_parsed_hgvs(hgvs_dup_del_expr) sv = self.hgvs_tools.parse(hgvs_dup_del_expr) if not sv: return None @@ -531,11 +660,9 @@ def _from_hgvs( location = self.hgvs_tools.build_cnv_location(sv, refget_accession) self.data_proxy.validate_location_bounds(sv.ac, location.start, location.end) - copies = kwargs.get("copies") if copies is not None: cnv = models.CopyNumberCount(location=location, copies=copies) else: - copy_change = kwargs.get("copy_change") if not copy_change: copy_change = ( models.CopyChange.LOSS diff --git a/tests/extras/test_allele_translator.py b/tests/extras/test_allele_translator.py index 30dffd45..feb9f2f6 100644 --- a/tests/extras/test_allele_translator.py +++ b/tests/extras/test_allele_translator.py @@ -352,34 +352,35 @@ def test_from_trailing_input_rejected(tlr, expr, fmt): @pytest.mark.vcr -def test_from_gnomad(tlr): +def test_from_gnomad(tlr) -> None: + """Public gnomAD translation preserves validation and normalization behavior.""" do_normalize = False assert ( - tlr._from_gnomad(snv_inputs["gnomad"], do_normalize=do_normalize).model_dump( - exclude_none=True - ) + tlr.from_gnomad( + snv_inputs["gnomad"], do_normalize=do_normalize, unrelated_option=True + ).model_dump(exclude_none=True) == snv_output ) assert ( - tlr._from_gnomad(mito_inputs["gnomad"], do_normalize=do_normalize).model_dump( + tlr.from_gnomad(mito_inputs["gnomad"], do_normalize=do_normalize).model_dump( exclude_none=True ) == mito_output ) assert ( - tlr._from_gnomad( + tlr.from_gnomad( deletion_inputs["gnomad"], do_normalize=do_normalize ).model_dump(exclude_none=True) == gnomad_deletion_output ) assert ( - tlr._from_gnomad( + tlr.from_gnomad( insertion_inputs["gnomad"], do_normalize=do_normalize ).model_dump(exclude_none=True) == gnomad_insertion_output ) assert ( - tlr._from_gnomad( + tlr.from_gnomad( duplication_inputs["gnomad"], do_normalize=do_normalize ).model_dump(exclude_none=True) == duplication_output @@ -387,42 +388,42 @@ def test_from_gnomad(tlr): # do_normalize defaults to true assert ( - tlr._from_gnomad(snv_inputs["gnomad"]).model_dump(exclude_none=True) + tlr.from_gnomad(snv_inputs["gnomad"]).model_dump(exclude_none=True) == snv_output ) assert ( - tlr._from_gnomad(mito_inputs["gnomad"]).model_dump(exclude_none=True) + tlr.from_gnomad(mito_inputs["gnomad"]).model_dump(exclude_none=True) == mito_output ) assert ( - tlr._from_gnomad(deletion_inputs["gnomad"]).model_dump(exclude_none=True) + tlr.from_gnomad(deletion_inputs["gnomad"]).model_dump(exclude_none=True) == deletion_output_normalized ) assert ( - tlr._from_gnomad(insertion_inputs["gnomad"]).model_dump(exclude_none=True) + tlr.from_gnomad(insertion_inputs["gnomad"]).model_dump(exclude_none=True) == insertion_output ) assert ( - tlr._from_gnomad(duplication_inputs["gnomad"]).model_dump(exclude_none=True) + tlr.from_gnomad(duplication_inputs["gnomad"]).model_dump(exclude_none=True) == duplication_output_normalized ) - assert tlr._from_gnomad("17-83129587-GTTGWCACATGA-G") + assert tlr.from_gnomad("17-83129587-GTTGWCACATGA-G") # Test valid characters - assert tlr._from_gnomad( + assert tlr.from_gnomad( "7-2-ACGTURYKMSWBDHVN-ACGTURYKMSWBDHVN", require_validation=False ) # Invalid input. Ref does not match regex - assert not tlr._from_gnomad("13-32936732-helloworld-C") + assert not tlr.from_gnomad("13-32936732-helloworld-C") # Ref != Actual ref invalid_var = "13-32936732-G-C" error_msg = "Reference mismatch at GRCh38:13 position 32936731-32936732 (input gave 'G' but correct ref is 'C')" with pytest.raises(DataProxyValidationError) as e: - tlr._from_gnomad(invalid_var) + tlr.from_gnomad(invalid_var) assert str(e.value) == error_msg with pytest.raises(DataProxyValidationError) as e: @@ -430,93 +431,98 @@ def test_from_gnomad(tlr): assert str(e.value) == error_msg # require_validation set to False - assert tlr._from_gnomad(invalid_var, require_validation=False) + assert tlr.from_gnomad(invalid_var, require_validation=False) @pytest.mark.vcr -def test_from_hgvs(tlr): +def test_from_hgvs(tlr) -> None: + """Translate HGVS expressions directly through the public method.""" do_normalize = False assert ( - tlr._from_hgvs(snv_inputs["hgvs"], do_normalize=do_normalize).model_dump( - exclude_none=True - ) + tlr.from_hgvs( + snv_inputs["hgvs"], do_normalize=do_normalize, assembly_name="GRCh38" + ).model_dump(exclude_none=True) == snv_output ) assert ( - tlr._from_hgvs(mito_inputs["hgvs"], do_normalize=do_normalize).model_dump( + tlr.from_hgvs(mito_inputs["hgvs"], do_normalize=do_normalize).model_dump( exclude_none=True ) == mito_output ) assert ( - tlr._from_hgvs(deletion_inputs["hgvs"], do_normalize=do_normalize).model_dump( + tlr.from_hgvs(deletion_inputs["hgvs"], do_normalize=do_normalize).model_dump( exclude_none=True ) == deletion_output ) assert ( - tlr._from_hgvs(insertion_inputs["hgvs"], do_normalize=do_normalize).model_dump( + tlr.from_hgvs(insertion_inputs["hgvs"], do_normalize=do_normalize).model_dump( exclude_none=True ) == insertion_output ) assert ( - tlr._from_hgvs( - duplication_inputs["hgvs"], do_normalize=do_normalize - ).model_dump(exclude_none=True) + tlr.from_hgvs(duplication_inputs["hgvs"], do_normalize=do_normalize).model_dump( + exclude_none=True + ) == duplication_output ) @pytest.mark.vcr -def test_from_spdi(tlr): +def test_from_spdi(tlr) -> None: + """Translate SPDI expressions directly through the public method.""" do_normalize = False assert ( - tlr._from_spdi(snv_inputs["spdi"], do_normalize=do_normalize).model_dump( - exclude_none=True - ) + tlr.from_spdi( + snv_inputs["spdi"], do_normalize=do_normalize, require_validation=False + ).model_dump(exclude_none=True) == snv_output ) assert ( - tlr._from_spdi(mito_inputs["spdi"], do_normalize=do_normalize).model_dump( + tlr.from_spdi(mito_inputs["spdi"], do_normalize=do_normalize).model_dump( exclude_none=True ) == mito_output ) for spdi_del_expr in deletion_inputs["spdi"]: assert ( - tlr._from_spdi(spdi_del_expr, do_normalize=do_normalize).model_dump( + tlr.from_spdi(spdi_del_expr, do_normalize=do_normalize).model_dump( exclude_none=True ) == deletion_output ), spdi_del_expr for spdi_ins_expr in insertion_inputs["spdi"]: assert ( - tlr._from_spdi(spdi_ins_expr, do_normalize=do_normalize).model_dump( + tlr.from_spdi(spdi_ins_expr, do_normalize=do_normalize).model_dump( exclude_none=True ) == insertion_output ), spdi_ins_expr assert ( - tlr._from_spdi( - duplication_inputs["spdi"], do_normalize=do_normalize - ).model_dump(exclude_none=True) + tlr.from_spdi(duplication_inputs["spdi"], do_normalize=do_normalize).model_dump( + exclude_none=True + ) == duplication_output ) @pytest.mark.vcr -def test_to_spdi(tlr): +def test_to_spdi(tlr) -> None: + """Direct public exports agree with format dispatch.""" # do_normalize defaults to true spdiexpr = snv_inputs["spdi"] allele = tlr.to_vrs(spdiexpr, "spdi") to_spdi = tlr.from_vrs(allele, "spdi") + assert tlr.to_spdi(allele, unrelated_option=True) == to_spdi assert len(to_spdi) == 1 assert spdiexpr == to_spdi[0] @pytest.mark.vcr -def test_to_spdi_with_ref(tlr): +def test_to_spdi_with_ref(tlr) -> None: + """Direct and dispatched SPDI exports honor the reference sequence limit.""" spdi_expr_no_ref = "NC_000019.10:44908821:1:T" spdi_expr_with_ref = "NC_000019.10:44908821:C:T" @@ -529,6 +535,7 @@ def test_to_spdi_with_ref(tlr): assert spdi_expr_no_ref == to_spdi_no_ref[0] to_spdi_with_ref = tlr.from_vrs(allele_with_ref, "spdi", ref_seq_limit=None) + assert tlr.to_spdi(allele_with_ref, ref_seq_limit=None) == to_spdi_with_ref assert len(to_spdi_with_ref) == 1 assert spdi_expr_with_ref == to_spdi_with_ref[0] @@ -734,19 +741,21 @@ def test_to_spdi_with_ref(tlr): @pytest.mark.parametrize(("hgvsexpr", "expected"), hgvs_tests) @pytest.mark.vcr -def test_hgvs(tlr, hgvsexpr, expected): +def test_hgvs(tlr, hgvsexpr, expected) -> None: + """Direct public exports agree with format dispatch.""" # do_normalize defaults to true allele = tlr.to_vrs(hgvsexpr, "hgvs") assert allele.model_dump(exclude_none=True) == expected to_hgvs = tlr.from_vrs(allele, "hgvs") + assert tlr.to_hgvs(allele, ref_seq_limit=0) == to_hgvs assert (hgvsexpr in to_hgvs) or ( hgvs_tests_to_hgvs_map.get(hgvsexpr, hgvsexpr) in to_hgvs ) @pytest.mark.vcr -def test_rle_seq_limit(tlr): +def test_rle_seq_limit(tlr, monkeypatch) -> None: """Test that for an ReferenceLengthExpression over 50bp, the sequence is not returned when rle_seq_limit is set to 50bp, but is included in the state when rle_seq_limit is set to None. @@ -788,6 +797,15 @@ def test_rle_seq_limit(tlr): ) assert allele_with_seq.model_dump(exclude_none=True) == a_dict_with_seq + # Omitted limits use the current instance setting; explicit None stays unlimited. + monkeypatch.setattr(tlr, "rle_seq_limit", None) + assert tlr.from_hgvs(input_hgvs_expr).state.sequence is not None + assert tlr.from_hgvs(input_hgvs_expr, rle_seq_limit=0).state.sequence is None + + monkeypatch.setattr(tlr, "rle_seq_limit", 0) + assert tlr.from_hgvs(input_hgvs_expr).state.sequence is None + assert tlr.from_hgvs(input_hgvs_expr, rle_seq_limit=None).state.sequence is not None + output_hgvs_expr = tlr.from_vrs(allele_with_seq, "hgvs") assert output_hgvs_expr == [input_hgvs_expr] @@ -815,14 +833,14 @@ def test_to_hgvs_iri_ref_keyerror(tlr): @pytest.mark.vcr -def test_reference_allele_rle(tlr): +def test_reference_allele_rle(tlr) -> None: """Test that reference alleles (REF==ALT) are normalized to ReferenceLengthExpression. Added to address https://github.com/ga4gh/vrs-python/issues/587 """ # Test with gnomad format gnomad_ref_allele = "1-100210778-AA-AA" - allele = tlr._from_gnomad(gnomad_ref_allele) + allele = tlr.from_gnomad(gnomad_ref_allele) expected = { "type": "Allele", @@ -847,7 +865,7 @@ def test_reference_allele_rle(tlr): # Test with SPDI format (REF==ALT) spdi_ref_allele = "NC_000001.11:100210777:AA:AA" - allele_spdi = tlr._from_spdi(spdi_ref_allele) + allele_spdi = tlr.from_spdi(spdi_ref_allele) assert allele_spdi.model_dump(exclude_none=True) == expected @@ -972,7 +990,8 @@ def test_translate_to_invalid_fmt(tlr): tlr.from_vrs(models.Allele.model_validate(snv_output), fmt="gnomad") -def test_from_vrs_dict(): +@pytest.mark.parametrize("direct", [False, True], ids=["dispatch", "direct"]) +def test_from_vrs_dict(direct: bool) -> None: """Regression test for ga4gh/vrs-python#489. Translating a VRS dict must resolve the model class from its `type` @@ -981,15 +1000,17 @@ def test_from_vrs_dict(): tlr = AlleleTranslator(data_proxy=None, identify=False) # valid VRS dict translates to a VRS object (previously raised TypeError) - allele = tlr.to_vrs(snv_output, fmt="vrs") + allele = ( + tlr.from_vrs_dict(snv_output) if direct else tlr.to_vrs(snv_output, fmt="vrs") + ) assert isinstance(allele, models.Allele) assert allele.type == "Allele" assert allele.location.start == snv_output["location"]["start"] assert allele.location.end == snv_output["location"]["end"] # non-dict and missing-type inputs still return None - assert tlr._from_vrs("NC_000019.10:g.44908822C>T") is None - assert tlr._from_vrs({"location": {}}) is None + assert tlr.from_vrs_dict("NC_000019.10:g.44908822C>T") is None + assert tlr.from_vrs_dict({"location": {}}) is None from_vrs_non_variation_type_cases = [ @@ -1004,24 +1025,28 @@ def test_from_vrs_dict(): @pytest.mark.parametrize( "case", from_vrs_non_variation_type_cases, ids=lambda c: c["id"] ) -def test_from_vrs_non_variation_type(case): +@pytest.mark.parametrize("direct", [False, True], ids=["dispatch", "direct"]) +def test_from_vrs_non_variation_type(case: dict, direct: bool) -> None: """A VRS dict whose `type` is not a VRS variation class is rejected, rather than resolved to whatever `ga4gh.vrs.models` attribute has that name """ tlr = AlleleTranslator(data_proxy=None, identify=False) + parse = tlr.from_vrs_dict if direct else tlr.to_vrs + options = {} if direct else {"fmt": "vrs"} + with pytest.raises(ValidationError) as e: - tlr.to_vrs({"type": case["type"]}, fmt="vrs") + parse({"type": case["type"]}, **options) assert e.value.errors()[0]["type"] == "union_tag_invalid" -def test_from_hgvs_uncertain_range_raises(tlr): +def test_from_hgvs_uncertain_range_raises(tlr) -> None: """Uncertain-range HGVS cannot be represented as an Allele with a literal sequence state; use CnvTranslator instead. See issue #609. No cassette needed: the rejection happens before any data-proxy call. """ with pytest.raises(ValueError, match="Uncertain-range"): - tlr._from_hgvs("NC_000019.9:g.(11211022_11213339)_(11217364_11218067)dup") + tlr.from_hgvs("NC_000019.9:g.(11211022_11213339)_(11217364_11218067)dup") @pytest.mark.vcr diff --git a/tests/extras/test_cnv_translator.py b/tests/extras/test_cnv_translator.py index abfe8957..fa0fb5f4 100644 --- a/tests/extras/test_cnv_translator.py +++ b/tests/extras/test_cnv_translator.py @@ -103,9 +103,9 @@ def tlr(rest_dataproxy): @pytest.mark.parametrize(("hgvsexpr", "copy_change", "expected"), from_hgvs_cx_tests) @pytest.mark.vcr -def test_from_hgvs_cx(tlr, hgvsexpr, copy_change, expected): - """Test that _from_hgvs works correctly for copy number change""" - cx = tlr._from_hgvs(hgvsexpr, copy_change=copy_change) +def test_from_hgvs_cx(tlr, hgvsexpr, copy_change, expected) -> None: + """Test that from_hgvs works correctly for copy number change""" + cx = tlr.from_hgvs(hgvsexpr, copy_change=copy_change) assert cx.model_dump(exclude_none=True) == expected @@ -157,31 +157,33 @@ def test_from_hgvs_cx(tlr, hgvsexpr, copy_change, expected): @pytest.mark.parametrize(("hgvsexpr", "copies", "expected"), from_hgvs_cn_tests) @pytest.mark.vcr -def test_from_hgvs_cn(tlr, hgvsexpr, copies, expected): - """Test that _from_hgvs works correctly for copy number count""" - cn = tlr._from_hgvs(hgvsexpr, copies=copies) +def test_from_hgvs_cn(tlr, hgvsexpr, copies, expected) -> None: + """Test that from_hgvs works correctly for copy number count""" + cn = tlr.from_hgvs(hgvsexpr, copies=copies) assert cn.model_dump(exclude_none=True) == expected @pytest.mark.vcr -def test_from_hgvs_cn_copies_zero(tlr): +def test_from_hgvs_cn_copies_zero(tlr) -> None: """Test that copies=0 produces CopyNumberCount, not CopyNumberChange. copies=0 is a valid input (homozygous deletion), but 0 is falsy in Python so it was previously treated as missing and fell through to CopyNumberChange. """ - cn = tlr._from_hgvs("NC_000013.11:g.26440969_26443305del", copies=0) + cn = tlr.from_hgvs( + "NC_000013.11:g.26440969_26443305del", copies=0, do_normalize=False + ) assert cn.type == "CopyNumberCount" assert cn.copies == 0 @pytest.mark.vcr -def test_from_hgvs_coding_coordinates(tlr): +def test_from_hgvs_coding_coordinates(tlr) -> None: """CDS-relative (``c.``) inputs are converted to transcript-relative coordinates via ``c_to_n`` before the VRS location is built; without the conversion the resulting ``SequenceLocation`` would point into the 5' UTR. """ - cx = tlr._from_hgvs("NM_001331029.1:c.100_200del") + cx = tlr.from_hgvs("NM_001331029.1:c.100_200del") assert cx.model_dump(exclude_none=True) == { "id": "ga4gh:CX.lJiuo6QPsrSEKI7EeX0xpX5Iqx0R_Kal", "type": "CopyNumberChange", @@ -296,9 +298,9 @@ def test_from_hgvs_coding_coordinates(tlr): @pytest.mark.parametrize(("hgvsexpr", "expected"), from_hgvs_uncertain_range_tests) @pytest.mark.vcr -def test_from_hgvs_uncertain_range(tlr, hgvsexpr, expected): +def test_from_hgvs_uncertain_range(tlr, hgvsexpr, expected) -> None: """Test uncertain-range HGVS expressions produce SequenceLocations with Range start/end (VRS 2.x). """ - cx = tlr._from_hgvs(hgvsexpr) + cx = tlr.from_hgvs(hgvsexpr) assert cx.model_dump(exclude_none=True) == expected diff --git a/tests/test_hgvs_uncertain_ranges.py b/tests/test_hgvs_uncertain_ranges.py index 4a83545b..eb722cb0 100644 --- a/tests/test_hgvs_uncertain_ranges.py +++ b/tests/test_hgvs_uncertain_ranges.py @@ -314,7 +314,7 @@ def tools(self): def test_uncertain_range_returns_range_for_del(self, hgvs_parser, tools): """Uncertain-range rejection is a translator-layer concern (see - :meth:`AlleleTranslator._from_hgvs`), not a helper-layer one. At the + :meth:`AlleleTranslator.from_hgvs`), not a helper-layer one. At the helper level, passing an uncertain-range ``sv`` directly returns :class:`models.Range` values for ``start`` and ``end`` — quiet but deterministic, and the boundary is locked by this test.