studies package

Subpackages

Submodules

studies.query_transformer module

class studies.query_transformer.QueryTransformer(gene_scores_db: GeneScoresDb, chromosomes: list[str], chr_prefix: str)[source]

Bases: QueryTransformerProtocol

Transform genotype data query WEB parameters into query variants.

FILTER_RENAMES_MAP: ClassVar[dict[str, str]] = {'effectTypes': 'effect_types', 'familyIds': 'family_ids', 'genders': 'sexes', 'geneSymbols': 'genes', 'personIds': 'person_ids', 'regionS': 'regions', 'variantTypes': 'variant_type'}
extract_person_set_collection_query(study: WDAEStudy, kwargs: dict[str, Any]) PSCQuery[source]
get_unique_family_variants(kwargs: dict[str, Any]) bool[source]
transform_kwargs(study: WDAEStudy, **kwargs: Any) dict[str, Any][source]

Transform WEB query variants params into genotype data params.

Requires a study wrapper to handle study context specific arguments, such as person set collections and phenotype filters.

Returns None if the query is deemed empty.

studies.query_transformer.get_or_create_query_transformer(gpf_instance: GPFInstance) QueryTransformer[source]

Get or create query transformer singleton instance.

studies.query_transformer.make_query_transformer(gpf_instance: GPFInstance) QueryTransformer[source]

studies.response_transformer module

class studies.response_transformer.ResponseTransformer(gene_scores_db: GeneScoresDb | None)[source]

Bases: ResponseTransformerProtocol

Helper class to transform genotype browser response.

PHENOTYPE_ATTRS: ClassVar[dict[str, Callable]] = {'carrier_phenotypes': <function ResponseTransformer.<lambda>>, 'family_phenotypes': <function ResponseTransformer.<lambda>>}
SPECIAL_ATTRS: ClassVar[dict[str, Callable]] = {'alternative': <function ResponseTransformer.<lambda>>, 'best_st': <function ResponseTransformer.<lambda>>, 'carrier_person_attributes': <function ResponseTransformer.<lambda>>, 'carrier_person_ids': <function ResponseTransformer.<lambda>>, 'effect_details': <function ResponseTransformer.<lambda>>, 'effects': <function ResponseTransformer.<lambda>>, 'family': <function ResponseTransformer.<lambda>>, 'family_person_attributes': <function ResponseTransformer.<lambda>>, 'family_person_ids': <function ResponseTransformer.<lambda>>, 'family_structure': <function ResponseTransformer.<lambda>>, 'full_effect_details': <function ResponseTransformer.<lambda>>, 'genes': <function ResponseTransformer.<lambda>>, 'genotype': <function ResponseTransformer.<lambda>>, 'inheritance_type': <function ResponseTransformer.<lambda>>, 'is_denovo': <function ResponseTransformer.<lambda>>, 'location': <function ResponseTransformer.<lambda>>, 'position': <function ResponseTransformer.<lambda>>, 'raw_effects': <function ResponseTransformer.<lambda>>, 'reference': <function ResponseTransformer.<lambda>>, 'seen_in_affected': <function ResponseTransformer.<lambda>>, 'seen_in_unaffected': <function ResponseTransformer.<lambda>>, 'variant': <function ResponseTransformer.<lambda>>, 'worst_effect': <function ResponseTransformer.<lambda>>}
STREAMING_CHUNK_SIZE = 20
build_variant_row(study: WDAEStudy, v: SummaryVariant | FamilyVariant, column_descs: list[dict], **kwargs: str | None) list[source]

Construct response row for a variant.

transform_gene_view_summary_variant(variant: SummaryVariant, frequency_column: str) Generator[dict[str, Any], None, None][source]

Transform gene view summary response into dicts.

transform_gene_view_summary_variant_download(variants: Iterable[SummaryVariant], frequency_column: str, summary_variant_ids: set[str]) Iterator[str][source]

Transform gene view summary response into rows.

variant_transformer(study: WDAEStudy, pheno_values: dict[str, Any] | None) Callable[[FamilyVariant], FamilyVariant][source]

Build and return a variant transformer function.

studies.response_transformer.get_or_create_response_transformer(gpf_instance: GPFInstance) ResponseTransformer[source]

Get or create response transformer singleton instance.

studies.response_transformer.make_response_transformer(gpf_instance: GPFInstance) ResponseTransformer[source]
studies.response_transformer.members_in_order_get_family_structure(mio: list[Person]) str[source]

studies.study_wrapper module

class studies.study_wrapper.QueryTransformerProtocol(*args, **kwargs)[source]

Bases: Protocol

Protocol for query transformer interface.

