Source code for dae.variants_loaders.raw.loader

"""Base classes and helpers for variant loaders."""
from __future__ import annotations

import argparse
import copy
import logging
import os
import pathlib
import time
from abc import ABC, abstractmethod
from import Generator, Sequence
from enum import Enum
from typing import (

import numpy as np
import pandas as pd

from dae.annotation.annotation_pipeline import AnnotationPipeline, AttributeInfo
from dae.effect_annotation.effect import AlleleEffects
from dae.genomic_resources.reference_genome import ReferenceGenome
from dae.pedigrees.families_data import FamiliesData
from import Family
from dae.utils.variant_utils import best2gt, get_locus_ploidy
from dae.variants.attributes import GeneticModel, Sex, TransmissionType
from dae.variants.family_variant import (
from dae.variants.variant import SummaryVariant

logger = logging.getLogger(__name__)

[docs]class ArgumentType(Enum): ARGUMENT = 1 OPTION = 2
[docs]class CLIArgument: """Defines class for handling CLI arguments in variant loaders. This class handles the logic for CLI argument operations such as parsing arguments, transforming to dict, transforming a parsed argument back to a CLI argument and adding itself to an existing ArgumentParser. Construction closely mirrors the ArgumentParser argument format. """ def __init__( self, argument_name: str, has_value: bool = True, default_value: Optional[Union[int, str, bool]] = None, destination: Optional[str] = None, help_text: Optional[str] = None, action: Optional[str] = None, value_type: Optional[Type[str]] = None, metavar: Optional[str] = None, nargs: Optional[str] = None, raw: bool = False, ) -> None: self.argument_name = argument_name self.has_value = has_value self.default_value = default_value self.destination = destination self.value_type = value_type self.metavar = metavar self.help_text = help_text self.nargs = nargs self.action = action self.raw = raw self.arg_type = ArgumentType.OPTION if destination is None: self.destination = self._default_destination() def __repr__(self) -> str: return f"{self.argument_name} ({self.arg_type})" def _default_destination(self) -> Optional[str]: if self.argument_name.startswith("--"): self.arg_type = ArgumentType.OPTION else: self.arg_type = ArgumentType.ARGUMENT return None return self.argument_name[2:].replace("-", "_")
[docs] def add_to_parser(self, parser: argparse.ArgumentParser) -> None: """Add this argument to argsparser.""" kwargs = { "type": self.value_type, "help": self.help_text, "default": self.default_value, } if self.arg_type == ArgumentType.OPTION: kwargs["dest"] = self.destination if self.action: # TODO: # For some reason kwargs["type"] = self.value_type gets tuple-ized # should find a different workaround del kwargs["type"] kwargs["action"] = self.action else: kwargs["metavar"] = self.metavar kwargs["nargs"] = self.nargs parser.add_argument(self.argument_name, **kwargs) # type: ignore
[docs] def build_option( self, params: dict, use_defaults: bool = False, ) -> Optional[str]: """Build an option.""" if self.arg_type == ArgumentType.ARGUMENT: return None for key, value in params.items(): if key == self.destination: if self.has_value: if value is not None: if value == self.default_value: continue if self.raw: value = value.encode("unicode-escape")\ .decode().replace("\\\\", "\\") return f'{self.argument_name} "{value}"' if use_defaults and self.default_value is not None: value = self.default_value if self.raw: value = value.encode("unicode-escape")\ .decode().replace("\\\\", "\\") return f'{self.argument_name} "{value}"' else: return f"{self.argument_name}" return None
[docs] def parse_cli_argument( self, argv: argparse.Namespace, use_defaults: bool = False, ) -> None: """Parse the command line argument from the `argv` object. Args: argv (argparse.Namespace): The command line arguments. use_defaults (bool, optional): Whether to use default values if the argument is None. Defaults to False. """ if self.destination not in argv: # type: ignore return assert self.destination is not None argument = getattr(argv, self.destination) if argument is None: if self.default_value is not None and use_defaults: setattr(argv, self.destination, self.default_value)
FamilyGenotypeIterator = Generator[ tuple[Family, np.ndarray, Optional[np.ndarray]], None, None]
[docs]class FamiliesGenotypes(ABC): """A base class for family genotypes.""" def __init__(self) -> None: pass
[docs] @abstractmethod def family_genotype_iterator(self) -> FamilyGenotypeIterator: pass
[docs]class CLILoader(ABC): """Base class for loader classes that require cli arguments.""" def __init__( self, params: Optional[dict[str, Any]] = None, ) -> None: self.arguments = self._arguments() self.params: Dict[str, Any] = params if params else {} def _add_argument(self, argument: CLIArgument) -> None: self.arguments.append(argument) @classmethod def _arguments(cls) -> List[CLIArgument]: return []
[docs] @classmethod def cli_defaults(cls) -> dict[str, Any]: """Build a dictionary with default values for CLI arguments.""" argument_destinations = [ arg.destination for arg in cls._arguments() if arg.destination is not None ] defaults = [ arg.default_value for arg in cls._arguments() if arg.destination is not None ] return dict(zip(argument_destinations, defaults))
[docs] @classmethod def cli_arguments( cls, parser: argparse.ArgumentParser, options_only: bool = False, ) -> None: """Add command-line arguments specific for the CLILoader class. Args: parser (argparse.ArgumentParser): The ArgumentParser object to add the arguments to. options_only (bool, optional): If True, only adds options (not arguments) to the parser. Defaults to False. """ for argument in cls._arguments(): if options_only and argument.arg_type == ArgumentType.ARGUMENT: continue argument.add_to_parser(parser)
[docs] @classmethod def build_cli_arguments(cls, params: dict) -> str: """Return a string with cli arguments.""" built_arguments = [] for argument in cls._arguments():"adding to CLI arguments: %s", argument) built_arguments.append(argument.build_option(params)) nonnull_arguments = (x for x in built_arguments if x is not None) result = " ".join(nonnull_arguments)"result CLI arguments: %s", result) return result
[docs] def build_arguments_dict(self) -> Dict[str, Union[str, bool]]: """Build a dictionary with the argument destinations as keys.""" result = {} for argument in self._arguments(): if argument.arg_type == ArgumentType.ARGUMENT: continue if argument.destination in self.params: result[argument.destination] = \ self.params[argument.destination] logger.debug( "building arguments from %s into dict: %s", self.params, result) return result
[docs] @classmethod def parse_cli_arguments( cls, argv: argparse.Namespace, use_defaults: bool = False, ) -> Tuple[List[str], Dict[str, Any]]: """Parse cli arguments.""" for arg in cls._arguments(): arg.parse_cli_argument(argv, use_defaults=use_defaults) return [], {}
FullVariantsIterator = Generator[ tuple[SummaryVariant, list[FamilyVariant]], None, None] FamilyVariantsIterator = Generator[ FamilyVariant, None, None]
[docs]class VariantsLoader(CLILoader): """Base class for all variant loaders.""" def __init__( self, families: FamiliesData, filenames: Union[str, list[str]], genome: ReferenceGenome, transmission_type: TransmissionType = TransmissionType.transmitted, params: Optional[Dict[str, Any]] = None, attributes: Optional[Dict[str, Any]] = None, ) -> None: params = params if params else {} super().__init__(params=params) assert isinstance(families, FamiliesData) self.families = families if isinstance(filenames, str): filenames = [filenames] self.filenames = filenames assert isinstance(transmission_type, TransmissionType) self.transmission_type = transmission_type self.genome = genome if attributes is None: self._attributes = {} else: self._attributes = copy.deepcopy(attributes) self.arguments = [] @property def variants_filenames(self) -> list[str]: return self.filenames
[docs] def get_attribute(self, key: str) -> Any: return self._attributes.get(key, None)
[docs] def set_attribute(self, key: str, value: Any) -> None: self._attributes[key] = value
[docs] def reset_regions(self, regions: Optional[Union[str, list[str]]]) -> None: pass
@property def annotation_schema(self) -> Optional[list[AttributeInfo]]: return None @classmethod def _arguments(cls) -> list[CLIArgument]: return []
[docs] @abstractmethod def full_variants_iterator(self) -> FullVariantsIterator: pass
[docs] def family_variants_iterator(self) -> FamilyVariantsIterator: for _, fvs in self.full_variants_iterator(): for fvariant in fvs: yield fvariant
[docs] @abstractmethod def close(self) -> None: """Close resources used by the loader."""
@property @abstractmethod def chromosomes(self) -> list[str]: """Return list of all chromosomes."""
[docs]class VariantsLoaderDecorator(VariantsLoader): """Base class for wrapping and decoring a variant loader.""" def __init__(self, variants_loader: VariantsLoader) -> None: super().__init__( variants_loader.families, variants_loader.filenames, genome=variants_loader.genome, transmission_type=variants_loader.transmission_type, params=variants_loader.params, attributes=variants_loader._attributes, ) self.variants_loader = variants_loader
[docs] def get_attribute(self, key: str) -> Any: result = self._attributes.get(key, None) if result is not None: return result return self.variants_loader.get_attribute(key)
def __getattr__(self, attr: str) -> Any: return getattr(self.variants_loader, attr, None) @property def annotation_schema(self) -> Optional[list[AttributeInfo]]: return self.variants_loader.annotation_schema
[docs] @classmethod def build_cli_arguments(cls, params: dict) -> str: return cls.variants_loader.build_cli_arguments(params)
[docs] @classmethod def cli_defaults(cls) -> dict[str, Any]: return cls.variants_loader.cli_defaults()
[docs] @classmethod def cli_arguments( cls, parser: argparse.ArgumentParser, options_only: bool = False, ) -> None: cls.variants_loader.cli_arguments(parser, options_only)
[docs] def build_arguments_dict(self) -> dict: return self.variants_loader.build_arguments_dict()
[docs] def close(self) -> None: self.variants_loader.close()
@property def chromosomes(self) -> list[str]: return self.variants_loader.chromosomes
[docs]class AnnotationDecorator(VariantsLoaderDecorator): """Base class for annotators.""" SEP1 = "!" SEP2 = "|" SEP3 = ":" CLEAN_UP_COLUMNS = set( [ "alternatives_data", "family_variant_index", "family_id", "variant_sexes", "variant_roles", "variant_inheritance", "variant_in_member", "genotype_data", "best_state_data", "genetic_model_data", "inheritance_data", "frequency_data", "genomic_scores_data", "effect_type", "effect_gene", ], )
[docs] @staticmethod def build_annotation_filename(filename: str) -> str: """Return the corresponding annotation file for filename.""" path = pathlib.Path(filename) suffixes = path.suffixes if not suffixes: return f"{filename}-eff.txt" suffix = "".join(suffixes) replace_with = suffix.replace(".", "-") filename = filename.replace(suffix, replace_with) return f"{filename}-eff.txt"
[docs] @staticmethod def has_annotation_file(variants_filename: str) -> bool: annotation_filename = AnnotationDecorator\ .build_annotation_filename(variants_filename) return os.path.exists(annotation_filename)
[docs] @staticmethod def save_annotation_file( variants_loader: AnnotationPipelineDecorator, filename: str, sep: str = "\t", ) -> None: """Save annotation file.""" # def convert_array_of_strings_to_string(a): # if not a: # return None # return AnnotationDecorator.SEP1.join(a) common_columns = [ "chrom", "position", "reference", "alternative", "bucket_index", "summary_variant_index", "allele_index", "allele_count", ] if variants_loader.annotation_schema is not None: other_columns = list( filter( lambda name: name not in common_columns and name not in AnnotationDecorator.CLEAN_UP_COLUMNS, [ for attr in variants_loader.annotation_schema], ), ) else: other_columns = [] header = common_columns.copy() header.extend(["effects"]) header.extend(other_columns) with open(filename, "w", encoding="utf8") as outfile: outfile.write(sep.join(header)) outfile.write("\n") for summary_variant, _ in variants_loader.full_variants_iterator(): for allele_index, summary_allele in enumerate( summary_variant.alleles): line = [] rec = summary_allele.attributes rec["allele_index"] = allele_index line_values = [ *[rec.get(col, "") for col in common_columns], summary_allele.effects, *[rec.get(col, "") for col in other_columns], ] for value in line_values: line.append(str(value) if value is not None else "") outfile.write(sep.join(line)) outfile.write("\n")
[docs]class EffectAnnotationDecorator(AnnotationDecorator): """Annotate variants with an effect.""" def __init__( self, variants_loader: VariantsLoader, effect_annotator: AnnotationPipeline, ): super().__init__(variants_loader) self.set_attribute( "extra_attributes", variants_loader.get_attribute("extra_attributes"), ) self.effect_annotator = effect_annotator
[docs] def full_variants_iterator(self) -> FullVariantsIterator: for (summary_variant, family_variants) in \ self.variants_loader.full_variants_iterator(): for sallele in summary_variant.alt_alleles: context: Dict[str, Any] = {} attributes = self.effect_annotator.annotate( sallele.get_annotatable(), context) assert "allele_effects" in attributes, attributes allele_effects = attributes["allele_effects"] assert isinstance(allele_effects, AlleleEffects), \ (type(allele_effects), allele_effects) # pylint: disable=protected-access sallele._effects = allele_effects yield summary_variant, family_variants
[docs] def close(self) -> None: self.effect_annotator.close() super().close()
[docs]class AnnotationPipelineDecorator(AnnotationDecorator): """Annotate variants by processing them through an annotation pipeline.""" def __init__( self, variants_loader: VariantsLoader, annotation_pipeline: AnnotationPipeline, ) -> None: super().__init__(variants_loader) self.annotation_pipeline = annotation_pipeline logger.debug( "creating annotation pipeline decorator with " "annotation pipeline: %s", annotation_pipeline.get_attributes()) self.set_attribute("annotation_schema", self.annotation_schema) self.set_attribute( "extra_attributes", variants_loader.get_attribute("extra_attributes"), ) @property def annotation_schema(self) -> list[AttributeInfo]: return self.annotation_pipeline.get_attributes()
[docs] def full_variants_iterator( self, ) -> FullVariantsIterator: with as annotation_pipeline: internal_attributes = set( for attr in self.annotation_pipeline.get_attributes() if attr.internal ) for (summary_variant, family_variants) in \ self.variants_loader.full_variants_iterator(): for sallele in summary_variant.alt_alleles: attributes = annotation_pipeline.annotate( sallele.get_annotatable()) if "allele_effects" in attributes: allele_effects = attributes["allele_effects"] assert isinstance(allele_effects, AlleleEffects), \ attributes # pylint: disable=protected-access sallele._effects = allele_effects del attributes["allele_effects"] public_attributes = { key: value for key, value in attributes.items() if key not in internal_attributes } sallele.update_attributes(public_attributes) yield summary_variant, family_variants
[docs] def close(self) -> None: self.annotation_pipeline.close()
[docs]class StoredAnnotationDecorator(AnnotationDecorator): """Annotate variant using a stored annotator.""" def __init__( self, variants_loader: VariantsGenotypesLoader, annotation_filename: str, ) -> None: super().__init__(variants_loader) assert os.path.exists(annotation_filename) self.annotation_filename = annotation_filename
[docs] @staticmethod def decorate( variants_loader: VariantsGenotypesLoader, source_filename: str, ) -> VariantsLoader: """Wrap variants_loader into a StoredAnnotationDecorator.""" annotation_filename = StoredAnnotationDecorator \ .build_annotation_filename( source_filename, ) if not os.path.exists(annotation_filename): logger.warning("stored annotation missing %s", annotation_filename) return variants_loader return StoredAnnotationDecorator( variants_loader, annotation_filename, )
@classmethod def _convert_array_of_strings(cls, token: str) -> Optional[list[str]]: if not token: return None token = token.strip() words = [w.strip() for w in token.split(cls.SEP1)] return words @staticmethod def _convert_string(token: str) -> Optional[str]: if not token: return None return token @classmethod def _load_annotation_file( cls, filename: str, sep: str = "\t", ) -> pd.DataFrame: assert os.path.exists(filename) with open(filename, "r", encoding="utf8") as infile: annot_df = pd.read_csv( infile, sep=sep, index_col=False, dtype={ "chrom": str, "position": np.int32, # 'effects': str, }, converters={ "cshl_variant": cls._convert_string, "effects": cls._convert_string, "effect_gene_genes": cls._convert_array_of_strings, "effect_gene_types": cls._convert_array_of_strings, "effect_details_transcript_ids": cls._convert_array_of_strings, "effect_details_details": cls._convert_array_of_strings, }, encoding="utf-8", ).replace({np.nan: None}) special_columns = set(annot_df.columns) & set( ["alternative", "effect_type"], ) for col in special_columns: annot_df[col] = ( annot_df[col] .astype(object) .where(pd.notnull(annot_df[col]), None) ) return annot_df
[docs] def full_variants_iterator( self, ) -> FullVariantsIterator: variant_iterator = self.variants_loader.full_variants_iterator() start = time.time() annot_df = self._load_annotation_file(self.annotation_filename) elapsed = time.time() - start "Storred annotation file (%s) loaded in in %.2f sec", self.annotation_filename, elapsed) start = time.time() records = annot_df.to_dict(orient="records") index = 0 while index < len(records): sumary_variant, family_variants = next( variant_iterator, (None, None)) if sumary_variant is None: return assert family_variants is not None variant_records = [] current_record = records[index] while current_record["summary_variant_index"] == \ sumary_variant.summary_index: variant_records.append(current_record) index += 1 if index >= len(records): break current_record = records[index] assert len(variant_records) > 0, sumary_variant for sallele in sumary_variant.alleles: sallele.update_attributes( variant_records[sallele.allele_index]) # type: ignore yield sumary_variant, family_variants elapsed = time.time() - start "Storred annotation file (%s) parsed in %.2f sec", self.annotation_filename, elapsed)
[docs]class VariantsGenotypesLoader(VariantsLoader): """Base class for variants loaders. Calculate missing best states and adds a genetic model value to the family variant and its alleles. """ def __init__( self, families: FamiliesData, filenames: Union[str, list[str]], genome: ReferenceGenome, transmission_type: TransmissionType = TransmissionType.transmitted, regions: Optional[List[str]] = None, expect_genotype: bool = True, expect_best_state: bool = False, params: Optional[Dict[str, Any]] = None) -> None: params = params if params else {} super().__init__( families=families, filenames=filenames, transmission_type=transmission_type, genome=genome, params=params) self.regions: Sequence[Optional[str]] if regions is None or isinstance(regions, str): self.regions = [regions] else: self.regions = regions self._adjust_chrom_prefix = lambda chrom: chrom self._unadjust_chrom_prefix = lambda chrom: chrom if params.get("add_chrom_prefix", None): self._chrom_prefix = params["add_chrom_prefix"] self._adjust_chrom_prefix = self._add_chrom_prefix self._unadjust_chrom_prefix = self._del_chrom_prefix elif params.get("del_chrom_prefix", None): self._chrom_prefix = params["del_chrom_prefix"] self._adjust_chrom_prefix = self._del_chrom_prefix self._unadjust_chrom_prefix = self._add_chrom_prefix self.expect_genotype = expect_genotype self.expect_best_state = expect_best_state @classmethod def _arguments(cls) -> list[CLIArgument]: arguments = super()._arguments() arguments.append(CLIArgument( "--add-chrom-prefix", value_type=str, help_text="Add specified prefix to each chromosome name in " "variants file", )) arguments.append(CLIArgument( "--del-chrom-prefix", value_type=str, help_text="Remove specified prefix to each chromosome name in " "variants file", )) return arguments @abstractmethod def _full_variants_iterator_impl(self) -> FullVariantsIterator: pass
[docs] def reset_regions(self, regions: Optional[Union[str, list[str]]]) -> None: if regions is None: self.regions = [] elif isinstance(regions, str): self.regions = [regions] else: self.regions = regions
@classmethod def _get_diploid_males(cls, family_variant: FamilyVariant) -> List[bool]: res = [] assert is not None assert == (2, len( for member_idx, member in enumerate(, ): if in (Sex.F, Sex.U): continue res.append(bool([1, member_idx] != -2)) return res @classmethod def _calc_genetic_model( cls, family_variant: FamilyVariant, genome: ReferenceGenome, ) -> GeneticModel: if family_variant.chromosome in ("X", "chrX"): male_ploidy = get_locus_ploidy( family_variant.chromosome, family_variant.position, Sex.M, genome, ) if male_ploidy == 2: if not all(cls._get_diploid_males(family_variant)): return GeneticModel.X_broken return GeneticModel.pseudo_autosomal if any(cls._get_diploid_males(family_variant)): return GeneticModel.X_broken return GeneticModel.X # We currently assume all other chromosomes are autosomal return GeneticModel.autosomal @classmethod def _calc_best_state( cls, family_variant: FamilyVariant, genome: ReferenceGenome, force: bool = True) -> Optional[np.ndarray]: assert is not None male_ploidy = get_locus_ploidy( family_variant.chromosome, family_variant.position, Sex.M, genome, ) if family_variant.chromosome in ("X", "chrX") and male_ploidy == 1: best_state = calculate_simple_best_state(, family_variant.allele_count, ) male_ids = [ person.person_id for person in if == Sex.M ] male_indices = for idx in male_indices: # A male with a haploid genotype for X cannot # have two alternative alleles, therefore there # must be one or two reference alleles left over # from the simple best state calculation if best_state[0, idx] in (1, 2): best_state[0, idx] -= 1 elif np.any(best_state[:, idx] == 2): best_state[best_state[:, idx] == 2, idx] -= 1 # else: # print( # "WARN: can't handle broken X best state:", # family_variant, mat2str(, # file=sys.stderr) return best_state if force: return calculate_simple_best_state(, family_variant.allele_count, ) return None @classmethod def _calc_genotype( cls, family_variant: FamilyVariant, genome: ReferenceGenome) -> Tuple[np.ndarray, GeneticModel]: # pylint: disable=protected-access best_state = family_variant._best_state assert best_state is not None genotype = best2gt(best_state) male_ploidy = get_locus_ploidy( family_variant.chromosome, family_variant.position, Sex.M, genome, ) ploidy = np.sum(best_state, 0) genetic_model = GeneticModel.autosomal if family_variant.chromosome in ("X", "chrX"): genetic_model = GeneticModel.X if male_ploidy == 2: genetic_model = GeneticModel.pseudo_autosomal male_ids = [ person_id for person_id, person in if == Sex.M ] male_indices = for idx in male_indices: if ploidy[idx] != male_ploidy: genetic_model = GeneticModel.X_broken break elif np.any(ploidy == 1): genetic_model = GeneticModel.autosomal_broken return genotype, genetic_model def _add_chrom_prefix(self, chrom: str) -> str: # there is an important invariant between this and _del_chrom_prefix # we don't know if this or _del_chrom_prefix will be executed first so # _add_chrom_prefix should undo _del_chrom_prefix # _del_chrom_prefix should undo _add_chrom_prefix assert self._chrom_prefix is not None return f"{self._chrom_prefix}{chrom}" def _del_chrom_prefix(self, chrom: str) -> str: assert self._chrom_prefix is not None assert chrom.startswith(self._chrom_prefix) return chrom[len(self._chrom_prefix):]
[docs] def full_variants_iterator( self, ) -> FullVariantsIterator: full_iterator = self._full_variants_iterator_impl() for summary_variant, family_variants in full_iterator: chrom = self._adjust_chrom_prefix(summary_variant.chromosome) # pylint: disable=protected-access summary_variant._chromosome = chrom for summary_allele in summary_variant.alleles: summary_allele._chrom = chrom summary_allele._attributes["chrom"] = chrom for family_variant in family_variants: if self.expect_genotype: assert family_variant._best_state is None assert family_variant._genetic_model is None assert is not None family_variant._genetic_model = self._calc_genetic_model( family_variant, self.genome, ) family_variant._best_state = self._calc_best_state( family_variant, self.genome, force=False, ) for allele in family_variant.alleles: fallele = cast(FamilyAllele, allele) fallele._best_state = family_variant.best_state fallele._genetic_model = family_variant.genetic_model elif self.expect_best_state and is None: assert family_variant._best_state is not None assert family_variant._genetic_model is None assert is None (, family_variant._genetic_model, ) = self._calc_genotype(family_variant, self.genome) for allele in family_variant.alleles: fallele = cast(FamilyAllele, allele) = fallele._genetic_model = family_variant.genetic_model yield summary_variant, family_variants