"""Build ranked, annotated hit lists from regression results.
A regression run writes full, level-specific, and selected coefficient tables
to a uniquely named ``results/<kind>[_n]`` directory. This module combines the
available tables into one sortable row per gene, adds effect estimates and
uncertainty when the backend provides them, calculates guide-direction
agreement, and joins optional gene metadata.
Metadata joins are validated as many-to-one. Repeated transcript records are
collapsed before the join so they cannot duplicate a gene in the hit list.
Backends without frequentist p-values are ranked by bootstrap selection
frequency; other supported backends are ranked by q-value.
The module is headless and reads files already written by a regression run.
Public API::
from spacr.hits import build_hit_list, load_results
hits = build_hit_list("/data/plate7/results/ols_2")
strong = hits.filter(max_q=0.05, min_agreement=0.66, min_guides=2)
strong.write_csv("/tmp/hits.csv")
print(strong.to_markdown(limit=20))
"""
from __future__ import annotations
import html
import math
import os
import re
from dataclasses import dataclass, field
from typing import (Any, Callable, Dict, Iterable, List, Mapping, Optional,
Sequence, Tuple, Union)
import numpy as np
import pandas as pd
from . import tabular
__all__ = [
"DEFAULT_ALPHA",
"Hit",
"HitList",
"NO_P_VALUE_TYPES",
"RESULT_FILES",
"benjamini_hochberg",
"build_hit_list",
"coefficient_levels",
"family_labels",
"gene_of",
"grna_agreement",
"guide_of",
"join_metadata",
"load_gene_metadata",
"load_results",
"tested_family",
]
#: The files :func:`spacr.ml.perform_regression` writes into its results
#: folder, by the role this module needs them for.
RESULT_FILES: Dict[str, str] = {
"all": "results.csv",
"gene": "results_gene.csv",
"grna": "results_grna.csv",
"significant": "results_significant.csv",
}
#: Backends that report a coefficient but no frequentist p-value, so a
#: q-value would be a correction applied to a number that is not a p-value.
#: Mirrors :data:`spacr.ml.NO_P_VALUE_TYPES`; kept as a literal so importing
#: this module does not drag torch in, and asserted equal in the test suite.
NO_P_VALUE_TYPES: Tuple[str, ...] = ("lasso", "elasticnet", "group_lasso")
#: The coefficient families a run can fit, in the order a reader is offered
#: them: the guide is the unit the screen measures, the gene is what the
#: guides are evidence about. ``level='both'`` fits one of each.
COEFFICIENT_LEVELS: Tuple[str, ...] = ("grna", "gene")
#: The FDR a hit list defaults to calling a hit at.
DEFAULT_ALPHA = 0.05
#: The bootstrap selection frequency a hit list defaults to calling a hit at
#: for the penalised backends, which have no p-value to correct.
#: :data:`DEFAULT_ALPHA` cannot serve for both: a q-value of 0.05 is strict
#: and a selection frequency of 0.05 is "chosen in one bootstrap in twenty",
#: which would call almost every gene a hit and report it as significant.
#: Matches ``lasso_selection_threshold``'s own default in
#: :func:`spacr.ml.perform_regression`.
DEFAULT_SELECTION_THRESHOLD = 0.6
#: Flags a row can carry. Each one is a reason to look twice, not a reason to
#: drop the row — dropping it would hide the very thing the flag is for.
FLAG_CONTROL = "control"
FLAG_SINGLE_GUIDE = "single-guide"
FLAG_GUIDES_DISAGREE = "guides-disagree"
FLAG_NO_GUIDES = "no-guide-rows"
FLAG_NO_METADATA = "unannotated"
#: What each flag means, for a legend under the table.
FLAG_MEANING: Dict[str, str] = {
FLAG_CONTROL: "a control gRNA or gene, not a screen candidate",
FLAG_SINGLE_GUIDE: "called by one guide, so nothing corroborates it",
FLAG_GUIDES_DISAGREE: "fewer than half of this gene's guides agree in sign",
FLAG_NO_GUIDES: "no per-gRNA rows for this gene, so agreement is unknown",
FLAG_NO_METADATA: "no row in any metadata file matched this gene",
}
_BRACKET = re.compile(r"\[(.*?)\]")
#: Design-matrix terms that are covariates rather than hypotheses: the
#: intercept and the explicit plate, row, column, and screen effects fitted
#: to soak up layout and experiment artefacts. Match term prefixes so a real
#: guide whose identifier contains ``row`` or ``column`` remains testable.
NUISANCE_TERMS = re.compile(
r"^(?:Intercept$|C\(.+\)\[[^]]+\]$|"
r"(?:plateID|rowID|columnID|screenID)(?:\[|$))",
re.IGNORECASE,
)
[docs]
def tested_family(features: Iterable[Any]) -> np.ndarray:
"""Identify coefficient terms included in multiple testing.
Parameters
----------
features : iterable of Any
Design-matrix term names.
Returns
-------
numpy.ndarray of bool
Boolean mask aligned with ``features``. Guide and gene terms are
``True``; intercept and layout nuisance terms are ``False``.
Notes
-----
The mask follows the family corrected by
:func:`spacr.ml.perform_regression`. Nuisance terms are fitted as
covariates but are excluded from hit plots and multiple-testing
correction.
Examples
--------
>>> tested_family(["Intercept", "fraction:grna[233460_1]"]).tolist()
[False, True]
"""
series = pd.Series([str(value) for value in features], dtype=object)
if series.empty:
return np.zeros(0, dtype=bool)
return ~series.str.contains(NUISANCE_TERMS, regex=True).to_numpy(dtype=bool)
[docs]
def family_labels(features: Iterable[Any]) -> np.ndarray:
"""Label the multiple-testing family of each coefficient term.
Parameters
----------
features : iterable of Any
Design-matrix term names.
Returns
-------
numpy.ndarray of object
Labels aligned with ``features``. Values are ``'grna'`` for guide
terms, ``'gene'`` for gene terms, and ``''`` for nuisance terms.
Notes
-----
Runs with ``level='both'`` contain separate guide and gene testing
families. Corrections should be applied within each non-empty label rather
than across the pooled coefficient table. Explicit ``:gene[...]`` and
``:grna[...]`` labels take precedence over identifier shape.
Examples
--------
>>> family_labels(["Intercept", "fraction:grna[233460_1]",
... "gene_fraction:gene[233460]"]).tolist()
['', 'grna', 'gene']
"""
series = pd.Series([str(value) for value in features], dtype=object)
if series.empty:
return np.zeros(0, dtype=object)
labels = np.full(len(series), "gene", dtype=object)
for index, term in enumerate(series):
if NUISANCE_TERMS.search(term):
labels[index] = ""
continue
lower = term.lower()
if ":gene[" in lower:
continue
if ":grna[" in lower:
labels[index] = "grna"
continue
match = _BRACKET.search(term)
if match and _is_guide_token(match.group(1).removeprefix("T.")):
labels[index] = "grna"
return labels
[docs]
def coefficient_levels(frame: Optional[pd.DataFrame]) -> pd.Series:
"""Identify the model level associated with each coefficient row.
Parameters
----------
frame : pandas.DataFrame or None
Coefficient table. ``None`` returns an empty series. If neither
``level`` nor ``feature`` is present, every row is assigned ``''``.
Returns
-------
pandas.Series
``'grna'``, ``'gene'``, or ``''`` for each row, indexed like
``frame``. An empty value denotes a nuisance term or a term whose
level cannot be inferred.
Notes
-----
The explicit ``level`` column is preferred. Feature-name inference
supports tables written before that column was introduced. This
distinction matters for ``level='both'`` fits because both models contain
an ``Intercept`` whose name alone does not identify its model.
Examples
--------
>>> import pandas as pd
>>> coefficient_levels(pd.DataFrame(
... {"feature": ["Intercept", "fraction:grna[233460_1]"],
... "level": ["grna", "grna"]})).tolist()
['grna', 'grna']
>>> coefficient_levels(pd.DataFrame(
... {"feature": ["Intercept", "gene_fraction:gene[233460]"]})).tolist()
['', 'gene']
"""
if frame is None:
return pd.Series([], dtype=object)
columns = getattr(frame, "columns", ())
if "feature" in columns:
levels = pd.Series(family_labels(frame["feature"]), index=frame.index,
dtype=object)
else:
levels = pd.Series([""] * len(frame), index=frame.index, dtype=object)
if "level" in columns:
recorded = frame["level"].astype(object).map(
lambda value: str(value).strip().lower()
if value is not None and value == value else "")
known = recorded.isin(COEFFICIENT_LEVELS)
levels = levels.where(~known, recorded)
return levels
[docs]
def gene_of(feature: Any) -> Optional[str]:
"""Extract the gene identifier from a model term.
Parameters
----------
feature : Any
Design-matrix term containing an identifier in square brackets.
Returns
-------
str or None
Gene identifier, or ``None`` when the term contains no bracketed
identifier. Guide suffixes are removed, while the strain prefix and
numeric portion of a VEuPathDB accession are retained.
Examples
--------
``gene_fraction:gene[233460]`` and ``fraction:grna[233460_1]`` both map
to ``233460``. ``fraction:grna[TGGT1_231640_3]`` maps to
``TGGT1_231640``.
"""
if feature is None or (isinstance(feature, float) and math.isnan(feature)):
return None
match = _BRACKET.search(str(feature))
if not match:
return None
token = re.sub(r"^T\.", "", match.group(1))
return _gene_id_of(token)
_GENE_ID_PREFIXES = ("TGGT1", "TGME49", "TGVEG", "TGRH88", "TGARI",
"TGCAST", "TGP89", "TGCOUG", "TGMAS", "TGFOU",
"PF3D7", "PBANKA", "PY17X", "PCHAS", "PKNH", "PVP01",
"CPATCC", "CHUDEA", "CPBGF", "NCLIV", "BBOV", "TA",
"ETH", "EHXH", "CSUI")
def _gene_id_of(token: Any) -> Optional[str]:
"""Normalize a bracketed gene or guide token to its gene identifier.
Numeric guide tokens lose their trailing guide suffix. VEuPathDB
accessions retain the ``prefix_number`` gene identifier and lose only an
optional suffix after that identifier.
Examples
--------
``233460_1`` becomes ``233460`` and ``TGGT1_231640_3`` becomes
``TGGT1_231640``.
"""
text = str(token or "").strip()
if not text:
return None
parts = text.split("_")
if len(parts) > 1 and parts[0].upper() in _GENE_ID_PREFIXES:
return "_".join(parts[:2])
return parts[0] or None
def _is_guide_token(token: Any) -> bool:
"""Whether ``token`` adds a guide suffix to its normalized gene id."""
text = str(token or "").strip()
gene = _gene_id_of(text)
return bool(text and gene and gene != text)
[docs]
def guide_of(feature: Any) -> Optional[str]:
"""Extract the guide identifier from a model term.
Parameters
----------
feature : Any
Design-matrix term containing an identifier in square brackets.
Returns
-------
str or None
Complete guide identifier, or ``None`` for gene and nuisance terms.
Explicit ``:gene[...]`` and ``:grna[...]`` labels take precedence
over identifier shape, so an underscore within a VEuPathDB gene
accession is not mistaken for a guide suffix.
"""
if feature is None or (isinstance(feature, float) and math.isnan(feature)):
return None
feature_text = str(feature)
match = _BRACKET.search(feature_text)
if not match:
return None
token = re.sub(r"^T\.", "", match.group(1))
if re.search(r":gene\[", feature_text, flags=re.IGNORECASE):
return None
if re.search(r":grna\[", feature_text, flags=re.IGNORECASE):
return token
return token if _is_guide_token(token) else None
[docs]
def benjamini_hochberg(p_values: Sequence[Any]) -> np.ndarray:
"""Benjamini-Hochberg FDR q-values for a vector of p-values.
The step-up procedure, with the monotonicity enforced by the running
minimum from the largest p-value down — without it a q-value can come out
smaller than one belonging to a smaller p-value, which reads as a hit
ranking that disagrees with itself.
``NaN`` p-values (a term the backend could not test) stay ``NaN`` and are
excluded from ``m``: correcting for a test that was not run inflates every
other q-value.
:param p_values: p-values; anything non-finite is treated as untested.
:returns: q-values aligned with the input.
"""
values = np.asarray(p_values, dtype=float)
q = np.full(values.shape, np.nan, dtype=float)
testable = np.isfinite(values)
m = int(testable.sum())
if m == 0:
return q
order = np.argsort(values[testable], kind="mergesort")
ranked = values[testable][order]
scaled = ranked * m / np.arange(1, m + 1)
adjusted = np.minimum.accumulate(scaled[::-1])[::-1]
adjusted = np.clip(adjusted, 0.0, 1.0)
out = np.empty(m, dtype=float)
out[order] = adjusted
q[testable] = out
return q
[docs]
def grna_agreement(gene_effects: Mapping[str, float],
grna_frame: Optional[pd.DataFrame]
) -> Dict[str, Tuple[int, int, List[str]]]:
"""How many of each gene's guides push the same way as the gene.
:param gene_effects: ``{gene id: gene-level coefficient}``.
:param grna_frame: the per-gRNA coefficient table (``feature`` and
``coefficient`` columns). ``None`` or empty means "no guide-level
evidence", which is reported as such rather than as agreement.
:returns: ``{gene id: (n_agree, n_guides, [guide ids that agree])}``. A
guide whose coefficient is exactly zero counts as a guide but agrees
with nothing — a penalised backend shrinks non-contributing guides to
zero, and counting those as agreement would turn a lasso's sparsity
into corroboration.
"""
result: Dict[str, Tuple[int, int, List[str]]] = {
gene: (0, 0, []) for gene in gene_effects}
if grna_frame is None or grna_frame.empty:
return result
if "feature" not in grna_frame.columns:
return result
per_gene: Dict[str, List[Tuple[str, float]]] = {}
for _, row in grna_frame.iterrows():
gene = gene_of(row.get("feature"))
if gene is None:
continue
guide = row.get("grna")
if guide is None or (isinstance(guide, float) and math.isnan(guide)):
match = _BRACKET.search(str(row.get("feature", "")))
guide = re.sub(r"^T\.", "", match.group(1)) if match else ""
try:
coefficient = float(row.get("coefficient"))
except (TypeError, ValueError):
continue
if not math.isfinite(coefficient):
continue
per_gene.setdefault(gene, []).append((str(guide), coefficient))
for gene, guides in per_gene.items():
target = gene_effects.get(gene)
if target is None or not math.isfinite(float(target)):
result[gene] = (0, len(guides), [])
continue
wanted = math.copysign(1.0, float(target))
agree = [name for name, value in guides
if value != 0.0 and math.copysign(1.0, value) == wanted]
result[gene] = (len(agree), len(guides), sorted(agree))
return result
[docs]
def load_results(folder: Union[str, os.PathLike]) -> Dict[str, pd.DataFrame]:
"""Read the coefficient tables a regression results folder holds.
:param folder: A ``results/<kind>[_n]`` directory written by
:func:`spacr.ml.perform_regression`.
:returns: ``{role: DataFrame}`` for whichever of :data:`RESULT_FILES`
exist. A folder with none of them yields an empty dict rather than an
exception — "that is not a results folder" is something the caller
must be able to say to a user.
:raises FileNotFoundError: when ``folder`` does not exist at all.
"""
root = os.path.abspath(os.path.expanduser(os.fspath(folder)))
if not os.path.isdir(root):
raise FileNotFoundError(f"no results folder at {root}")
found: Dict[str, pd.DataFrame] = {}
for role, name in RESULT_FILES.items():
path = os.path.join(root, name)
if os.path.isfile(path):
try:
found[role] = tabular.read_table(path, report=None)
except (pd.errors.EmptyDataError, pd.errors.ParserError):
continue
return found
@dataclass(frozen=True)
[docs]
class Hit:
"""One gene, everything known about it, and why to believe it.
:param gene: the gene id parsed from the model term.
:param feature: the model term itself, so a row can be traced back.
:param effect: the fitted coefficient — the effect size.
:param std_err: its standard error, when the backend reported one.
:param ci_low: lower bound of the 95% interval; ``nan`` without an error.
:param ci_high: upper bound of the same.
:param p_value: the reported p-value; ``nan`` for a backend that has none.
:param q_value: Benjamini-Hochberg FDR over the genes tested here.
:param selection_frequency: bootstrap selection frequency, for the
penalised backends that rank by it instead of by a p-value.
:param n_guides: how many of this gene's guides the per-gRNA table holds.
:param n_agree: how many of them push the same way as the gene effect.
:param agreement: ``n_agree / n_guides``; ``nan`` with no guide rows.
:param agreeing_guides: which guides those are.
:param n_obs: observations behind the gene term (well x guide rows).
:param condition: ``nc`` / ``pc`` / ``control`` / ``other`` as the
regression labelled it.
:param direction: ``up`` or ``down``, from the sign of the effect.
:param rank: 1-based position in the list as ranked.
:param flags: reasons to look twice; see :data:`FLAG_MEANING`.
:param annotation: the metadata columns that joined onto this gene.
"""
gene: str
feature: str = ""
effect: float = float("nan")
std_err: float = float("nan")
ci_low: float = float("nan")
ci_high: float = float("nan")
p_value: float = float("nan")
q_value: float = float("nan")
selection_frequency: float = float("nan")
n_guides: int = 0
n_agree: int = 0
agreement: float = float("nan")
agreeing_guides: Tuple[str, ...] = ()
n_obs: int = 0
condition: str = ""
direction: str = ""
rank: int = 0
flags: Tuple[str, ...] = ()
annotation: Dict[str, Any] = field(default_factory=dict)
@property
[docs]
def name(self) -> str:
"""The most human name available: an annotated one, else the id."""
for column in ("Gene Name", "gene_name", "Name", "Product Description",
"product", "description"):
value = self.annotation.get(column)
if value is not None and str(value).strip() and str(value) != "nan":
return str(value).strip()
return self.gene
[docs]
def to_dict(self) -> Dict[str, Any]:
"""A flat, JSON-serializable row: the fields, then the annotation."""
row: Dict[str, Any] = {
"rank": self.rank, "gene": self.gene, "name": self.name,
"effect": self.effect, "std_err": self.std_err,
"ci_low": self.ci_low, "ci_high": self.ci_high,
"p_value": self.p_value, "q_value": self.q_value,
"selection_frequency": self.selection_frequency,
"n_guides": self.n_guides, "n_agree": self.n_agree,
"agreement": self.agreement,
"agreeing_guides": ";".join(self.agreeing_guides),
"n_obs": self.n_obs, "condition": self.condition,
"direction": self.direction, "flags": ";".join(self.flags),
"feature": self.feature,
}
for column, value in self.annotation.items():
row.setdefault(column, value)
return row
@dataclass(frozen=True)
[docs]
class HitList:
"""A ranked list of hits plus everything needed to interpret it.
:param hits: the rows, already ranked.
:param source: the results folder or a description of where it came from.
:param regression_type: the backend, when it could be determined.
:param ranking: how the list is ordered — ``"q-value"`` or
``"selection-frequency"``.
:param alpha: the FDR the ``significant`` count is taken at.
:param n_terms: how many model terms the source table held.
:param n_genes: how many distinct genes were tested.
:param filters: the filters applied to reach this list, as data.
:param notes: metadata collapses, missing files, backend caveats.
"""
hits: Tuple[Hit, ...] = ()
source: str = ""
regression_type: str = ""
ranking: str = "q-value"
alpha: float = DEFAULT_ALPHA
n_terms: int = 0
n_genes: int = 0
filters: Dict[str, Any] = field(default_factory=dict)
notes: Tuple[str, ...] = ()
[docs]
def __len__(self) -> int:
"""How many rows the list holds."""
return len(self.hits)
[docs]
def __iter__(self):
"""Iterate the rows in rank order."""
return iter(self.hits)
[docs]
def __getitem__(self, index):
"""Index or slice the rows; a slice returns a :class:`HitList`."""
if isinstance(index, slice):
return self._with(self.hits[index], dict(self.filters))
return self.hits[index]
[docs]
def gene(self, gene: str) -> Optional[Hit]:
"""The row for one gene id, or ``None``.
:param gene: exact gene identifier to look up.
"""
for hit in self.hits:
if hit.gene == gene:
return hit
return None
[docs]
def significant(self, alpha: Optional[float] = None) -> "HitList":
"""The rows that clear the FDR (or the selection threshold)."""
cut = self.alpha if alpha is None else float(alpha)
if self.ranking == "selection-frequency":
return self.filter(min_selection=cut)
return self.filter(max_q=cut)
[docs]
def top(self, n: int) -> "HitList":
"""The first ``n`` rows, still ranked.
:param n: maximum number of ranked rows to retain.
"""
return self._with(self.hits[:max(0, int(n))],
dict(self.filters, top=int(n)))
[docs]
def filter(self, *,
max_q: Optional[float] = None,
max_p: Optional[float] = None,
min_effect: Optional[float] = None,
min_agreement: Optional[float] = None,
min_guides: Optional[int] = None,
min_selection: Optional[float] = None,
direction: Optional[str] = None,
conditions: Optional[Iterable[str]] = None,
exclude_controls: bool = False,
genes: Optional[Iterable[str]] = None,
query: str = "",
predicate: Optional[Callable[[Hit], bool]] = None,
) -> "HitList":
"""Return a narrowed list. Every argument is optional and ANDed.
A row whose value for a criterion is missing FAILS that criterion
rather than passing it: a gene with no q-value has not been shown to
clear an FDR, and letting missing data through a filter is how an
untested term ends up in a hit list.
:param max_q: keep rows with ``q_value <= max_q``.
:param max_p: keep rows with ``p_value <= max_p``.
:param min_effect: keep rows with ``abs(effect) >= min_effect``.
:param min_agreement: keep rows whose guide agreement is at least this.
:param min_guides: keep rows with at least this many guides.
:param min_selection: keep rows with at least this bootstrap
selection frequency.
:param direction: ``"up"`` or ``"down"``.
:param conditions: keep only these ``condition`` values.
:param exclude_controls: drop ``nc`` / ``pc`` / ``control`` rows.
:param genes: keep only these gene ids.
:param query: case-insensitive substring, matched against the gene
id, the name and every annotation value.
:param predicate: an arbitrary extra test.
:returns: a new :class:`HitList`; the receiver is unchanged.
"""
wanted = set(conditions) if conditions is not None else None
keep_genes = set(genes) if genes is not None else None
needle = query.strip().casefold()
def _ok(hit: Hit) -> bool:
"""Return whether ``hit`` satisfies every supplied filter."""
if max_q is not None and not _at_most(hit.q_value, max_q):
return False
if max_p is not None and not _at_most(hit.p_value, max_p):
return False
if min_effect is not None and not _at_least(hit.effect, min_effect,
absolute=True):
return False
if min_agreement is not None and not _at_least(hit.agreement,
min_agreement):
return False
if min_guides is not None and hit.n_guides < int(min_guides):
return False
if min_selection is not None and not _at_least(
hit.selection_frequency, min_selection):
return False
if direction and hit.direction != direction:
return False
if wanted is not None and hit.condition not in wanted:
return False
if exclude_controls and hit.condition in ("nc", "pc", "control"):
return False
if keep_genes is not None and hit.gene not in keep_genes:
return False
if needle and needle not in _searchable(hit):
return False
if predicate is not None and not predicate(hit):
return False
return True
applied = {
"max_q": max_q, "max_p": max_p, "min_effect": min_effect,
"min_agreement": min_agreement, "min_guides": min_guides,
"min_selection": min_selection, "direction": direction,
"conditions": sorted(wanted) if wanted is not None else None,
"exclude_controls": exclude_controls or None,
"genes": sorted(keep_genes) if keep_genes is not None else None,
"query": query or None,
}
merged = dict(self.filters)
merged.update({k: v for k, v in applied.items() if v is not None})
return self._with(tuple(hit for hit in self.hits if _ok(hit)), merged)
def _with(self, hits: Sequence[Hit], filters: Dict[str, Any]) -> "HitList":
"""A copy carrying different rows, ranks renumbered from 1."""
renumbered = tuple(
Hit(**{**hit.__dict__, "rank": index + 1})
for index, hit in enumerate(hits))
return HitList(
hits=renumbered, source=self.source,
regression_type=self.regression_type, ranking=self.ranking,
alpha=self.alpha, n_terms=self.n_terms, n_genes=self.n_genes,
filters=filters, notes=self.notes)
[docs]
def columns(self) -> List[str]:
"""Column order for the table forms, annotation columns last."""
base = ["rank", "gene", "name", "effect", "std_err", "ci_low",
"ci_high", "p_value", "q_value", "selection_frequency",
"n_guides", "n_agree", "agreement", "n_obs", "condition",
"direction", "flags", "agreeing_guides", "feature"]
extra: List[str] = []
for hit in self.hits:
for column in hit.annotation:
if column not in base and column not in extra:
extra.append(column)
return base + extra
[docs]
def to_frame(self) -> pd.DataFrame:
"""The list as a DataFrame, one row per gene, in rank order."""
rows = [hit.to_dict() for hit in self.hits]
frame = pd.DataFrame(rows, columns=self.columns())
return frame
[docs]
def write_csv(self, path: Union[str, os.PathLike]) -> str:
"""Write the table as CSV and return the path written.
:param path: destination CSV path.
Through :func:`spacr.tabular.write_table`, so a hit list is written
with the same column spellings every spaCR reader expects to find.
"""
target = os.path.abspath(os.path.expanduser(os.fspath(path)))
return tabular.write_table(self.to_frame(), target)
[docs]
def summary(self) -> Dict[str, Any]:
"""Counts and settings, for a header line or a run digest.
This is the block :mod:`spacr.methods_export` puts in front of the
model: every number a results paragraph would quote, computed here
rather than by whatever writes the prose.
"""
significant = self.significant()
up = sum(1 for hit in significant if hit.direction == "up")
down = sum(1 for hit in significant if hit.direction == "down")
corroborated = sum(1 for hit in significant
if hit.n_guides >= 2 and _at_least(hit.agreement,
0.5))
effects = [abs(hit.effect) for hit in significant
if math.isfinite(hit.effect)]
return {
"source": self.source,
"regression_type": self.regression_type,
"ranking": self.ranking,
"alpha": self.alpha,
"n_terms": self.n_terms,
"n_genes_tested": self.n_genes,
"n_listed": len(self.hits),
"n_significant": len(significant),
"n_up": up,
"n_down": down,
"n_corroborated": corroborated,
"max_abs_effect": max(effects) if effects else float("nan"),
"median_abs_effect": (float(np.median(effects)) if effects
else float("nan")),
"top_genes": [hit.gene for hit in significant[:10]],
"filters": dict(self.filters),
"flag_counts": self.flag_counts(),
"notes": list(self.notes),
}
[docs]
def flag_counts(self) -> Dict[str, int]:
"""``{flag: how many rows carry it}``."""
counts: Dict[str, int] = {}
for hit in self.hits:
for flag in hit.flags:
counts[flag] = counts.get(flag, 0) + 1
return dict(sorted(counts.items()))
[docs]
def to_markdown(self, limit: int = 50) -> str:
"""A Markdown table of the top ``limit`` rows, with its legend.
The form the list travels in: pasted into an email, a lab notebook or
an issue. No trailing newline.
"""
header = [f"# Hit list — {self.source or 'regression'}"]
summary = self.summary()
header.append("")
header.append(
f"{summary['n_significant']} of {summary['n_genes_tested']} genes "
f"tested clear {'a selection frequency of' if self.ranking == 'selection-frequency' else 'FDR'}"
f" {self.alpha:g} "
f"({summary['n_up']} up, {summary['n_down']} down; "
f"{summary['n_corroborated']} corroborated by at least two "
f"guides).")
if self.regression_type:
header.append(f"Model: {self.regression_type}.")
for note in self.notes:
header.append(f"Note: {note}")
header.append("")
shown = self.hits[:max(0, int(limit))]
columns = ["rank", "gene", "name", "effect", "p_value", "q_value",
"guides", "agreement", "flags"]
header.append("| " + " | ".join(columns) + " |")
header.append("|" + "|".join(["---"] * len(columns)) + "|")
for hit in shown:
header.append("| " + " | ".join([
str(hit.rank), hit.gene, hit.name,
_fmt(hit.effect), _fmt(hit.p_value), _fmt(hit.q_value),
f"{hit.n_agree}/{hit.n_guides}",
_fmt(hit.agreement), ", ".join(hit.flags) or "-",
]) + " |")
if len(self.hits) > len(shown):
header.append("")
header.append(f"…and {len(self.hits) - len(shown)} more rows.")
used = self.flag_counts()
if used:
header.append("")
header.append("Flags:")
header.extend(f"* **{flag}** — {FLAG_MEANING.get(flag, flag)} "
f"({count} row(s))"
for flag, count in used.items())
return "\n".join(header)
[docs]
def to_html(self, limit: int = 500) -> str:
"""A standalone HTML table — the form handed to a collaborator.
Self-contained: no stylesheet, no script, no network. It opens in a
browser on a machine that has never heard of spaCR, which is the
whole requirement.
"""
summary = self.summary()
rows = []
for hit in self.hits[:max(0, int(limit))]:
cells = [str(hit.rank), hit.gene, hit.name, _fmt(hit.effect),
_fmt(hit.p_value), _fmt(hit.q_value),
f"{hit.n_agree}/{hit.n_guides}", _fmt(hit.agreement),
", ".join(hit.flags)]
rows.append("<tr>" + "".join(
f"<td>{html.escape(cell)}</td>" for cell in cells) + "</tr>")
head = ("rank", "gene", "name", "effect", "p", "q", "guides agreeing",
"agreement", "flags")
return (
"<!doctype html><meta charset='utf-8'>"
f"<title>spaCR hit list — {html.escape(self.source or '')}</title>"
"<style>body{font-family:system-ui,sans-serif;margin:2rem}"
"table{border-collapse:collapse}td,th{border:1px solid #ccc;"
"padding:.25rem .5rem;font-size:14px}th{background:#eee}</style>"
f"<h1>Hit list</h1><p>{html.escape(self.source or '')}</p>"
f"<p>{summary['n_significant']} of {summary['n_genes_tested']} "
f"genes clear {self.alpha:g}; {summary['n_up']} up, "
f"{summary['n_down']} down.</p>"
"<table><tr>" + "".join(f"<th>{h}</th>" for h in head) + "</tr>"
+ "".join(rows) + "</table>")
[docs]
def write_html(self, path: Union[str, os.PathLike]) -> str:
"""Write :meth:`to_html` to a file and return the path.
:param path: destination HTML path.
"""
target = os.path.abspath(os.path.expanduser(os.fspath(path)))
os.makedirs(os.path.dirname(target) or ".", exist_ok=True)
with open(target, "w", encoding="utf-8") as handle:
handle.write(self.to_html())
return target
def _at_most(value: Any, limit: float) -> bool:
"""True when ``value`` is a real number no greater than ``limit``."""
try:
number = float(value)
except (TypeError, ValueError):
return False
return math.isfinite(number) and number <= float(limit)
def _at_least(value: Any, limit: float, *, absolute: bool = False) -> bool:
"""True when ``value`` is a real number no smaller than ``limit``.
``absolute`` compares the magnitude instead, for a criterion such as
``min_effect`` that is about how far a gene moved and not which way. The
absolute value is taken HERE, on the far side of the conversion, so a
field holding something that is not a number is excluded like every other
unreadable one rather than raising out of the filter.
"""
try:
number = float(value)
except (TypeError, ValueError):
return False
if absolute:
number = abs(number)
return math.isfinite(number) and number >= float(limit)
def _fmt(value: Any) -> str:
"""Format a number for a table cell; an em dash for a missing one."""
try:
number = float(value)
except (TypeError, ValueError):
return str(value)
if not math.isfinite(number):
return "—"
if number and (abs(number) < 1e-3 or abs(number) >= 1e5):
return f"{number:.3g}"
return f"{number:.4g}"
def _searchable(hit: Hit) -> str:
"""Everything about a row that a text query should match, folded."""
parts = [hit.gene, hit.name, hit.feature, hit.condition]
parts.extend(str(value) for value in hit.annotation.values())
return " ".join(parts).casefold()
[docs]
def build_hit_list(source: Union[str, os.PathLike, Mapping[str, pd.DataFrame]],
*,
metadata_files: Sequence[Union[str, os.PathLike]] = (),
metadata_key: str = "Gene ID",
toxoplasma: bool = False,
regression_type: str = "",
alpha: float = DEFAULT_ALPHA,
include_controls: bool = True,
) -> HitList:
"""Build the ranked, annotated hit list of one regression run.
:param source: a results folder, or a ``{role: DataFrame}`` mapping as
:func:`load_results` returns — the second form is what a caller with
the frames already in hand (or a test) uses.
:param metadata_files: annotation CSVs to join, each collapsed to one row
per gene first. See :func:`load_gene_metadata`.
:param metadata_key: the gene identifier column in those files.
:param toxoplasma: also join the bundled *Toxoplasma* annotation — gene
name, signal peptide and transmembrane, hyperLOPIT compartment, the
published CRISPR fitness scores, and tachyzoite / tissue-cyst /
EES1-5 expression. Applied AFTER ``metadata_files`` so a column the
user's own file supplies is never replaced by the bundle's.
:param regression_type: the backend, if known. Only affects how the list
is ranked: the penalised backends have no p-value, so they rank by
bootstrap selection frequency and carry no q-value.
:param alpha: the FDR (or the selection frequency) a hit is called at.
:param include_controls: keep the control rows in the list. They belong
there by default — a screen whose positive control is not near the
top has a problem, and that is visible only if it is listed.
:returns: a :class:`HitList`, ranked and with exactly one row per gene.
:raises FileNotFoundError: when ``source`` is a path that is not a folder.
:raises ValueError: when no gene-level coefficients could be found.
"""
notes: List[str] = []
if isinstance(source, Mapping):
frames = dict(source)
where = str(frames.pop("__source__", "")) or "(frames)"
else:
where = os.path.abspath(os.path.expanduser(os.fspath(source)))
frames = load_results(where)
gene_frame = _gene_level(frames)
if gene_frame is None or gene_frame.empty:
raise ValueError(
f"no gene-level coefficients in {where}: expected "
f"{RESULT_FILES['gene']} or a {RESULT_FILES['all']} carrying "
f"gene terms.")
n_terms = int(len(frames.get("all", gene_frame)))
table = gene_frame.copy()
table["gene"] = table["feature"].map(gene_of)
table = table.dropna(subset=["gene"])
duplicated = int(table["gene"].duplicated().sum())
if duplicated:
notes.append(
f"{duplicated} duplicate gene term(s) in the coefficient table; "
f"the first of each is kept.")
table = table.drop_duplicates(subset=["gene"], keep="first")
if not include_controls and "condition" in table.columns:
table = table[~table["condition"].isin(["nc", "pc", "control"])]
table = table.reset_index(drop=True)
ranking = ("selection-frequency"
if str(regression_type).strip().lower() in NO_P_VALUE_TYPES
else "q-value")
if ranking == "selection-frequency" and float(alpha) == DEFAULT_ALPHA:
alpha = DEFAULT_SELECTION_THRESHOLD
if ranking == "q-value":
table["q_value"] = benjamini_hochberg(
table.get("p_value", pd.Series([np.nan] * len(table))))
else:
table["q_value"] = np.nan
notes.append(
f"{regression_type} reports no frequentist p-value, so this list "
f"is ranked by bootstrap selection frequency and carries no "
f"q-value. Treat it as a selection method, not a hypothesis test.")
effects = {str(row["gene"]): float(row["coefficient"])
for _, row in table.iterrows()
if _finite(row.get("coefficient"))}
agreement = grna_agreement(effects, frames.get("grna"))
if frames.get("grna") is None or frames["grna"].empty:
notes.append(
"No per-gRNA coefficient table was found, so guide agreement "
"could not be computed for any gene.")
joined, join_notes = join_metadata(table, metadata_files,
key=metadata_key)
notes.extend(join_notes)
if toxoplasma:
from .annotation import annotate
before, had = len(joined), set(joined.columns)
joined = annotate(joined, key_column="gene", quiet=True)
if len(joined) != before:
raise ValueError(
f"the Toxoplasma annotation changed the row count from "
f"{before} to {len(joined)}.")
gained = [c for c in joined.columns if c not in had]
notes.append(
f"Bundled Toxoplasma annotation joined by gene number: "
f"{len(gained)} column(s).")
if len(joined) != len(table):
raise ValueError(
f"the metadata join changed the row count from {len(table)} to "
f"{len(joined)}; the annotation is not one row per gene.")
annotated_columns = [c for c in joined.columns if c not in table.columns]
hits = [
_hit(row, agreement, annotated_columns, ranking)
for _, row in joined.iterrows()
]
hits = _rank(hits, ranking)
result = HitList(
hits=tuple(hits), source=where, regression_type=regression_type,
ranking=ranking, alpha=float(alpha), n_terms=n_terms,
n_genes=len(hits), notes=tuple(notes))
_assert_one_row_per_gene(result)
return result
def _gene_level(frames: Mapping[str, pd.DataFrame]) -> Optional[pd.DataFrame]:
"""The gene-level coefficient table, however this run spelled it.
``results_gene.csv`` when it exists; otherwise the gene rows of
``results.csv``, which is what a run that predates the split wrote.
Which rows those are is :func:`coefficient_levels`' answer rather than a
second pattern match, so the hit list and the results panel cannot come
to disagree about what a gene row is.
"""
gene = frames.get("gene")
if gene is not None and not gene.empty and "feature" in gene.columns:
return gene
everything = frames.get("all")
if everything is None or everything.empty:
return None
if "feature" not in everything.columns:
return None
return everything[coefficient_levels(everything) == "gene"]
def _finite(value: Any) -> bool:
"""True when ``value`` converts to a finite float."""
try:
return math.isfinite(float(value))
except (TypeError, ValueError):
return False
def _hit(row: Mapping[str, Any],
agreement: Mapping[str, Tuple[int, int, List[str]]],
annotation_columns: Sequence[str], ranking: str) -> Hit:
"""Turn one joined coefficient row into a :class:`Hit`."""
gene = str(row["gene"])
effect = float(row["coefficient"]) if _finite(row.get("coefficient")) \
else float("nan")
std_err = float(row["std_err"]) if _finite(row.get("std_err")) \
else float("nan")
if math.isfinite(std_err) and math.isfinite(effect):
ci_low, ci_high = effect - 1.96 * std_err, effect + 1.96 * std_err
else:
ci_low = ci_high = float("nan")
n_agree, n_guides, agreeing = agreement.get(gene, (0, 0, []))
ratio = (n_agree / n_guides) if n_guides else float("nan")
flags: List[str] = []
condition = str(row.get("condition", "") or "")
if condition in ("nc", "pc", "control"):
flags.append(FLAG_CONTROL)
if n_guides == 0:
flags.append(FLAG_NO_GUIDES)
elif n_guides == 1:
flags.append(FLAG_SINGLE_GUIDE)
elif math.isfinite(ratio) and ratio < 0.5:
flags.append(FLAG_GUIDES_DISAGREE)
annotation = {column: row.get(column) for column in annotation_columns}
if annotation_columns and all(
value is None or (isinstance(value, float) and math.isnan(value))
for value in annotation.values()):
flags.append(FLAG_NO_METADATA)
return Hit(
gene=gene, feature=str(row.get("feature", "")), effect=effect,
std_err=std_err, ci_low=ci_low, ci_high=ci_high,
p_value=float(row["p_value"]) if _finite(row.get("p_value"))
else float("nan"),
q_value=float(row["q_value"]) if _finite(row.get("q_value"))
else float("nan"),
selection_frequency=float(row["selection_frequency"])
if _finite(row.get("selection_frequency")) else float("nan"),
n_guides=int(n_guides), n_agree=int(n_agree), agreement=ratio,
agreeing_guides=tuple(agreeing),
n_obs=int(row["n_gene"]) if _finite(row.get("n_gene")) else 0,
condition=condition,
direction=("up" if math.isfinite(effect) and effect > 0
else "down" if math.isfinite(effect) and effect < 0
else ""),
flags=tuple(flags), annotation=annotation)
def _rank(hits: Sequence[Hit], ranking: str) -> List[Hit]:
"""Order the rows and stamp a 1-based rank on each.
Significance first, magnitude second: a screen's question is "which genes
changed the phenotype", and two genes at the same q-value are separated
by how much they moved it. Untestable rows sort last rather than being
dropped — a coefficient with no p-value is still a number somebody may
need to see.
"""
if ranking == "selection-frequency":
def key(hit: Hit):
"""Rank finite selection and effect magnitude high, then gene."""
selection = (hit.selection_frequency
if math.isfinite(hit.selection_frequency) else -1.0)
magnitude = abs(hit.effect) if math.isfinite(hit.effect) else -1.0
return (-selection, -magnitude, hit.gene)
else:
def key(hit: Hit):
"""Rank finite q low, effect magnitude high, then gene."""
q = hit.q_value if math.isfinite(hit.q_value) else float("inf")
magnitude = abs(hit.effect) if math.isfinite(hit.effect) else -1.0
return (q, -magnitude, hit.gene)
ordered = sorted(hits, key=key)
return [Hit(**{**hit.__dict__, "rank": index + 1})
for index, hit in enumerate(ordered)]
def _assert_one_row_per_gene(hit_list: HitList) -> None:
"""Refuse to hand back a list that names a gene twice.
The invariant this module exists to keep. It is checked on the OUTPUT
rather than trusted from the inputs because there are three places a
duplicate can enter — a coefficient table with two terms for one gene, a
metadata file with one row per transcript, and a second metadata file
that repeats the first — and a hit list that counts one gene as two
findings is worse than no hit list.
"""
seen = set()
for hit in hit_list.hits:
if hit.gene in seen:
raise ValueError(
f"gene {hit.gene!r} appears more than once in the hit list; "
f"a metadata file with one row per transcript is the usual "
f"cause. Collapse it to one row per gene first.")
seen.add(hit.gene)