Source code for dae.variants_loaders.cnv.loader

"""Defines CNV loader class and helpers.

Copy Number Variants (CNV) loader :class:`CNVLoader`

This modules provides a class :class:`CNVLoader` to facilitate loading CNVs
specified in various input formats.

There are three groups of input parameters that could be configured
by the CNVLoader parameters:

- location of the variant - VCF-like vs CSHL-like of the variant position;

- variant genotype - list of person_ids vs CSHL-like family/best state
  description of the genotype for given family

- variant type - flexible CNV+/CNV- variant type description.

To configure the :class:`CNVLoader` you need to pass `params` dictionary
to the constructor of the class.

Parameters that are used to configure input data colums are:

Location of the CNVs

- `cnv_location` - column name, that is interpreted as variant

- `cnv_chrom` - column name, interpreted as the chromosome

- `cnv_start` - column name, interpreted as the start position of the CNVs

- `cnv_end` - column name, interpreted as the end position of the CNVs

Genotype of the CNVs

- `cnv_family_id` - column name, specifying the family for the CNVs

- `cnv_best_state` - column name, specifying the best state fore the CNVs

- `cnv_person_id` - column name, specifying a person, that has given CNV

Variant type for CNVs

- `cnv_variant_type` - column name, specifying the CNV variant type

- `cnv_plus_values` - list of the values in column `cnv_variant_type` that
  are interpreted as `CNV+`

- `cnv_minus_values` - list of values in column `cnv_variant_type` that are
  interpreted as `CNV-`

Additional parameters

Additional parameters, that configure the behavior of the :class:`CNVLoader`

- `cnv_sep` - separator character, that split columns in the lines of the
  input file

- `cnv_transmission_type` - the CNV loader is used mostly for importing
  de Novo variants. In rare cases when we use this loader to import
  transmitted CNV variants we should pass this parameter to specify
  that the varirants are not `denovo`.

import argparse
import logging
from import Generator
from copy import copy
from pathlib import Path
from typing import Any, Dict, List, Optional, TextIO, Tuple, Union

import pandas as pd

from dae.genomic_resources.reference_genome import ReferenceGenome
from dae.pedigrees.families_data import FamiliesData
from dae.utils.regions import Region
from dae.variants.attributes import Inheritance
from dae.variants.family_variant import FamilyVariant
from dae.variants.variant import SummaryVariant, SummaryVariantFactory
from dae.variants_loaders.cnv.flexible_cnv_loader import flexible_cnv_loader
from dae.variants_loaders.raw.loader import (

logger = logging.getLogger(__name__)

def _cnv_loader(
        filepath_or_buffer: Union[str, Path, TextIO],
        families: FamiliesData,
        genome: ReferenceGenome,
        cnv_chrom: Optional[str] = None,
        cnv_start: Optional[str] = None,
        cnv_end: Optional[str] = None,
        cnv_location: Optional[str] = None,
        cnv_person_id: Optional[str] = None,
        cnv_family_id: Optional[str] = None,
        cnv_best_state: Optional[str] = None,
        cnv_variant_type: Optional[str] = None,
        cnv_plus_values: Optional[List[str]] = None,
        cnv_minus_values: Optional[List[str]] = None,
        cnv_sep: str = "\t",
        **kwargs: Any) -> pd.DataFrame:
    """Flexible load of CNV variants.

    This function uses flexible variant loader infrastructure to
    load variants from a CNVs data input and transform them into a pandas
    # pylint: disable=too-many-arguments,too-many-locals"unexpected parameters passed to _cnv_loader: %s", kwargs)

    variant_generator = flexible_cnv_loader(

    data = []
    for record in variant_generator:

    df: pd.DataFrame = pd.DataFrame.from_records(
        data, columns=[
            "chrom", "pos", "pos_end",
            "family_id", "best_state",

    df = df.sort_values(
        by=["chrom", "pos", "pos_end"])

    df = df.rename(
            "pos": "position",
            "pos_end": "end_position",
    return df

[docs]class CNVLoader(VariantsGenotypesLoader): """Defines CNV loader class.""" def __init__( self, families: FamiliesData, cnv_filenames: list[Union[str, Path, TextIO]], genome: ReferenceGenome, regions: Optional[list[str]] = None, params: Optional[Dict[str, Any]] = None): if params is None: params = {} if params.get("cnv_transmission_type") == "denovo": transmission_type = TransmissionType.denovo else: transmission_type = TransmissionType.transmitted super().__init__( families=families, filenames=[str(fn) for fn in cnv_filenames], transmission_type=transmission_type, genome=genome, regions=regions, expect_genotype=False, expect_best_state=True, params=params, )"CNV loader params: %s", params) self.genome = genome self.set_attribute("source_type", "cnv") self.reset_regions(regions) assert isinstance(cnv_filenames, list) assert len(cnv_filenames) == 1 cnv_filename = cnv_filenames[0]"CNV loader params: %s", self.params) self.cnv_df = _cnv_loader( cnv_filename, families, genome, **self.params, ) self._init_chromosomes() def _init_chromosomes(self) -> None: self._chromosomes = list(self.cnv_df.chrom.unique()) self._chromosomes = [ self._adjust_chrom_prefix(chrom) for chrom in self._chromosomes ] all_chromosomes = self.genome.chromosomes if all(chrom in set(all_chromosomes) for chrom in self._chromosomes): self._chromosomes = sorted( self._chromosomes, key=all_chromosomes.index) @property def chromosomes(self) -> list[str]: return self._chromosomes @classmethod def _arguments(cls) -> list[CLIArgument]: arguments = super()._arguments() arguments.append(CLIArgument( "cnv_file", value_type=str, metavar="<variants filename>", help_text="cnv variants file", )) arguments.append(CLIArgument( "--cnv-location", value_type=str, default_value="location", help_text="The label or index of the" " column containing the CSHL-style" " location of the variant. [Default: location]", )) # arguments.append(CLIArgument( # "--cnv-chrom", # value_type=str, # default_value="chrom", # help_text="The label or index of the" # " column containing the chromosome" # " of the variant. [Default: chrom]", # )) # arguments.append(CLIArgument( # "--cnv-start", # value_type=str, # default_value="pos", # help_text="The label or index of the" # " column containing the start" # " of the CNV. [Default: pos]", # )) # arguments.append(CLIArgument( # "--cnv-end", # value_type=str, # default_value="pos_end", # help_text="The label or index of the" # " column containing the end" # " of the CNV variant. [Default: pos_end]", # )) arguments.append(CLIArgument( "--cnv-family-id", value_type=str, default_value="familyId", help_text="The label or index of the" " column containing family's ID." " [Default: familyId]", )) arguments.append(CLIArgument( "--cnv-best-state", value_type=str, default_value="bestState", help_text="The label or index of the" " column containing the variant's" " best state. [Default: bestState]", )) arguments.append(CLIArgument( "--cnv-person-id", value_type=str, help_text="The label or index of the" " column containing the ids of the people in which" " the variant is. [Default: None]", )) arguments.append(CLIArgument( "--cnv-variant-type", value_type=str, default_value="variant", help_text="The label or index of the" " column containing the variant's" " type. [Default: variant]", )) arguments.append(CLIArgument( "--cnv-plus-values", value_type=str, default_value="CNV+", help_text="The cnv+ value used in the columns containing" " the variant's type. [Default: CNV+]", )) arguments.append(CLIArgument( "--cnv-minus-values", value_type=str, default_value="CNV-", help_text="The cnv- value used in the columns containing" " the variant's type. [Default: CNV-]", )) arguments.append(CLIArgument( "--cnv-sep", value_type=str, default_value="\t", help_text="CNV file field separator. [Default: `\\t`]", )) arguments.append(CLIArgument( "--cnv-transmission-type", value_type=str, default_value="denovo", help_text="CNV transmission type. [Default: `denovo`]", )) return arguments
[docs] def reset_regions(self, regions: Optional[Union[str, list[str]]]) -> None: super().reset_regions(regions) result = [] for reg in self.regions: if reg is None: continue result.append(Region.from_str(reg)) self.regions = result # type: ignore
def _is_in_regions(self, summary_variant: SummaryVariant) -> bool: if len(self.regions) == 0: return True isin = [ r.isin( # type: ignore self._adjust_chrom_prefix(summary_variant.chrom), summary_variant.position, ) for r in self.regions if r is not None ] return any(isin)
[docs] def close(self) -> None: pass
def _full_variants_iterator_impl( self, ) -> Generator[Tuple[SummaryVariant, List[FamilyVariant]], None, None]: # pylint: disable=too-many-locals group = self.cnv_df.groupby( ["chrom", "position", "end_position", "variant_type"], sort=False).agg(list) for num_idx, (idx, values) in enumerate(group.iterrows()): chrom, position, end_position, variant_type = idx # type: ignore position = int(position) end_position = int(end_position) summary_rec = { "chrom": chrom, "reference": None, "alternative": None, "position": position, "end_position": end_position, "summary_variant_index": num_idx, "variant_type": variant_type, "allele_index": 0, } alt_rec = copy(summary_rec) del summary_rec["end_position"] del summary_rec["variant_type"] alt_rec["allele_index"] = 1 svar = SummaryVariantFactory.summary_variant_from_records( [summary_rec, alt_rec], self.transmission_type, ) if not self._is_in_regions(svar): continue fvs = [] extra_attributes_keys = filter( lambda x: x not in ["best_state", "family_id"], values.keys(), ) for f_idx, family_id in enumerate( values.get("family_id")): # type: ignore best_state = values.get("best_state")[f_idx] # type: ignore assert best_state is not None family = self.families.get(family_id) if family is None: continue fvar = FamilyVariant( svar, family, None, best_state) # type: ignore extra_attributes = {} for attr in extra_attributes_keys: attr_val = values.get(attr)[f_idx] # type: ignore extra_attributes[attr] = [attr_val] fvar.update_attributes(extra_attributes) fvs.append(fvar) yield svar, fvs
[docs] def full_variants_iterator(self) -> FullVariantsIterator: full_iterator = super().full_variants_iterator() for summary_variants, family_variants in full_iterator: for fvar in family_variants: for fallele in fvar.family_alt_alleles: if self.transmission_type == TransmissionType.denovo: inheritance = [ Inheritance.denovo if mem is not None else inh for inh, mem in zip( fallele.inheritance_in_members, fallele.variant_in_members, ) ] # pylint: disable=protected-access fallele._inheritance_in_members = inheritance yield summary_variants, family_variants
[docs] @classmethod def parse_cli_arguments( cls, argv: argparse.Namespace, use_defaults: bool = False, ) -> Tuple[List[str], Dict[str, Any]]: if argv.cnv_file is None: return [], {} return [argv.cnv_file], { "cnv_location": argv.cnv_location, "cnv_person_id": argv.cnv_person_id, "cnv_family_id": argv.cnv_family_id, "cnv_variant_type": argv.cnv_variant_type, "cnv_plus_values": argv.cnv_plus_values, "cnv_minus_values": argv.cnv_minus_values, "cnv_best_state": argv.cnv_best_state, "cnv_sep": argv.cnv_sep, "cnv_transmission_type": argv.cnv_transmission_type, "add_chrom_prefix": argv.add_chrom_prefix, "del_chrom_prefix": argv.del_chrom_prefix, }