import abc
from collections.abc import Iterable
from typing import Any, cast
from dae.person_sets import PersonSetCollection
from studies.study_wrapper import WDAEStudy
from enrichment_api.enrichment_helper import EnrichmentHelper
from enrichment_api.enrichment_serializer import EnrichmentSerializer
[docs]
class BaseEnrichmentBuilder:
"""Base class for enrichment builders."""
def __init__(
self, enrichment_helper: EnrichmentHelper,
study: WDAEStudy,
):
assert enrichment_helper.study.study_id == study.study_id
self.enrichment_helper = enrichment_helper
self.study = study
enrichment_config = study.enrichment_config
assert enrichment_config is not None
self.enrichment_config = enrichment_config
[docs]
@abc.abstractmethod
def build(
self, gene_syms: Iterable[str],
background_id: str | None, counting_id: str | None,
) -> list[dict[str, Any]]:
raise NotImplementedError
[docs]
class EnrichmentBuilder(BaseEnrichmentBuilder):
"""Build enrichment tool test."""
def __init__(
self,
enrichment_helper: EnrichmentHelper,
study: WDAEStudy,
):
super().__init__(enrichment_helper, study)
self.results: list[dict[str, Any]]
[docs]
def build_results(
self, gene_syms: Iterable[str],
background_id: str | None, counting_id: str | None,
) -> list[dict[str, Any]]:
"""Build and return a list of enrichment results.
Returns:
A list of dictionaries representing the enrichment results.
"""
results = []
person_set_collection = cast(
PersonSetCollection,
self.study.genotype_data.get_person_set_collection(
self.enrichment_helper.get_selected_person_set_collections(),
),
)
effect_types = self.enrichment_config["effect_types"]
enrichment_result = self.enrichment_helper.calc_enrichment_test(
person_set_collection.id,
gene_syms,
effect_types,
background_id=background_id,
counter_id=counting_id,
)
for ps_id, effect_res in enrichment_result.items():
res: dict[str, Any] = {}
person_set = person_set_collection.person_sets[ps_id]
children_stats = person_set.get_children_stats()
res["childrenStats"] = {
"M": children_stats.male,
"F": children_stats.female,
"U": children_stats.unspecified,
}
res["selector"] = person_set.name
res["geneSymbols"] = list(gene_syms)
res["peopleGroupId"] = person_set_collection.id
res["peopleGroupValue"] = ps_id
res["datasetId"] = self.study.study_id
for effect_type, enrichment_res in effect_res.items():
res[effect_type] = {}
res[effect_type]["all"] = enrichment_res.all
res[effect_type]["rec"] = enrichment_res.rec
res[effect_type]["female"] = enrichment_res.female
res[effect_type]["male"] = enrichment_res.male
results.append(res)
return results
[docs]
def build(
self, gene_syms: Iterable[str],
background_id: str | None, counting_id: str | None,
) -> list[dict[str, Any]]:
results = self.build_results(gene_syms, background_id, counting_id)
serializer = EnrichmentSerializer(self.enrichment_config, results)
results = serializer.serialize()
self.results = results
return self.results