from typing import Any
from gain import logging
from gain.annotation.annotatable import Annotatable
from gain.annotation.annotation_config import (
AnnotatorInfo,
)
from gain.annotation.annotation_pipeline import (
AnnotationPipeline,
Annotator,
AttributeSpec,
)
from gain.annotation.annotator_base import (
AnnotatedValues,
AnnotatorBase,
fold_own_values,
)
from gain.gene_sets.gene_set import (
GeneSet,
build_gene_set_collection_from_resource,
)
from gain.genomic_resources import GenomicResource
from gain.genomic_resources.resource_types import GENE_SET_TYPES
logger = logging.getLogger(__name__)
[docs]
def build_gene_set_annotator(
pipeline: AnnotationPipeline,
info: AnnotatorInfo,
) -> Annotator:
"""Create a gene set annotator."""
# Before the input_gene_list check, for the reason the gene score
# annotator resolves its resource first: the resource is the thing
# this annotator was configured to read.
gene_set_resource = GeneSetAnnotator.resolve_resource(pipeline, info)
input_gene_list = GeneSetAnnotator.resolve_input_gene_list(
pipeline, info)
return GeneSetAnnotator(
pipeline,
info,
gene_set_resource,
input_gene_list,
)
[docs]
class GeneSetAnnotator(AnnotatorBase):
"""Gene set annotator class."""
#: Shared with the collection that opens the resource, rather than
#: spelled again here: a third spelling added there would otherwise
#: be refused by this annotator before the collection could accept
#: it -- the "stated in N places" fault gain#1329 is about.
ACCEPTED_RESOURCE_TYPES = GENE_SET_TYPES
DEFAULT_AGGREGATOR_TYPE = "list"
def __init__(
self,
pipeline: AnnotationPipeline | None,
info: AnnotatorInfo,
gene_set_resource: GenomicResource,
input_gene_list: str,
):
self.gene_set_resource = gene_set_resource
self.gene_set_collection = build_gene_set_collection_from_resource(
self.gene_set_resource)
self.gene_sets: list[GeneSet] | None = None
self.input_gene_list = input_gene_list
info.resources += [gene_set_resource]
info.documentation = (
"This gene set collection annotator uses the "
f"**{self.gene_set_collection.collection_id}** "
f"gene set collection."
)
self._info = info
super().__init__(pipeline, info)
[docs]
def get_attribute_specs(self) -> dict[str, AttributeSpec]:
gene_sets_list = self.gene_set_collection \
.get_gene_sets_list_statistics()
if gene_sets_list is None:
logger.info(
"The gene set collection statistics for %s is empty.",
self.gene_set_collection.collection_id,
)
self.gene_set_collection.load()
gene_sets_list = [
{"name": gs.name, "count": gs.count,
"desc": gs.desc or gs.name}
for gs in sorted(
self.gene_set_collection.get_all_gene_sets(),
key=lambda gs: (-gs.count, gs.name),
)
]
result: dict[str, AttributeSpec] = {
"in_sets": AttributeSpec(
source="in_sets", value_type="object", description=(
"List of the gene sets of the collection, "
"which have at least one gene from the input gene "
"list"
)),
}
result.update({
gs["name"]: AttributeSpec(
source=gs["name"],
value_type="object",
description=f"({gs['count']}) {gs['desc']}",
is_default=False,
)
for gs in gene_sets_list
})
return result
[docs]
def get_attribute_defaults(
self, spec: AttributeSpec,
) -> dict[str, Any]:
if spec.source == "in_sets":
return {}
return {"aggregator": self.DEFAULT_AGGREGATOR_TYPE}
@property
def used_context_attributes(self) -> tuple[str, ...]:
return (self.input_gene_list,)
[docs]
def open(self) -> Annotator:
self.gene_set_collection.load()
self.gene_sets = self.gene_set_collection.get_all_gene_sets()
super().open()
return self
def _do_annotate(
self,
annotatable: Annotatable | None, # ruff: ignore[unused-method-argument]
context: dict[str, Any],
) -> AnnotatedValues:
"""Answer the gene sets the input gene list meets, already reduced.
Every gene set in the collection is intersected, because
``in_sets`` names the ones that matched and is not restricted to
the configured attributes. Each configured per-set attribute
then takes its own intersection folded by its own aggregator --
``list`` by default -- and ``in_sets``, which names none, passes
through as it is.
This annotator reduces for ITSELF (gain#1133): what it holds is
an intersection rather than a stream of records, so there is no
score-side fold to move it onto, and the base no longer has one.
"""
genes = context.get(self.input_gene_list)
if genes is None:
return self._empty_result()
genes_set = set(genes)
if self.gene_sets is None:
raise ValueError(
f"The GeneSetAnnotator {self.gene_set_resource} "
f"is not open.")
in_sets: list[str] = []
intersections: dict[str, Any] = {"in_sets": in_sets}
for gs in self.gene_sets:
intersecting = list(genes_set.intersection(set(gs.syms)))
intersections[gs.name] = intersecting
if intersecting:
in_sets.append(gs.name)
return fold_own_values(self._attributes, intersections)