"""Annotate perturbations and controls.
Several later steps (:func:`~mantispy.pp.sphere`, :func:`~mantispy.tl.grit`) default to ``reference="negcon"``, which reads ``Metadata_Control_Type``; :func:`annotate_controls` writes it along with the ``Metadata_Control`` boolean.
"""
from __future__ import annotations
from collections.abc import Sequence
import numpy as np
from anndata import AnnData
from mantispy._core.frames import as_frame, categorize_metadata
from mantispy._core.logging import get_logger
from mantispy._core.mutation import inplace_or_copy
PERTURBATION_KEYS = (
"Metadata_Perturbation",
"Metadata_Compound",
"Metadata_Treatment",
"Metadata_BroadSample",
)
[docs]
def find_perturbation_key(adata: AnnData, perturbation_key: str | None = None) -> str:
"""Resolve which ``obs`` column holds the perturbation identity.
Args:
adata: Object whose ``obs`` columns are searched.
perturbation_key: Column to use, returned unchanged once it is known to be present.
``None`` takes the first column of ``PERTURBATION_KEYS`` that is there.
Returns:
The name of the ``obs`` column that identifies the perturbation.
Raises:
KeyError: If `perturbation_key` is not an ``obs`` column, or no column was named and none of ``PERTURBATION_KEYS`` is present.
"""
if perturbation_key is not None:
if perturbation_key not in adata.obs:
raise KeyError(f"obs has no column {perturbation_key!r}")
return perturbation_key
for candidate in PERTURBATION_KEYS:
if candidate in adata.obs:
return candidate
raise KeyError(
f"none of {list(PERTURBATION_KEYS)} is in obs; pass perturbation_key= to say which "
"column identifies the perturbation"
)
[docs]
@inplace_or_copy()
def annotate_controls(
adata: AnnData,
negcon: Sequence[str] = ("DMSO",),
poscon: Sequence[str] | None = None,
perturbation_key: str | None = None,
copy: bool = False,
) -> AnnData | None:
"""Mark negative (and optionally positive) controls.
Args:
adata: Object to annotate.
negcon: Perturbation values that are negative controls.
poscon: Perturbation values that are positive controls, if any.
perturbation_key: Column holding the perturbation, auto-detected from ``PERTURBATION_KEYS`` when omitted.
copy: Return a modified copy instead of mutating in place.
Returns:
``None``, or the modified copy when ``copy=True``.
Writes ``obs["Metadata_Control_Type"]`` and the boolean ``obs["Metadata_Control"]`` it implies
(``True`` for any control). Steps defaulting to ``reference="negcon"`` read the type column, so the
negcon rows are selected by it alone.
Notes:
The classes named here are authoritative: rows matching `negcon` become ``"negcon"`` and earlier
``"negcon"`` rows that no longer match become treatments, and likewise for `poscon` when it is given.
Every other class an upstream annotation set, such as ``"poscon"`` or ``"empty"`` wells from a plate
map or :func:`annotate_jump`, is kept.
"""
key = find_perturbation_key(adata, perturbation_key)
values = adata.obs[key].astype(str)
is_negcon = values.isin([str(v) for v in negcon]).to_numpy()
if not is_negcon.any():
get_logger().warning(
"no negative controls found: none of %s appear in obs[%r]. "
"Steps defaulting to reference='negcon' will fail until this is set.",
list(negcon),
key,
)
if "Metadata_Control_Type" in adata.obs:
upstream = as_frame(adata.obs)["Metadata_Control_Type"]
control_type = upstream.astype(object).where(upstream.notna(), "treatment").astype(str).to_numpy(dtype=object)
else:
control_type = np.full(adata.n_obs, "treatment", dtype=object)
control_type[(control_type == "negcon") & ~is_negcon] = "treatment"
control_type[is_negcon] = "negcon"
if poscon is not None:
is_poscon = values.isin([str(v) for v in poscon]).to_numpy()
control_type[(control_type == "poscon") & ~is_poscon] = "treatment"
control_type[is_poscon] = "poscon"
adata.obs["Metadata_Control_Type"] = control_type
adata.obs["Metadata_Control"] = control_type != "treatment"
return None
[docs]
@inplace_or_copy(expects="well")
def annotate_jump(adata: AnnData, kind: str = "compound", copy: bool = False) -> AnnData | None:
"""Join the JUMP annotation onto profiles read from the Cell Painting Gallery.
A JUMP plate parquet records only the source, plate and well of a profile.
What the well contained is published in a separate repository keyed by ``Metadata_JCP2022``.
Args:
adata: Well-level JUMP profiles carrying ``Metadata_Source``, ``Metadata_Plate`` and ``Metadata_Well``, or ``Metadata_JCP2022`` as the assembled profiles do.
kind: Which annotation to join, ``"compound"``, ``"crispr"`` or ``"orf"``.
copy: Return an annotated copy instead of annotating in place.
Returns:
``None``, or the annotated copy.
Adds ``Metadata_JCP2022`` (the perturbation identifier), ``Metadata_Perturbation``, ``Metadata_Perturbation_Type`` and ``Metadata_Control``.
For compounds ``Metadata_Perturbation_Type`` is ``"compound"``, it adds ``Metadata_InChIKey``, and the controls are JUMP's DMSO wells.
For CRISPR and ORF ``Metadata_Perturbation`` is the gene symbol, ``Metadata_Perturbation_Type`` is the kind, ``Metadata_Control`` marks any control well, and it adds ``Metadata_Gene``, ``Metadata_Control_Type`` (``"negcon"``, ``"poscon"`` or ``"treatment"``) and ``Metadata_ChromosomeArm``, the arm the gene sits on.
Notes:
Downloads about 14 MB of annotation once and caches it.
Wells the annotation does not cover are kept and logged, since an unannotated well is still a measurement.
"""
from mantispy.io._jump import join_jump_annotation
obs = as_frame(adata.obs)
joined = join_jump_annotation(obs.copy(), kind=kind)
joined.index = obs.index
adata.obs = categorize_metadata(joined)
get_logger().info(
"annotate_jump: %d perturbations over %d wells, %d of them controls",
int(joined["Metadata_JCP2022"].nunique()),
len(joined),
int(joined["Metadata_Control"].sum()),
)
return None