abstract extract_person_set_collection_query(study: WDAEStudy, kwargs: dict[str, Any]) PSCQuery[source]
abstract get_unique_family_variants(kwargs: dict[str, Any]) bool[source]
abstract transform_kwargs(study: WDAEStudy, **kwargs: Any) dict[str, Any][source]
class studies.study_wrapper.ResponseTransformerProtocol(*args, **kwargs)[source]

Bases: Protocol

Protocol for response transformer interface.

abstract build_variant_row(study: WDAEStudy, v: SummaryVariant | FamilyVariant, column_descs: list[dict], **kwargs: str | None) list[source]
abstract transform_gene_view_summary_variant(variant: SummaryVariant, frequency_column: str) Generator[dict[str, Any], None, None][source]
abstract transform_gene_view_summary_variant_download(variants: Iterable[SummaryVariant], frequency_column: str, summary_variant_ids: set[str]) Iterator[str][source]

Produce an iterator for a download file response.

abstract variant_transformer(study: WDAEStudy, pheno_values: dict[str, Any] | None) Callable[[FamilyVariant], FamilyVariant][source]
class studies.study_wrapper.WDAEAbstractStudy(genotype_data: GenotypeData | None = None, phenotype_data: PhenotypeData | None = None)[source]

Bases: object

A genotype and phenotype data wrapper for use in the wdae module.

property config: dict[str, Any]
property config_columns: dict[str, Any] | None
property description: str | None
property families: FamiliesData
property genotype_data: GenotypeData
get_children_ids(*, leaves: bool = True) list[str][source]

Return the list of children ids.

property has_pheno_data: bool
property is_genotype: bool
property is_group: bool
property is_phenotype: bool
make_config(genotype_data: GenotypeData | None, phenotype_data: PhenotypeData | None) dict[str, Any][source]

Create a configuration for the WDAEStudy.

property name: str
property parents: set[str]
property person_set_collections: dict[str, PersonSetCollection]
property phenotype_data: PhenotypeData
abstract query_variants_wdae_streaming(kwargs: dict[str, Any], sources: list[dict[str, Any]], query_transformer: QueryTransformerProtocol, response_transformer: ResponseTransformerProtocol, max_variants_count: int | None = 10000, *, max_variants_message: bool = False) Generator[list | None, None, None][source]

Wrap query variants method for WDAE streaming.

property study_id: str
class studies.study_wrapper.WDAEStudy(genotype_data: GenotypeData | None, phenotype_data: PhenotypeData | None)[source]

Bases: WDAEAbstractStudy

A genotype and phenotype data wrapper for use in the wdae module.

static build_genotype_data_all_datasets(genotype_data: GenotypeData) dict[str, Any][source]

Prepare response for all genotype datasets.

static build_genotype_data_description(gpf_instance: Any, genotype_data: GenotypeData, person_set_collection_configs: dict[str, Any] | None) dict[str, Any][source]

Build and return genotype data group description.

static get_columns_as_sources(config: dict[str, Any], column_ids: list[str]) list[dict[str, Any]][source]

Return the list of column sources.

get_gene_view_summary_variants(frequency_column: str, query_transformer: QueryTransformerProtocol, response_transformer: ResponseTransformerProtocol, **kwargs: Any) Generator[dict[str, Any], None, None][source]

Return gene browser summary variants.

get_gene_view_summary_variants_download(frequency_column: str, query_transformer: QueryTransformerProtocol, response_transformer: ResponseTransformerProtocol, **kwargs: Any) Iterable[source]

Return gene browser summary variants for downloading.

query_variants_wdae(kwargs: dict[str, Any], sources: list[dict[str, Any]], query_transformer: QueryTransformerProtocol, response_transformer: ResponseTransformerProtocol, max_variants_count: int | None = 10000, *, max_variants_message: bool = False) Iterable[list][source]

Wrap query variants method for WDAE streaming.

query_variants_wdae_streaming(kwargs: dict[str, Any], sources: list[dict[str, Any]], query_transformer: QueryTransformerProtocol, response_transformer: ResponseTransformerProtocol, max_variants_count: int | None = 10000, *, max_variants_message: bool = False) Generator[list | None, None, None][source]

Wrap query variants method for WDAE streaming of variants.

class studies.study_wrapper.WDAEStudyGroup(genotype_data: GenotypeData | None, pheno_data: PhenotypeData | None, children: list[WDAEStudy])[source]

Bases: WDAEStudy

Genotype data study wrapper class for WDAE.

get_studies_ids(*, leaves: bool = True) list[str][source]
studies.study_wrapper.get_clean_config(config: dict[str, Any]) dict[str, Any][source]
studies.study_wrapper.remove_none_values(config: dict[str, Any]) dict[str, Any][source]

Remove None values from config

Module contents