"""Test every gene against every measurement in one matrix operation.
:func:`sweep` asks which genes change which measurements across the whole
screen at once rather than one gene at a time.
The associations are computed as one matrix product. A representative screen
with 1,376 guides x 785 measurements over 1,366 wells is computed in 0.1
seconds; fitting the same pairs as about a million separate regressions takes
hours.
The module applies three corrections. Each addresses an error found while
sweeping one gene by hand:
* Identifier columns are excluded from the measurements.
`pathogen_object_label` came out at p=2.5e-07 for EAF1, but it is a label:
the selected cells sit lower in the segmentation's numbering, which is a
position in a list. `pathogen_pathogen` is the same column under another
name (spearman 0.9979, identical in 141,626 of 226,467 rows).
* P values are corrected for the number of tests (Benjamini-Hochberg).
With 785 measurements, `pathogen_solidity` at p=1.8e-03 does not survive
the correction.
* Each measurement's circularity is reported in its own column. The
classification score is a function of the image, so a measurement it
already tracks cannot corroborate a result derived from it. On this screen
spearman(pred, pathogen_channel_1_mean_intensity) = -0.389, and the
result that first looked strongest for GRA14 was that measurement, in the
direction the correlation predicts.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
import numpy as np
import pandas as pd
from .figures.style import figure_style, theme_target
__all__ = [
"IDENTIFIER_PATTERNS",
"SweepResult",
"is_measurement",
"measurement_columns",
"sweep",
"gene_fractions",
"gene_of_guide",
]
#: WHOLE TOKENS, not substrings. `object` and `label` catch the
#: segmentation's own numbering, which is what produced a p of 2.5e-07 and
#: means nothing.
#:
#: MATCHED AS TOKENS BECAUSE SUBSTRINGS ARE WRONG HERE: "path" is in
#: "png_path" and it is also in "pathogen", so a substring rule silently
#: dropped EVERY pathogen measurement -- a third of the screen -- and the
#: sweep reported the remainder as though that were all there was. Caught by
#: the test that asks whether `pathogen_area` is a measurement.
IDENTIFIER_PATTERNS: Tuple[str, ...] = (
"label", "id", "object", "screen", "source", "prc", "prcf", "prcfo",
"plate", "row", "col", "column", "field", "well", "path", "file",
"name", "png", "index",
)
def _tokens(name: str) -> Tuple[str, ...]:
"""``plateID`` and ``object_label`` alike, as lower-case words."""
spaced = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", str(name).strip())
return tuple(t for t in re.split(r"[^A-Za-z0-9]+", spaced.lower()) if t)
[docs]
def is_measurement(name: str) -> bool:
"""Whether ``name`` measures the object rather than naming it.
:param name: candidate column name to classify.
"""
parts = _tokens(name)
if not parts:
return False
return not any(token in IDENTIFIER_PATTERNS for token in parts)
[docs]
def measurement_columns(frame: pd.DataFrame) -> List[str]:
"""Every numeric column of ``frame`` that is a measurement.
:param frame: object-measurement table whose numeric columns are screened.
A column whose ranks correlate with an identifier column at above 0.99 is
also excluded, whatever it is called. `pathogen_pathogen` passes the name
test but matches the object label to four decimal places, so it is
detected from its values rather than its name.
"""
numeric = [c for c in frame.columns
if pd.api.types.is_numeric_dtype(frame[c])]
named = [c for c in numeric if is_measurement(c)]
identifiers = [c for c in numeric if not is_measurement(c)]
if not identifiers:
return named
usable = [c for c in named
if pd.to_numeric(frame[c], errors="coerce").nunique() >= 3]
if not usable:
return []
left = frame[usable].apply(pd.to_numeric, errors="coerce").rank()
right = frame[identifiers].apply(pd.to_numeric, errors="coerce").rank()
left = left.fillna(left.mean()).to_numpy(dtype=float)
right = right.fillna(right.mean()).to_numpy(dtype=float)
left = left - left.mean(axis=0)
right = right - right.mean(axis=0)
left /= (np.linalg.norm(left, axis=0) + 1e-12)
right /= (np.linalg.norm(right, axis=0) + 1e-12)
twin = np.abs(left.T @ right).max(axis=1) > 0.99
return [c for c, is_twin in zip(usable, twin) if not is_twin]
@dataclass(frozen=True)
[docs]
class SweepResult:
"""Result of :func:`sweep`: the effect grid and a tidy per-pair table.
:param table: tidy result frame with one row per fitted guide or
gene-measurement pair and its effect, significance, support, and
control evidence.
:param effects: dense blocked association matrix indexed by fitted guide
or gene and columned by measurement; these effects are repeated in
``table``.
:param n_wells: number of well identities shared by wells and fractions
before guide-specific filtering; unlike ``table['n_wells']``, this is
screen-wide.
:param n_blocks: number of distinct block labels removed from a fitted
design, or zero when the sweep returned before fitting.
:param dropped: numeric wells columns omitted from the effect grid,
including automatically rejected identifier or duplicate columns and
explicitly dropped measurements.
:param circularity_known: whether at least three joined score values
produced a finite measurement-to-score circularity; when false,
circularity is NaN and :meth:`survivors` refuses a circularity cutoff.
"""
table: pd.DataFrame
effects: pd.DataFrame
n_wells: int
n_blocks: int
dropped: Tuple[str, ...] = ()
#: Whether the score actually joined to the wells. False means the
#: circularity column is NaN and MUST NOT be read as "not circular".
circularity_known: bool = False
[docs]
def survivors(self, alpha: float = 0.05,
max_circularity: float = 1.0) -> pd.DataFrame:
"""Rows past the correction, optionally past a circularity bar too.
:raises ValueError: a circularity bar was asked for and the score
never joined to the wells. Filtering on a column of NaN would
return an empty table.
"""
out = self.table[self.table["q"] < float(alpha)]
if max_circularity < 1.0:
if not self.circularity_known:
raise ValueError(
"circularity was never computed -- the score did not join "
"to any well, so filtering on it would return nothing and "
"look like an answer. Check the plate names match "
"(a score CSV may say 'pplate1' where the database says "
"'plate1').")
out = out[out["circularity"] < float(max_circularity)]
return out.sort_values("q")
[docs]
def describe(self) -> str:
"""Summarize grid size, significance, circularity, and omitted inputs."""
survived = int((self.table["q"] < 0.05).sum())
clean = (int(((self.table["q"] < 0.05)
& (self.table["circularity"] < 0.15)).sum())
if self.circularity_known else None)
head = (f"{len(self.effects.index):,} gene/guide(s) x "
f"{len(self.effects.columns):,} measurement(s) over "
f"{self.n_wells:,} wells in {self.n_blocks} block(s). "
f"{survived:,} pass Benjamini-Hochberg at 0.05")
middle = (f", of which {clean:,} are not already tracked by the score. "
if clean is not None else
" (circularity NOT computed -- the score joined to no "
"well, so it must not be read as 'not circular'). ")
return (head + middle
+ f"{len(self.dropped):,} input column(s) were left out.")
def _residualise(matrix: np.ndarray, blocks: np.ndarray) -> np.ndarray:
"""Subtract each block's mean -- the plate cannot become a gene effect."""
out = matrix.astype(float, copy=True)
for value in np.unique(blocks):
mask = blocks == value
if mask.sum():
out[mask] -= np.nanmean(out[mask], axis=0)
return np.nan_to_num(out)
#: Known organism prefixes used only when a guide name does not match the
#: structural ``<organism>_<gene>_<guide>`` form written by ``process_reads``.
#: Structural parsing remains the primary rule so libraries from other
#: organisms are not pooled under their organism prefix.
GUIDE_PREFIXES: Tuple[str, ...] = ("TGGT1_", "TGME49_", "TGVEG_", "TGRH88_")
[docs]
def gene_of_guide(guide: Any, prefix: Optional[str] = None) -> Optional[str]:
"""Return the gene a guide name belongs to, for both spellings in use.
`spacr.hits.gene_of` reads a regression design term --
`fraction:grna[225160_1]` -- and truncates at the first underscore.
Applied to the bare `TGGT1_225160_2` that a count table carries, that rule
returns `TGGT1`, the organism, for every guide in the screen. Only a
bracketed term is therefore passed to `hits.gene_of`.
A bare name is parsed by its number of underscore-separated parts.
`process_reads` splits `<organism>_<gene>_<guide>` on exactly three
components, so the gene of a name with three or more parts is the middle
part(s), and the gene of a two-part name is the first part. This works for
`PF3D7_0100100_1` and for a human library without listing either organism.
When an organism prefix has been removed -- the measured `prefix`, or one
of :data:`GUIDE_PREFIXES` -- the remainder is read as `<gene>_<guide>`, so
the gene is everything except the last part. `TGGT1_ROP18_kinase_1` gives
`ROP18_kinase`, not `kinase`.
:param guide: bare guide name or bracketed regression design term to parse.
:param prefix: an organism prefix measured from the library, as
:func:`spacr.control_names.common_prefix` returns. When given, it is
removed first and the remainder is read as `<gene>_<guide>`; this
handles prefixes that are not in :data:`GUIDE_PREFIXES`.
"""
text = str(guide or "").strip()
if not text:
return None
if "[" in text:
from .hits import gene_of as _design_gene_of
return _design_gene_of(text)
head = str(prefix or "").strip()
stripped = False
if head and text.startswith(f"{head}_"):
text = text[len(head) + 1:]
stripped = True
else:
for known in GUIDE_PREFIXES:
if text.upper().startswith(known):
text = text[len(known):]
stripped = True
break
parts = [p for p in text.split("_")]
if stripped:
if len(parts) >= 2:
gene = "_".join(parts[:-1]).strip()
else:
gene = parts[0].strip()
elif len(parts) >= 3:
gene = "_".join(parts[1:-1]).strip()
else:
gene = parts[0].strip()
return gene or None
[docs]
def gene_fractions(fractions: pd.DataFrame,
gene_of: Optional[Any] = None) -> pd.DataFrame:
"""Return each gene's fraction per well as the sum of its guides' fractions.
This is the rule the regression applies (`wells_for_coefficient` with
`guide_aggregation='sum'`), so a gene-level sweep uses the same gene
fractions as the fit that identified the gene.
:param fractions: well-by-guide fraction matrix. Columns assigned to the
same gene are summed row-wise.
:param gene_of: guide name -> gene id. Defaults to
:func:`gene_of_guide`. A guide for which it raises or returns no gene
is left out.
:returns: one column per gene. Guides that name no gene are left out
rather than pooled into an "unknown" gene.
"""
if gene_of is None:
gene_of = gene_of_guide
mapping: Dict[str, List[str]] = {}
for guide in fractions.columns:
gene = None
try:
gene = gene_of(str(guide))
except Exception: # noqa: BLE001
gene = None
if gene:
mapping.setdefault(str(gene), []).append(guide)
if not mapping:
return pd.DataFrame(index=fractions.index)
return pd.DataFrame(
{gene: fractions[guides].sum(axis=1) for gene, guides in mapping.items()},
index=fractions.index)
[docs]
def sweep(wells: pd.DataFrame, fractions: pd.DataFrame, *,
blocks: Optional[Sequence] = None,
scores: Optional[Sequence[float]] = None,
alpha: float = 0.05,
controls: Optional[Sequence[str]] = None,
min_wells: int = 5,
measurements: Optional[Iterable[str]] = None,
drop_measurements: Optional[Iterable[str]] = None,
drop_guides: Optional[Iterable[str]] = None,
max_share: Optional[float] = None,
max_wells_fraction: Optional[float] = None,
level: str = "guide") -> SweepResult:
"""Associate every guide with every measurement, blocked and corrected.
:param wells: one row per well, the measurements in its columns.
:param fractions: one row per well, one column per guide, holding that
guide's fraction. Aligned to ``wells`` by index.
:param blocks: the plate of each well. Each block's mean is subtracted
before fitting. Absent means one block, so a plate difference can then
appear as a gene effect.
:param scores: the per-well classification score, used only to compute
each measurement's circularity. Absent leaves the ``circularity``
column NaN and :attr:`SweepResult.circularity_known` False.
:param controls: guide or gene names to mark as controls in the
``control`` column. They are not removed. The regression drops the
control columns because they are not part of the contrast it fits;
this sweep tests whether each guide or gene changes a measurement, and
the controls' results are reported for comparison.
:param drop_measurements: columns to leave out, by name; the complement of
``measurements``.
:param drop_guides: guides or genes to leave out, by name. A guide is
dropped when its own name or its gene (:func:`gene_of_guide`) is
listed, so gene ids work at either level.
:param max_share: drop a guide whose median well fraction, where it is
present, is above this.
:param max_wells_fraction: drop a guide present in more than this share
of wells. This prevalence filter is separate from ``max_share``, which
measures median abundance where the guide is present. A guide can be
common at low abundance or rare at high abundance, so the two filters
identify different forms of over-representation.
:param min_wells: guides present in fewer wells than this are dropped.
:param level: ``'guide'``, ``'gene'``, or ``'both'``. A gene's fraction in
a well is the sum of its guides' fractions, the same rule the
regression applies (:func:`gene_fractions`).
:returns: a :class:`SweepResult`. With ``'both'`` the table carries a
``level`` column and the guide rows stay reachable beside the gene
ones.
"""
wanted = str(level or "guide").strip().lower()
if wanted not in ("guide", "gene", "both"):
raise ValueError(
f"level must be 'guide', 'gene' or 'both'; got {level!r}")
if wanted != "guide":
genes = gene_fractions(fractions)
if wanted == "gene":
fractions = genes
elif len(genes.columns):
fractions = pd.concat(
[fractions, genes.rename(columns=lambda g: f"{g} (gene)")],
axis=1)
common = wells.index.intersection(fractions.index)
wells = wells.loc[common]
fractions = fractions.loc[common]
n = int(len(common))
if n < 3:
empty = pd.DataFrame(columns=["guide", "measurement", "effect", "p",
"q", "circularity", "n_wells"])
return SweepResult(table=empty, effects=pd.DataFrame(), n_wells=n,
n_blocks=0)
chosen = list(measurements) if measurements is not None \
else measurement_columns(wells)
if drop_measurements:
unwanted = {str(c) for c in drop_measurements}
chosen = [c for c in chosen if str(c) not in unwanted]
dropped = tuple(c for c in wells.columns
if pd.api.types.is_numeric_dtype(wells[c])
and c not in chosen)
present = (fractions > 0).sum(axis=0)
guides = [g for g in fractions.columns if int(present.get(g, 0)) >= min_wells]
excluded: Dict[str, Tuple[str, ...]] = {}
if drop_guides:
unwanted = {str(g) for g in drop_guides}
gone = tuple(g for g in guides
if str(g) in unwanted
or str(gene_of_guide(g) or "") in unwanted)
if gone:
excluded["named"] = gone
guides = [g for g in guides if g not in set(gone)]
if max_wells_fraction is not None and n:
limit = float(max_wells_fraction)
gone = tuple(g for g in guides
if int(present.get(g, 0)) / n > limit)
if gone:
excluded["in too many wells"] = gone
guides = [g for g in guides if g not in set(gone)]
if max_share is not None:
limit = float(max_share)
gone = []
for g in guides:
column = pd.to_numeric(fractions[g], errors="coerce")
here = column[column > 0]
if len(here) and float(here.median()) > limit:
gone.append(g)
if gone:
excluded["too large a share"] = tuple(gone)
guides = [g for g in guides if g not in set(gone)]
for why, names in excluded.items():
shown = ", ".join(str(x) for x in names[:8])
more = f" and {len(names) - 8} more" if len(names) > 8 else ""
print(f"Sweep: {len(names)} guide(s) left out ({why}): {shown}{more}.")
if not chosen or not guides:
empty = pd.DataFrame(columns=["guide", "measurement", "effect", "p",
"q", "circularity", "n_wells"])
return SweepResult(table=empty, effects=pd.DataFrame(), n_wells=n,
n_blocks=0, dropped=dropped)
block = np.asarray(list(blocks) if blocks is not None else ["all"] * n)
n_blocks = int(len(np.unique(block)))
M = _residualise(wells[chosen].to_numpy(dtype=float), block)
F = _residualise(fractions[guides].to_numpy(dtype=float), block)
M /= (M.std(axis=0, keepdims=True) + 1e-12)
F /= (F.std(axis=0, keepdims=True) + 1e-12)
R = (F.T @ M) / max(n - n_blocks, 1)
R = np.clip(np.nan_to_num(R), -0.999999, 0.999999)
sq = F * F
n_eff = (sq.sum(axis=0) ** 2) / np.maximum((sq * sq).sum(axis=0), 1e-300)
n_eff = np.clip(n_eff, 3.0, float(n))
df_guide = np.maximum(n_eff - n_blocks - 1.0, 1.0)
df = df_guide[:, None]
presence = (fractions[guides] > 0)
share_of = np.round(
fractions[guides].where(presence).median(axis=0).fillna(0.0).to_numpy(), 4)
ubiquity = (presence.sum(axis=0).to_numpy() >= 0.9 * n)
marked = {str(c) for c in (controls or ())}
t = R * np.sqrt(df / (1.0 - R * R))
from scipy.stats import t as _t
p = 2.0 * _t.sf(np.abs(t), df)
circular = np.full(len(chosen), np.nan)
circularity_known = False
if scores is not None:
s = pd.Series(np.asarray(list(scores), dtype=float), index=common).rank()
block_m = wells[chosen].apply(pd.to_numeric, errors="coerce").rank()
sv = s.fillna(s.mean()).to_numpy(dtype=float)
mv = block_m.fillna(block_m.mean()).to_numpy(dtype=float)
sv = sv - sv.mean()
mv = mv - mv.mean(axis=0)
denom = (np.linalg.norm(sv) * np.linalg.norm(mv, axis=0)) + 1e-12
circular = np.abs((sv @ mv) / denom)
overlap = int(pd.Series(np.asarray(list(scores), dtype=float),
index=common).notna().sum())
circularity_known = overlap >= 3 and bool(np.isfinite(circular).any())
if not circularity_known:
circular = np.full(len(chosen), np.nan)
effects = pd.DataFrame(
R, index=[str(g).replace(" (gene)", "") for g in guides],
columns=chosen)
table = pd.DataFrame({
"guide": np.repeat(guides, len(chosen)),
"measurement": np.tile(chosen, len(guides)),
"effect": R.ravel(),
"p": p.ravel(),
"circularity": np.tile(circular, len(guides)),
"level": np.repeat(["gene" if str(g).endswith(" (gene)")
or wanted == "gene" else "guide"
for g in guides], len(chosen)),
"share": np.repeat(share_of, len(chosen)),
"ubiquitous": np.repeat(ubiquity, len(chosen)),
"control": np.repeat(
[str(g).replace(" (gene)", "") in marked for g in guides],
len(chosen)),
"n_wells": np.repeat([int(present.get(g, 0)) for g in guides],
len(chosen)),
"effective_wells": np.repeat(np.round(n_eff, 1), len(chosen)),
})
from .multiple_testing import adjust_p_values
q, _rejected = adjust_p_values(table["p"].to_numpy(), method="fdr_bh",
alpha=float(alpha))
table["q"] = q
table["guide"] = table["guide"].astype(str).str.replace(
" (gene)", "", regex=False)
table = table[["level", "guide", "measurement", "effect", "p", "q",
"circularity", "n_wells", "effective_wells", "share",
"ubiquitous", "control"]]
return SweepResult(table=table, effects=effects, n_wells=n,
n_blocks=n_blocks, dropped=dropped,
circularity_known=circularity_known)
[docs]
class HOUSE:
"""Define spaCR's scientific-figure palette and sizing constants.
The palette follows recurring colors in published apicomplexan-genomics
figures, including Waldman et al. (Cell, 2020; Figures 1 and 3) and
Giuliano et al. (Nature Microbiology, 2024; Figure 1). Grey represents
background data, while accent colors identify the smaller subset being
interpreted. Family colors remain stable across panels so readers do not
have to relearn category mappings.
Text, spines, and ticks use theme-aware colors selected by :func:`_readable`
rather than this palette, which keeps figures legible in spaCR's dark
interface while retaining consistent scientific data colors.
"""
GREY = "#B4B4B4"
GREY_DARK = "#7F7F7F"
BLUE = "#2E77BC"
BLUE_LIGHT = "#7FB3E0"
GREEN = "#2E7D4F"
RUST = "#C4441C"
CORAL = "#E8A88C"
GOLD = "#E8C33A"
OCHRE = "#C87A28"
PURPLE = "#8B4A82"
NAVY = "#1F3F6E"
SEQ = "Blues"
DIVERGING = "RdBu_r"
#: Family colours, assigned once and never re-mapped between panels --
#: the rule Waldman Fig 3 keeps for strains across the weight curve, the
#: survival curve and the cyst plot.
FAMILY = {
"pathogen": "#2E77BC", "nucleus": "#8B4A82", "cytoplasm": "#C87A28",
"cell": "#2E7D4F", "intensity": "#E8C33A", "shape": "#1F3F6E",
"other": "#B4B4B4",
}
#: Type sizes in points. Axis labels are the reference size; tick labels
#: use about 0.9x and annotations about 0.85x that size.
LABEL = 7.0
TICK = 6.2
NOTE = 6.0
#: Spines and ticks 0.6-0.7pt; data lines 1.1-1.4pt; reference 0.6pt.
SPINE = 0.65
DATA = 1.25
REFERENCE = 0.6
def _write(figure, path) -> None:
"""Save a sweep figure using the configured export appearance.
When export colors differ from display colors, axes decorations and the
figure background are mapped to the export palette while data colors are
preserved. Modified artists are restored after saving so the figure can
still be displayed by the caller.
Parameters
----------
figure : matplotlib.figure.Figure
Figure to save.
path : str, path-like, or None
Output path. A false value skips the export.
"""
if not path:
return
restore = []
try:
from .figure_style import export_colour, saved_figure_appearance
look = saved_figure_appearance()
except Exception:
look = None
if look is not None and getattr(look, "flip", False):
ground = getattr(look, "ground", None)
if ground:
before = figure.get_facecolor()
alpha = figure.patch.get_alpha()
figure.patch.set_facecolor(ground)
figure.patch.set_alpha(1.0)
restore.append(lambda: (figure.patch.set_facecolor(before),
figure.patch.set_alpha(alpha)))
for axes in figure.axes:
was = axes.get_facecolor()
was_alpha = axes.patch.get_alpha()
axes.set_facecolor(ground)
axes.patch.set_alpha(1.0)
restore.append(
lambda a=axes, c=was, al=was_alpha: (
a.set_facecolor(c), a.patch.set_alpha(al)))
for axes in figure.axes:
for _artist, getter, setter in _chrome_of(axes):
replacement = export_colour(getter(), "chrome", look)
if replacement is None:
continue
current = getter()
setter(replacement)
restore.append(lambda put=setter, old=current: put(old))
try:
from .plot import _checked_savefig
_checked_savefig(figure, path, dpi=200, bbox_inches="tight",
facecolor=figure.get_facecolor())
finally:
for undo in reversed(restore):
undo()
def _chrome_of(axes):
"""Collect color accessors for an axes' non-data artists.
Parameters
----------
axes : matplotlib.axes.Axes
Axes whose title, labels, spines, annotations, and legend are read.
Returns
-------
list of tuple
Triples containing an artist, its color getter, and its color setter.
"""
out = [(axes.title, axes.title.get_color, axes.title.set_color),
(axes.xaxis.label, axes.xaxis.label.get_color,
axes.xaxis.label.set_color),
(axes.yaxis.label, axes.yaxis.label.get_color,
axes.yaxis.label.set_color)]
for spine in axes.spines.values():
out.append((spine, spine.get_edgecolor, spine.set_edgecolor))
for label in list(axes.get_xticklabels()) + list(axes.get_yticklabels()):
out.append((label, label.get_color, label.set_color))
for text in axes.texts:
out.append((text, text.get_color, text.set_color))
legend = axes.get_legend()
if legend is not None:
for text in legend.get_texts():
out.append((text, text.get_color, text.set_color))
return out
def _readable(figure, *axes) -> str:
"""Apply the active theme's foreground style to completed axes.
The figure and axes backgrounds become transparent, while titles, labels,
ticks, spines, and legend text receive the resolved theme color and house
typography. Styling existing artists directly avoids relying on
``rcParams``, which only affect artists when they are created.
Parameters
----------
figure : matplotlib.figure.Figure
Figure whose background is made transparent.
*axes : matplotlib.axes.Axes or None
Axes to style. ``None`` entries are ignored.
Returns
-------
str
Resolved foreground color, suitable for additional annotations.
"""
from .figures.style import ROLES, TYPE_SCALE
ink = ROLES["reference"]
try:
from .figures.style import resolve_ink, theme_target
ink = resolve_ink(theme_target())
except Exception:
pass
try:
figure.patch.set_alpha(0.0)
except Exception:
pass
for axis in axes:
if axis is None:
continue
try:
axis.patch.set_alpha(0.0)
axis.title.set_color(ink)
axis.title.set_fontsize(TYPE_SCALE.get("label", 9))
axis.xaxis.label.set_color(ink)
axis.yaxis.label.set_color(ink)
axis.tick_params(color=ink, labelcolor=ink, which="both",
labelsize=HOUSE.TICK, width=HOUSE.SPINE,
length=2.6)
axis.xaxis.label.set_fontsize(HOUSE.LABEL)
axis.yaxis.label.set_fontsize(HOUSE.LABEL)
for spine in axis.spines.values():
spine.set_edgecolor(ink)
spine.set_linewidth(HOUSE.SPINE)
axis.spines["top"].set_visible(False)
axis.spines["right"].set_visible(False)
axis.grid(False, which="both")
legend = axis.get_legend()
if legend is not None:
for text in legend.get_texts():
text.set_color(ink)
except Exception:
continue
return ink
[docs]
def plot_sweep(result: "SweepResult", path: Optional[str] = None, *,
alpha: float = 0.05, max_circularity: float = 1.0,
top: int = 40, title: str = "", level: Optional[str] = None):
"""Plot a clustered heatmap of the effects that survive the correction.
The full grid is too large to read as an image: 1,240 guides x 767
measurements is 951,080 cells, all coloured whether or not they are
significant. The heatmap therefore draws only survivors -- the guides and
measurements with at least one entry past the correction -- and orders
rows and columns by correlation clustering.
:param result: what :func:`sweep` returned.
:param top: the most guides and measurements to draw, lowest q first.
Survivors beyond this limit are not drawn; read them from
``result.table``.
:returns: the matplotlib Figure, or ``None`` when nothing survived.
"""
import matplotlib.pyplot as plt
keep = result.survivors(alpha=alpha, max_circularity=max_circularity)
if not len(keep):
return None
drawn = str(level or "").strip().lower()
if "level" in keep.columns and keep["level"].nunique() > 1:
drawn = drawn or "gene"
if drawn and "level" in keep.columns:
keep = keep[keep["level"] == drawn]
if not len(keep):
return None
guides = (keep.groupby("guide")["q"].min().sort_values().head(top).index)
measures = (keep.groupby("measurement")["q"].min()
.sort_values().head(top).index)
grid = result.effects.loc[
[g for g in guides if g in result.effects.index],
[m for m in measures if m in result.effects.columns]]
if grid.empty:
return None
grid = _order_like_neighbours(grid)
height = max(3.0, 0.28 * len(grid.index) + 1.6)
width = max(5.0, 0.34 * len(grid.columns) + 3.2)
with figure_style(theme_target()):
figure, axes = plt.subplots(figsize=(width, height))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, grid, kind="heatmap", matrix=True)
limit = float(np.nanmax(np.abs(grid.to_numpy()))) or 1.0
image = axes.imshow(grid.to_numpy(), cmap="RdBu_r", vmin=-limit,
vmax=limit, aspect="auto")
axes.set_xticks(range(len(grid.columns)))
axes.set_xticklabels([c[:34] for c in grid.columns], rotation=90,
fontsize=7)
axes.set_yticks(range(len(grid.index)))
axes.set_yticklabels(grid.index, fontsize=7)
axes.set_title(title or
f"{len(keep):,} association(s) past BH at {alpha:g}"
+ (f", circularity < {max_circularity:g}"
if max_circularity < 1.0 else "")
+ (f" — {drawn}s" if drawn else ""),
fontsize=9)
bar = figure.colorbar(image, ax=axes, fraction=0.025, pad=0.01)
bar.set_label("effect (partial correlation, within plate)", fontsize=7)
bar.ax.tick_params(labelsize=6)
_readable(figure, axes)
figure.tight_layout()
_write(figure, path)
return figure
def _order_like_neighbours(grid: pd.DataFrame) -> pd.DataFrame:
"""Rows and columns ordered so similar ones are adjacent.
Hierarchical clustering when scipy is there, and a correlation-to-the-mean
ordering when it is not -- a picture that fell back to arrival order would
quietly lose the block structure the picture exists to show.
"""
try:
from scipy.cluster.hierarchy import leaves_list, linkage
from scipy.spatial.distance import pdist
def order(matrix):
"""Order one matrix axis by correlation-cluster neighbours.
:param matrix: numeric two-dimensional rows to arrange.
:returns: identity indices below three rows; otherwise average-
linkage leaf indices from correlation distances. Non-finite
matrix values are normalized for distance calculation and
undefined distances become 1.0. Any clustering failure is
handled by the parent's mean-order fallback.
"""
if matrix.shape[0] < 3:
return list(range(matrix.shape[0]))
distance = pdist(np.nan_to_num(matrix), metric="correlation")
distance = np.nan_to_num(distance, nan=1.0)
return list(leaves_list(linkage(distance, method="average")))
rows = order(grid.to_numpy())
columns = order(grid.to_numpy().T)
return grid.iloc[rows, columns]
except Exception: # noqa: BLE001
centre = grid.mean(axis=1).sort_values().index
columns = grid.mean(axis=0).sort_values().index
return grid.loc[centre, columns]
#: The measurement families, and the token that puts a column in one.
#:
#: Ordered: the first family whose token appears wins, so `pathogen_area`
#: is a pathogen measurement rather than a shape one. Deliberately COARSE --
#: six families a reader can hold in their head, not the 767 columns they
#: are drawn from. A column matching none is "other", which is a real answer
#: and not a failure: a screen may measure something none of these names.
MEASUREMENT_FAMILIES: Tuple[Tuple[str, Tuple[str, ...]], ...] = (
("pathogen", ("pathogen",)),
("nucleus", ("nucleus", "nucleolus")),
("cytoplasm", ("cytoplasm", "cyto")),
("cell", ("cell",)),
("intensity", ("intensity", "quartile", "percentile", "mean", "median",
"std", "skew", "kurtosis")),
("shape", ("area", "perimeter", "eccentricity", "solidity", "extent",
"diameter", "axis", "zernike", "moment")),
)
[docs]
def measurement_family(name: Any) -> str:
"""Which family ``name`` belongs to. See :data:`MEASUREMENT_FAMILIES`."""
tokens = set(_tokens(str(name)))
for family, marks in MEASUREMENT_FAMILIES:
if tokens & set(marks):
return family
return "other"
[docs]
def plot_effect_against_representation(
result: "SweepResult", path: Optional[str] = None, *,
alpha: float = 0.05, title: str = "",
level: Optional[str] = None):
"""Plot effect counts against each gene's effective representation.
The plot exposes representation as a possible confound without modifying
the underlying statistic. It shows whether a gene's number of significant
measurements follows the overall representation trend or departs from it.
x is the gene's effective well count (``effective_wells``, the
participation ratio that sets the degrees of freedom of each p-value) and
y is how many measurements it moved past the correction. With at least eight
genes, a dotted line shows the linear trend. A gene above the line has more significant
measurements than its effective well count predicts; a gene at the far
right with a high count may owe its count to statistical weight.
The x axis uses effective wells rather than `share`. `share` is the
median fraction a gene takes of the wells it is present in, so a rare gene
can score high on it. On this module's test fixture a gene in 18 of 120
wells has share 0.44 and one in all 120 has 0.20, the opposite of their
statistical power. The participation ratio is the quantity the test
used.
Controls are drawn with a distinct marker because their relationship to
the trend provides a useful assay calibration.
:returns: the matplotlib Figure, or ``None`` when nothing survived.
"""
import matplotlib.pyplot as plt
keep = result.table
drawn = _one_level(keep, level)
if drawn:
keep = keep[keep["level"] == drawn]
if not len(keep):
return None
passed = keep[keep["q"] < float(alpha)]
if not len(passed):
return None
per_gene = keep.groupby("guide").agg(
weight=("effective_wells", "first"),
share=("share", "first"),
wells=("n_wells", "first"),
control=("control", "first"))
per_gene["hits"] = passed.groupby("guide").size().reindex(
per_gene.index, fill_value=0)
per_gene = per_gene[per_gene["weight"].notna()]
if not len(per_gene):
return None
with figure_style(theme_target()):
figure, axes = plt.subplots(figsize=(7.2, 5.0))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, per_gene, x="weight", y="hits", hue="control", kind="scatter")
controls = per_gene[per_gene["control"].astype(bool)]
rest = per_gene[~per_gene["control"].astype(bool)]
axes.scatter(rest["weight"], rest["hits"], s=9, color=HOUSE.GREY,
edgecolor="none", zorder=2)
if len(controls):
axes.scatter(controls["weight"], controls["hits"], s=22,
color=HOUSE.RUST, edgecolor="none", zorder=4)
if len(per_gene) >= 8 and per_gene["weight"].nunique() > 2:
x = per_gene["weight"].to_numpy(dtype=float)
y = per_gene["hits"].to_numpy(dtype=float)
slope, intercept = np.polyfit(x, y, 1)
span = np.linspace(x.min(), x.max(), 50)
axes.plot(span, slope * span + intercept, color=HOUSE.GREY_DARK,
linewidth=HOUSE.REFERENCE, linestyle=":", zorder=3)
rho = float(np.corrcoef(x, y)[0, 1]) if len(set(x)) > 1 else np.nan
axes.set_title(
title or (f"hits vs representation — rho = {rho:.2f} "
f"({'weight explains much of the ranking' if abs(rho) >= 0.5 else 'weight does not explain the ranking'})"),
fontsize=9)
else:
axes.set_title(title or "hits vs representation", fontsize=9)
axes.set_xlabel("effective wells — the sample size each p was computed "
"on", fontsize=8)
axes.set_ylabel(f"measurements moved past BH at {alpha:g}", fontsize=8)
axes.tick_params(labelsize=7)
marks = [(f"{len(rest):,} genes", HOUSE.GREY)]
if len(controls):
marks.append((f"{len(controls):,} controls", HOUSE.RUST))
marks.append(("dotted: what weight alone predicts", HOUSE.GREY_DARK))
for i, (text, colour) in enumerate(marks):
axes.text(0.02, 0.97 - i * 0.062, text,
transform=axes.transAxes, fontsize=HOUSE.NOTE,
color=colour, va="top")
_readable(figure, axes)
figure.tight_layout()
_write(figure, path)
return figure
[docs]
def plot_measurement_families(result: "SweepResult",
path: Optional[str] = None, *,
alpha: float = 0.05, top: int = 14,
title: str = "", level: Optional[str] = None):
"""Plot each gene's significant measurements by family, as stacked bars.
The significant measurements of each gene are counted per measurement
family (six families plus "other"), which summarises hundreds of columns
in a form that can be compared between genes.
The families are coarse on purpose -- see :data:`MEASUREMENT_FAMILIES`.
:param result: completed guide/measurement sweep whose significant rows
are grouped into measurement families.
:param top: how many genes to draw, most-hits first.
:returns: the matplotlib Figure, or ``None`` when nothing survived.
"""
import matplotlib.pyplot as plt
keep = result.survivors(alpha=alpha)
drawn = _one_level(keep, level)
if drawn:
keep = keep[keep["level"] == drawn]
if not len(keep):
return None
keep = keep.assign(family=[measurement_family(m)
for m in keep["measurement"]])
counts = keep.pivot_table(index="guide", columns="family",
values="q", aggfunc="size").fillna(0)
order = counts.sum(axis=1).sort_values(ascending=False).head(top).index
counts = counts.loc[order]
if counts.empty:
return None
families = [f for f, _ in MEASUREMENT_FAMILIES] + ["other"]
families = [f for f in families if f in counts.columns]
with figure_style(theme_target()):
figure, axes = plt.subplots(
figsize=(7.6, max(3.0, 0.34 * len(counts.index) + 1.4)))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, counts, kind="heatmap", matrix=True)
left = np.zeros(len(counts.index))
for family in families:
values = counts[family].to_numpy(dtype=float)
axes.barh(range(len(counts.index)), values, left=left, height=0.68,
color=HOUSE.FAMILY.get(family, HOUSE.GREY), linewidth=0,
label=family)
left = left + values
axes.set_yticks(range(len(counts.index)))
axes.set_yticklabels(counts.index, fontsize=7)
axes.invert_yaxis()
axes.set_xlabel(f"measurements moved past BH at {alpha:g}", fontsize=8)
axes.tick_params(labelsize=7)
axes.set_title(title or "what kind of measurement each gene moves",
fontsize=9)
axes.legend(fontsize=7, frameon=False, ncol=min(4, len(families)))
_readable(figure, axes)
figure.tight_layout()
_write(figure, path)
return figure
[docs]
def plot_guide_concordance(result: "SweepResult", path: Optional[str] = None,
*, alpha: float = 0.05, top: int = 20,
title: str = ""):
"""Plot how often a gene's own guides agree on the sign of each effect.
Guide agreement is the internal control available in this design. Two
guides against the same gene are independent perturbations of it, so
agreement on the sign of an effect points to the gene, and disagreement
points to guide-specific effects.
Requires a table with guide rows: `sweep(..., level='guide')` or
`'both'`. A gene-level table has nothing to compare and returns
``None``.
:returns: the matplotlib Figure, or ``None`` when there is nothing to
compare.
"""
import matplotlib.pyplot as plt
table = result.table
if "level" in table.columns:
table = table[table["level"] == "guide"]
if not len(table):
return None
genes = [gene_of_guide(g) for g in table["guide"]]
table = table.assign(gene=genes)
table = table[table["gene"].notna()]
if not len(table):
return None
per_gene_guides = table.groupby("gene")["guide"].nunique()
table = table[table["gene"].isin(
per_gene_guides[per_gene_guides >= 2].index)]
if not len(table):
return None
passed = table[table["q"] < float(alpha)]
if not len(passed):
return None
rows = []
for (gene, _measurement), block in passed.groupby(["gene", "measurement"]):
signs = np.sign(block["effect"].to_numpy(dtype=float))
signs = signs[signs != 0]
if len(signs) < 2:
continue
rows.append({"gene": gene,
"agree": float(np.abs(signs.sum()) / len(signs)),
"guides": int(len(signs))})
if not rows:
return None
frame = pd.DataFrame(rows)
summary = frame.groupby("gene").agg(agreement=("agree", "mean"),
pairs=("agree", "size"))
summary = summary.sort_values("agreement", ascending=False).head(top)
with figure_style(theme_target()):
figure, axes = plt.subplots(
figsize=(6.4, max(2.8, 0.30 * len(summary.index) + 1.3)))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, frame, x="gene", y="agree", kind="strip")
positions = np.arange(len(summary.index))
rng = np.random.default_rng(0)
for row, gene in enumerate(summary.index):
values = frame.loc[frame["gene"] == gene, "agree"].to_numpy(float)
spread = rng.uniform(-0.13, 0.13, len(values))
axes.scatter(values, np.full(len(values), row) + spread, s=13,
color=HOUSE.GREY, edgecolor="none", zorder=2)
mean = float(np.mean(values))
colour = (HOUSE.BLUE if mean >= 0.99 else
HOUSE.RUST if mean < 0.6 else HOUSE.GREY_DARK)
axes.plot([mean, mean], [row - 0.28, row + 0.28], color=colour,
linewidth=HOUSE.DATA, zorder=3, solid_capstyle="butt")
axes.set_yticks(positions)
axes.set_yticklabels(
[f"{g} ({int(n)})" for g, n in zip(summary.index, summary["pairs"])],
fontsize=HOUSE.TICK)
axes.invert_yaxis()
axes.set_xlim(-0.03, 1.05)
axes.axvline(1.0, color=HOUSE.GREY, linewidth=HOUSE.REFERENCE,
linestyle=":", zorder=1)
axes.set_xlabel("share of a gene's guides agreeing on the sign")
axes.set_ylabel("")
axes.text(0.02, 0.02, "one point per measurement · line is the mean",
transform=axes.transAxes, fontsize=HOUSE.NOTE,
color=HOUSE.GREY_DARK, va="bottom")
axes.set_title(title or "do a gene's own guides agree?",
fontsize=HOUSE.LABEL)
figure.tight_layout()
_write(figure, path)
return figure
def _one_level(table: pd.DataFrame, level: Optional[str]) -> str:
"""Which level to draw, given what was asked and what the table holds.
Shared by every picture here for the reason `plot_sweep` states: a gene
row and its own guide rows drawn together are the same effect counted
several times, which reads as agreement between independent things.
"""
drawn = str(level or "").strip().lower()
if "level" not in table.columns:
return ""
if table["level"].nunique() > 1:
return drawn or "gene"
return drawn if drawn else ""
[docs]
def plot_grid_volcano(result: "SweepResult", path: Optional[str] = None, *,
alpha: float = 0.05, title: str = "",
level: Optional[str] = None):
"""Plot effect against -log10(p) for every gene x measurement pair.
This shows the whole grid, which :func:`plot_sweep` does not because it
draws only survivors. A screen where most pairs are significant looks
different here from one with a few strong effects.
Pairs past the correction are coloured by direction (green raises the
measurement, rust lowers it) and the rest are grey. When circularity is
known, significant pairs with circularity of at least 0.15 are ringed,
because the classifier already tracks them; when it is not known, the
legend says that circularity was not computed.
"""
import matplotlib.pyplot as plt
keep = result.table
drawn = _one_level(keep, level)
if drawn:
keep = keep[keep["level"] == drawn]
effect_values = pd.to_numeric(keep["effect"], errors="coerce")
p_values = pd.to_numeric(keep["p"], errors="coerce")
keep = keep[np.isfinite(effect_values) & np.isfinite(p_values)]
if not len(keep):
return None
effect = keep["effect"].to_numpy(dtype=float)
evidence = -np.log10(np.clip(keep["p"].to_numpy(dtype=float), 1e-300, 1.0))
with figure_style(theme_target()):
figure, axes = plt.subplots(figsize=(6.2, 4.6))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, lambda: pd.DataFrame({"effect": np.asarray(effect, dtype=float), "evidence": np.asarray(evidence, dtype=float)}), x="effect", y="evidence", kind="scatter")
passed = (keep["q"] < float(alpha)).to_numpy()
up = passed & (effect > 0)
down = passed & (effect < 0)
axes.scatter(effect, evidence, s=4.0, color=HOUSE.GREY, edgecolor="none",
rasterized=True, zorder=1)
axes.scatter(effect[up], evidence[up], s=5.0, color=HOUSE.GREEN,
edgecolor="none", zorder=3)
axes.scatter(effect[down], evidence[down], s=5.0, color=HOUSE.RUST,
edgecolor="none", zorder=3)
if result.circularity_known:
circular = passed & (
pd.to_numeric(keep["circularity"], errors="coerce").to_numpy()
>= 0.15)
if circular.any():
axes.scatter(effect[circular], evidence[circular], s=26,
facecolor="none", edgecolor=HOUSE.NAVY,
linewidth=0.7, zorder=4)
if passed.any():
cut = float(keep.loc[passed, "p"].max())
axes.axhline(-np.log10(max(cut, 1e-300)), color=HOUSE.GREY_DARK,
linewidth=HOUSE.REFERENCE, linestyle=":", zorder=2)
if passed.any():
best = keep.loc[passed].nsmallest(6, "q")
for _i, row in best.iterrows():
axes.annotate(f"{row['guide']} · {str(row['measurement'])[:22]}",
(float(row["effect"]),
-np.log10(max(float(row["p"]), 1e-300))),
fontsize=5.4, style="italic",
color=HOUSE.GREY_DARK, xytext=(3, 1),
textcoords="offset points", zorder=5)
marks = [("not significant", HOUSE.GREY),
(f"raises it (n={int(up.sum())})", HOUSE.GREEN),
(f"lowers it (n={int(down.sum())})", HOUSE.RUST)]
if result.circularity_known:
marks.append(("ringed: the score already tracks it", HOUSE.NAVY))
else:
marks.append(("circularity NOT computed", HOUSE.GREY_DARK))
for i, (text, colour) in enumerate(marks):
axes.text(0.02, 0.97 - i * 0.062, text, transform=axes.transAxes,
fontsize=HOUSE.NOTE, color=colour, va="top", ha="left")
axes.set_xlabel("effect on the measurement")
axes.set_ylabel("-log$_{10}$(p)")
axes.set_title(title or f"{len(keep):,} gene x measurement pair(s)",
fontsize=HOUSE.LABEL)
_readable(figure, axes)
figure.tight_layout()
_write(figure, path)
return figure
[docs]
def plot_gene_profile(result: "SweepResult", gene: Any,
path: Optional[str] = None, *, alpha: float = 0.05,
top: int = 24, title: str = ""):
"""Plot one gene's effects on its measurements, largest first.
Shows the measurements past the correction for ``gene`` (or, when none
pass, its largest effects), ordered by absolute effect, for example
"GRA14 raises pathogen intensity and lowers pathogen area".
Each bar is labelled with its measurement family, so a profile drawn from
one family can be read as one finding rather than as many.
"""
import matplotlib.pyplot as plt
name = str(gene)
table = result.table
mine = table[table["guide"].astype(str) == name]
if not len(mine):
return None
passed = mine[mine["q"] < float(alpha)]
shown = passed if len(passed) else mine
shown = shown.reindex(
shown["effect"].abs().sort_values(ascending=False).index).head(top)
if not len(shown):
return None
families = [measurement_family(m) for m in shown["measurement"]]
passed_here = (shown["q"] < float(alpha)).to_numpy()
signs = np.sign(shown["effect"].to_numpy(dtype=float))
palette = [
(HOUSE.GREEN if sign > 0 else HOUSE.RUST) if ok else HOUSE.GREY
for ok, sign in zip(passed_here, signs)]
with figure_style(theme_target()):
figure, axes = plt.subplots(
figsize=(7.0, max(3.0, 0.30 * len(shown) + 1.4)))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, shown, y="effect", kind="bar")
positions = range(len(shown))
axes.barh(list(positions), shown["effect"].to_numpy(dtype=float),
color=palette, height=0.68, linewidth=0)
axes.axvline(0.0, color=HOUSE.GREY_DARK, linewidth=HOUSE.REFERENCE)
axes.set_yticks(list(positions))
axes.set_yticklabels(
[f"{str(m)[:38]} · {fam}"
for m, fam in zip(shown["measurement"], families)],
fontsize=HOUSE.TICK)
axes.invert_yaxis()
axes.set_xlabel("effect (partial correlation, within plate)", fontsize=8)
axes.tick_params(labelsize=7)
for i, (text, colour) in enumerate((
("raises it", HOUSE.GREEN), ("lowers it", HOUSE.RUST),
(f"not past BH at {alpha:g}", HOUSE.GREY))):
axes.text(0.98, 0.03 + i * 0.055, text, transform=axes.transAxes,
fontsize=HOUSE.NOTE, color=colour, ha="right", va="bottom")
axes.set_title(
title or (f"{name} — {len(passed):,} measurement(s) past BH at "
f"{alpha:g}" if len(passed) else
f"{name} — NOTHING past BH at {alpha:g}; the strongest "
f"effects are shown"),
fontsize=9)
_readable(figure, axes)
figure.tight_layout()
_write(figure, path)
return figure
[docs]
def plot_gene_similarity(result: "SweepResult", path: Optional[str] = None, *,
alpha: float = 0.05, top: int = 30,
title: str = "", level: Optional[str] = None):
"""Plot the correlation between genes' effect profiles as a heatmap.
Two genes in one pathway are expected to move the same measurements in
the same direction. The other plots read each gene on its own; this one
compares genes, which shows whether a hit list contains one shared
phenotype or several.
Profiles are correlated across the whole effect row, not only the
significant entries, so a shared sub-threshold pattern contributes to
the similarity.
"""
import matplotlib.pyplot as plt
keep = result.survivors(alpha=alpha)
drawn = _one_level(keep, level)
if drawn:
keep = keep[keep["level"] == drawn]
if not len(keep):
return None
ranked = (keep.groupby("guide")["q"].min().sort_values().head(top).index)
genes = [g for g in ranked if g in result.effects.index]
if len(genes) < 2:
return None
profiles = result.effects.loc[genes].to_numpy(dtype=float)
profiles = np.nan_to_num(profiles, nan=0.0)
spread = profiles.std(axis=1, keepdims=True)
spread[spread <= 0] = 1.0
centred = (profiles - profiles.mean(axis=1, keepdims=True)) / spread
similarity = (centred @ centred.T) / max(profiles.shape[1], 1)
frame = pd.DataFrame(similarity, index=genes, columns=genes)
frame = _order_like_neighbours(frame)
size = max(3.6, 0.26 * len(frame.index) + 2.0)
with figure_style(theme_target()):
figure, axes = plt.subplots(figsize=(size, size))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, frame, kind="heatmap", matrix=True)
image = axes.imshow(frame.to_numpy(), cmap="RdBu_r", vmin=-1.0, vmax=1.0)
axes.set_xticks(range(len(frame.columns)))
axes.set_xticklabels(frame.columns, rotation=90, fontsize=7)
axes.set_yticks(range(len(frame.index)))
axes.set_yticklabels(frame.index, fontsize=7)
axes.set_title(title or f"do these {len(frame.index)} genes behave alike?",
fontsize=9)
bar = figure.colorbar(image, ax=axes, fraction=0.035, pad=0.02)
bar.set_label("correlation of effect profiles", fontsize=7)
bar.ax.tick_params(labelsize=6)
_readable(figure, axes)
figure.tight_layout()
_write(figure, path)
return figure
[docs]
def plot_measurement_hits(result: "SweepResult", path: Optional[str] = None,
*, alpha: float = 0.05, top: int = 26,
title: str = "", level: Optional[str] = None):
"""Plot measurements by how many genes move them past the correction.
This reads the grid along its measurement axis. A measurement moved by
half the library does not discriminate between genes; it more likely
reflects a plate effect, focus drift or confluence artefact, and it adds
a hit to many genes. Measurements moved by at least half of the
significant genes are ringed.
"""
import matplotlib.pyplot as plt
keep = result.survivors(alpha=alpha)
drawn = _one_level(keep, level)
if drawn:
keep = keep[keep["level"] == drawn]
if not len(keep):
return None
counts = keep.groupby("measurement")["guide"].nunique().sort_values(
ascending=False).head(top)
if not len(counts):
return None
total = int(keep["guide"].nunique())
share = counts.to_numpy(dtype=float) / max(total, 1)
strength = keep.groupby("measurement")["effect"].apply(
lambda v: float(np.nanmedian(np.abs(v)))).reindex(counts.index)
evidence = keep.groupby("measurement")["q"].min().reindex(counts.index)
evidence = -np.log10(np.clip(evidence.to_numpy(dtype=float), 1e-300, 1.0))
with figure_style(theme_target()):
figure, axes = plt.subplots(
figsize=(6.6, max(2.8, 0.28 * len(counts) + 1.4)))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, lambda: pd.DataFrame({"measurement": [str(v) for v in counts.index], "hits": np.asarray(counts, dtype=float), "strength": np.asarray(strength, dtype=float)}), x="strength", y="hits", kind="scatter")
rows = np.arange(len(counts))
sizes = 12.0 + 78.0 * (counts.to_numpy(dtype=float) / max(counts.max(), 1))
dots = axes.scatter(strength.to_numpy(dtype=float), rows, s=sizes,
c=evidence, cmap=HOUSE.SEQ, edgecolor="none",
zorder=3)
loud = share >= 0.5
if loud.any():
axes.scatter(strength.to_numpy(dtype=float)[loud], rows[loud],
s=sizes[loud], facecolor="none", edgecolor=HOUSE.RUST,
linewidth=0.8, zorder=4)
axes.set_yticks(rows)
axes.set_yticklabels([str(m)[:46] for m in counts.index],
fontsize=HOUSE.TICK)
axes.invert_yaxis()
axes.set_xlabel("median |effect| of the genes that move it")
bar = figure.colorbar(dots, ax=axes, fraction=0.03, pad=0.01)
bar.set_label("-log$_{10}$(q) of the best gene", fontsize=HOUSE.NOTE)
bar.ax.tick_params(labelsize=HOUSE.NOTE - 0.6)
bar.outline.set_visible(False)
axes.text(0.98, 0.02,
"dot size = genes moving it" + (
" · ringed: moved by half the library" if loud.any()
else ""),
transform=axes.transAxes, fontsize=HOUSE.NOTE,
color=HOUSE.GREY_DARK, va="bottom", ha="right")
axes.set_title(title or "which measurements discriminate",
fontsize=HOUSE.LABEL)
figure.tight_layout()
_write(figure, path)
return figure
[docs]
def plot_circularity(result: "SweepResult", path: Optional[str] = None, *,
alpha: float = 0.05, title: str = "",
level: Optional[str] = None):
"""Plot whether each surviving measurement adds independent evidence.
A measurement the classifier already tracks cannot corroborate a result
derived from that classifier. The plot exposes that dependence rather
than presenting a correlated measurement as separate confirmation.
Every point is drawn because the correlation cutoff is a judgment. This
lets users see where their hits fall and choose a threshold appropriate
to their screen.
"""
import matplotlib.pyplot as plt
if not result.circularity_known:
return None
keep = result.survivors(alpha=alpha)
drawn = _one_level(keep, level)
if drawn:
keep = keep[keep["level"] == drawn]
keep = keep[np.isfinite(keep["circularity"]) & np.isfinite(keep["effect"])]
if not len(keep):
return None
with figure_style(theme_target()):
figure, axes = plt.subplots(figsize=(7.0, 5.0))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, lambda: keep.assign(abs_effect=np.abs(keep["effect"].to_numpy(dtype=float))), x="abs_effect", y="circularity", kind="scatter")
circular = keep["circularity"].to_numpy(dtype=float)
above = circular >= 0.15
magnitude = np.abs(keep["effect"].to_numpy(dtype=float))
axes.scatter(magnitude[~above], circular[~above], s=11, color=HOUSE.GREY,
edgecolor="none", zorder=2)
axes.scatter(magnitude[above], circular[above], s=13, color=HOUSE.RUST,
edgecolor="none", zorder=3)
axes.axhline(0.15, color=HOUSE.GREY_DARK, linewidth=HOUSE.REFERENCE,
linestyle=":", zorder=1)
axes.text(0.98, 0.15, "0.15 — a working bar, not a law ",
transform=axes.get_yaxis_transform(), fontsize=HOUSE.NOTE,
ha="right", va="bottom", color=HOUSE.GREY_DARK)
axes.set_xlabel("|effect| of the gene on the measurement", fontsize=8)
axes.set_ylabel("|rho(classification score, measurement)|", fontsize=8)
axes.tick_params(labelsize=7)
above = int((circular >= 0.15).sum())
axes.set_title(
title or (f"{above:,} of {len(keep):,} surviving pair(s) sit on a "
f"measurement the score already tracks"), fontsize=9)
_readable(figure, axes)
figure.tight_layout()
_write(figure, path)
return figure
[docs]
def plot_calibration(result: "SweepResult", path: Optional[str] = None, *,
title: str = "", level: Optional[str] = None):
"""Plot observed against expected -log10(p) to check calibration.
The observed P values against the uniform they would follow if nothing
were real. A grid that hugs the diagonal has no signal; one that lifts
off it at the left has some; one that lifts off everywhere has a
systematic effect -- a plate term that did not get blocked out, or an
aggregation that correlated every well with itself.
The title reports the inflation factor lambda from the median P value;
0.9 to 1.15 is labelled calibrated. Check this plot before the others:
with inflated P values their significant results are not reliable.
"""
import matplotlib.pyplot as plt
keep = result.table
drawn = _one_level(keep, level)
if drawn:
keep = keep[keep["level"] == drawn]
values = keep["p"].to_numpy(dtype=float)
values = values[np.isfinite(values)]
if values.size < 2:
return None
observed = np.sort(np.clip(values, 1e-300, 1.0))
expected = (np.arange(1, observed.size + 1) - 0.5) / observed.size
with figure_style(theme_target()):
figure, axes = plt.subplots(figsize=(5.6, 5.4))
from .figures.bundle import _register_figure_data
_register_figure_data(figure, lambda: pd.DataFrame({"expected_-log10_p": -np.log10(np.asarray(expected, dtype=float)), "observed_-log10_p": -np.log10(np.asarray(observed, dtype=float))}), x="expected_-log10_p", y="observed_-log10_p", kind="scatter")
axes.plot(-np.log10(expected), -np.log10(observed), ".", markersize=2.6,
color=HOUSE.GREY, zorder=2)
edge = float(max(-np.log10(expected).max(), -np.log10(observed).max()))
axes.plot([0, edge], [0, edge], color=HOUSE.GREY_DARK,
linewidth=HOUSE.REFERENCE, linestyle=":", zorder=1)
from scipy.stats import chi2
median = float(np.median(observed))
lam = (chi2.isf(median, 1) / chi2.isf(0.5, 1)) if median > 0 else np.nan
axes.set_xlabel("expected -log10(p)", fontsize=8)
axes.set_ylabel("observed -log10(p)", fontsize=8)
axes.tick_params(labelsize=7)
axes.text(0.03, 0.95, "dotted: no effect anywhere",
transform=axes.transAxes, fontsize=HOUSE.NOTE,
color=HOUSE.GREY_DARK, va="top")
axes.set_title(
title or (f"calibration — lambda = {lam:.2f} "
f"({'calibrated' if 0.9 <= lam <= 1.15 else 'inflated' if lam > 1.15 else 'conservative'})"),
fontsize=9)
_readable(figure, axes)
figure.tight_layout()
_write(figure, path)
return figure