"""Join bundled *Toxoplasma* gene annotations to exported results.
The joined fields include gene names, predicted signal peptides and
transmembrane domains, published phenotype scores, stage-specific expression
and hyperLOPIT localisation.
All supported identifier forms are normalized to a bare gene number:
``TGGT1_224750``, ``TGME49_224750``, ``gene_fraction:gene[224750]`` and the
guide identifier ``224750_2`` resolve to gene ``224750`` through
:func:`gene_number`. Each source is reduced to one row per gene and merged
with ``many_to_one`` validation, preventing annotation duplicates from
multiplying result rows. A field unavailable in a source is omitted rather
than emitted as an all-missing column.
Bundled annotation sources:
toxoplasma_metadata.csv 2.97 MB gene name, product, expression
deeptmhmm.csv 1.11 MB signal peptide and transmembrane
phenotype.csv 0.48 MB the published CRISPR fitness screens
lopit.csv 0.20 MB hyperLOPIT/TAGM compartment
uniprot.csv 0.13 MB accession
``deeptmhmm.csv`` contains a DeepTMHMM analysis of 8,140 proteins. Summary
fields (``dtm_type``, ``n_tm`` and ``sp_length``) are joined by
:func:`annotate`; coordinates for as many as 24 transmembrane segments are
written separately by :func:`supplementary`. Per-segment sequence columns are
excluded because they are recoverable from the reference proteome.
"""
from __future__ import annotations
import os
import re
from functools import lru_cache
from typing import Dict, List, Optional, Sequence, Tuple
#: The first run of four or more digits in a name. Four, not six: ToxoDB gene
#: numbers are six digits today, but the rule has to reject the ``_2`` of a
#: guide id and the ``1`` of the ``TGGT1`` prefix, and a floor of four does
#: both without asserting a length the database never promised.
_DIGITS = re.compile(r"(\d{4,})")
#: Where the bundled tables live.
_DATA = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"resources", "data")
[docs]
def gene_number(value) -> Optional[str]:
"""The bare gene number named by ``value``, or ``None``.
:param value: feature, accession, guide, or design-term value to parse.
Accepts every spelling this project has: a design term
(``gene_fraction:gene[224750]``, ``fraction:grna[224750_2]``), a gene id
in either strain (``TGGT1_224750``, ``TGME49_224750``), a split gene
model (``TGME49_201180A``), a guide id (``224750_2``) or the bare number.
:returns: the number as a string, or ``None`` for anything that names no
gene -- ``Intercept``, ``rowID[T.r03]``, an empty cell, NaN.
A split gene model resolves to its parent number. ``TGME49_201180A`` and
``TGME49_201180B`` are both ``201180``, because the screen's library
targets the locus and cannot distinguish the two models. For this reason
every source here is deduplicated before it is joined.
"""
if value is None:
return None
text = str(value)
if text in ("", "nan", "NaN", "None", "<NA>"):
return None
match = _DIGITS.search(text)
return match.group(1) if match else None
def _read(filename: str):
"""The bundled CSV, or ``None`` with the reason printed.
A missing or unreadable table is not fatal. The export is still an
export; it simply carries fewer columns, and the console says which and
why rather than leaving the reader to notice.
"""
import pandas as pd
path = os.path.join(_DATA, filename)
if not os.path.isfile(path):
print(f"Toxoplasma annotation: {filename} is not bundled with this "
f"install, so its columns are left out of the export.")
return None
try:
return pd.read_csv(path, low_memory=False)
except Exception as error: # noqa: BLE001
print(f"Toxoplasma annotation: {filename} could not be read "
f"({error}), so its columns are left out of the export.")
return None
def _keyed(frame, source_column: str, wanted: Sequence[Tuple[str, str]]):
"""``frame`` reduced to the key and ``wanted``, one row per gene.
:param source_column: the column holding a gene id or number.
:param wanted: ``(source name, exported name)`` pairs. A pair whose
source column is absent is dropped, not filled with NaN.
"""
if frame is None:
return None
pairs = [(src, out) for src, out in wanted if src in frame.columns]
if not pairs or source_column not in frame.columns:
return None
out = frame[[source_column] + [src for src, _ in pairs]].copy()
out["_key"] = out[source_column].map(gene_number)
out = out.loc[out["_key"].notna()]
out = out.drop(columns=[source_column])
out = out.rename(columns=dict(pairs))
out = out.rename(columns={"_key": "gene_nr"})
return out.drop_duplicates(subset="gene_nr", keep="first")
@lru_cache(maxsize=1)
def _metadata():
"""Return bundled gene metadata keyed by ``gene_nr``, or ``None``."""
return _keyed(_read("toxoplasma_metadata.csv"), "Gene ID", (
("Gene Name or Symbol", "gene_name"),
("Product Description", "product_description"),
("sense - Tachyzoites", "expr_tachyzoite"),
("sense - Tissue cysts", "expr_bradyzoite"),
("sense - EES1", "expr_ees1"),
("sense - EES2", "expr_ees2"),
("sense - EES3", "expr_ees3"),
("sense - EES4", "expr_ees4"),
("sense - EES5", "expr_ees5"),
))
@lru_cache(maxsize=1)
def _topology():
"""Signal peptide and transmembrane, from this project's DeepTMHMM run.
``dtm_type`` is one field carrying both answers -- GLOB, SP, TM, SP+TM,
BETA -- so it is split into two booleans a reader can filter on, and the
raw label is kept beside them for anyone who wants BETA.
"""
frame = _keyed(_read("deeptmhmm.csv"), "identifier", (
("dtm_type", "topology"),
("n_tm", "n_transmembrane"),
("sp_length", "signal_peptide_length"),
))
if frame is None:
return None
label = frame["topology"].astype("string").str.upper()
frame["signal_peptide"] = label.isin(["SP", "SP+TM"])
frame["transmembrane"] = label.isin(["TM", "SP+TM"])
return frame[["gene_nr", "signal_peptide", "transmembrane",
"n_transmembrane", "signal_peptide_length", "topology"]]
@lru_cache(maxsize=1)
def _phenotype():
"""Return bundled ``fit_*`` phenotype scores keyed by ``gene_nr``."""
frame = _read("phenotype.csv")
if frame is None:
return None
columns = [c for c in frame.columns if c.startswith("fit_")]
return _keyed(frame, "gene_nr", tuple((c, c) for c in columns))
@lru_cache(maxsize=1)
def _lopit():
"""Return bundled TAGM localisation as ``hyperlopit`` by ``gene_nr``."""
return _keyed(_read("lopit.csv"), "gene_nr",
(("tagm_location", "hyperlopit"),))
@lru_cache(maxsize=1)
def _uniprot():
"""Return bundled UniProt accessions keyed by ``gene_nr``, or ``None``."""
return _keyed(_read("uniprot.csv"), "gene_nr", (("uniprot", "uniprot"),))
#: The sources, in the order their columns appear in an export.
SOURCES = (
("gene name and expression", _metadata),
("signal peptide and transmembrane", _topology),
("hyperLOPIT compartment", _lopit),
("CRISPR phenotype", _phenotype),
("UniProt accession", _uniprot),
)
[docs]
def columns() -> List[str]:
"""Every column :func:`annotate` can add, in order.
Empty when nothing is bundled. Used to state up front what an export will
carry, and by the tests, so a source added here cannot be forgotten.
"""
names: List[str] = []
for _label, source in SOURCES:
frame = source()
if frame is None:
continue
names.extend(c for c in frame.columns if c != "gene_nr")
return names
#: The name the annotation key is merged under. Private, and deliberately not
#: "gene_nr": the caller's table may carry that name itself, and a merge whose
#: right key collides with a left column is suffixed rather than shared.
_JOIN_KEY = "_spacr_annotation_gene_nr"
def _key_column(frame) -> Optional[str]:
"""The column of ``frame`` that names a gene, or ``None``.
Preference order, and it matters: an explicit ``gene`` column beats the
design term, because a table that carries both has already parsed the
term once and this must not disagree with it.
"""
for name in ("gene_nr", "gene", "gene_id", "Gene ID", "grna", "feature",
"variable", "index"):
if name in getattr(frame, "columns", ()):
keys = frame[name].map(gene_number)
if keys.notna().any():
return name
return None
[docs]
def annotate(frame, *, key_column: Optional[str] = None, quiet: bool = False):
"""``frame`` with the bundled *Toxoplasma* annotation joined on.
:param frame: any exported table -- coefficients, significant hits, the
montage sidecar.
:param key_column: the column naming a gene. Found automatically when not
given, and a table naming no gene is returned unchanged rather than
gaining a block of empty columns.
:param quiet: suppress the console line saying what was joined.
:returns: a copy of ``frame`` with the annotation columns added;
``frame`` itself is not modified. When no gene column is found,
``frame`` is returned as is.
Columns already present in ``frame`` are not overwritten -- a run that
computed its own ``gene_name`` keeps it, and the annotation's version is
left out rather than silently replacing it.
"""
import pandas as pd
if frame is None or not len(getattr(frame, "columns", ())):
return frame
key = key_column or _key_column(frame)
if key is None or key not in frame.columns:
if not quiet:
print("Toxoplasma annotation: this table names no gene column, "
"so nothing was joined onto it.")
return frame
out = frame.copy()
out["_gene_nr"] = out[key].map(gene_number)
added: List[str] = []
for label, source in SOURCES:
right = source()
if right is None:
continue
new = [c for c in right.columns
if c != "gene_nr" and c not in out.columns]
if not new:
if not quiet:
print(f"Toxoplasma annotation: {label} is already on this "
f"table, so it was not joined again.")
continue
keyed = right[["gene_nr"] + new].rename(
columns={"gene_nr": _JOIN_KEY})
out = out.merge(keyed, how="left",
left_on="_gene_nr", right_on=_JOIN_KEY,
validate="many_to_one")
out = out.drop(columns=[_JOIN_KEY])
added.extend(new)
matched = int(out["_gene_nr"].notna().sum())
out = out.drop(columns=["_gene_nr"])
if not quiet:
if added:
hit = int(out[added].notna().any(axis=1).sum()) if len(out) else 0
print(f"Toxoplasma annotation: {len(added)} column(s) joined onto "
f"{matched} of {len(frame)} row(s) by gene number; "
f"{hit} matched the annotation.")
else:
print("Toxoplasma annotation: no bundled table could be read, so "
"the export carries no annotation.")
return out
#: Columns of the supplementary topology table that are not per-segment.
_TOPOLOGY_SUMMARY = ("gene_nr", "identifier", "accession", "length",
"dtm_type", "n_signal", "n_tm", "sp_start", "sp_end",
"sp_length")
[docs]
def supplementary(genes=None, path=None):
"""The full DeepTMHMM table, as its own supplementary data table.
This table contains the complete bundled DeepTMHMM results for all
*Toxoplasma* proteins.
:param genes: restrict to these genes -- any spelling
:func:`gene_number` accepts. ``None`` writes all 8,140 proteins.
:param path: write here as CSV as well as returning the frame.
:returns: the table, or ``None`` when DeepTMHMM is not bundled.
This table is kept separate from :func:`annotate` because the
per-segment coordinates add 72 columns. :func:`annotate` joins only the
summary fields (signal peptide and transmembrane-helix count) to
coefficient exports; segment positions are written to this table.
Segment columns that are empty for every selected protein are dropped,
so a set of soluble proteins yields a table ending at `n_tm`; the rest
of this module follows the same rule.
"""
frame = _read("deeptmhmm.csv")
if frame is None:
return None
if genes is not None:
wanted = {gene_number(one) for one in genes} - {None}
keys = frame["gene_nr"].astype("string").map(gene_number)
frame = frame.loc[keys.isin(wanted)]
live = [c for c in frame.columns
if c in _TOPOLOGY_SUMMARY or frame[c].notna().any()]
frame = frame[live].reset_index(drop=True)
if path is not None:
target = os.path.abspath(os.path.expanduser(os.fspath(path)))
os.makedirs(os.path.dirname(target) or ".", exist_ok=True)
frame.to_csv(target, index=False)
print(f"Toxoplasma annotation: DeepTMHMM topology for "
f"{len(frame)} protein(s) written to {target}")
return frame
[docs]
def clear_cache() -> None:
"""Clear the cached bundled tables so the next access re-reads them."""
for _label, source in SOURCES:
source.cache_clear()
__all__ = ["SOURCES", "annotate", "clear_cache", "columns", "gene_number",
"supplementary"]
#: Columns UniProt is asked for, and what the joined table calls them.
#:
#: RENAMED, because the bundled path's columns are what every downstream
#: panel and export already reads. A UniProt annotation that called its gene
#: name "Gene Names" would be a second vocabulary for the same facts.
UNIPROT_COLUMNS = {
"Entry": "uniprot_accession",
"Entry Name": "uniprot_entry",
"Protein names": "product",
"Gene Names": "gene_name",
"Organism": "organism",
"Length": "protein_length",
"Subcellular location [CC]": "localisation",
"Function [CC]": "function",
"Signal peptide": "signal_peptide",
"Transmembrane": "transmembrane",
"Gene Ontology (biological process)": "go_process",
"Gene Ontology (cellular component)": "go_component",
"Gene Ontology (molecular function)": "go_function",
"AlphaFoldDB": "alphafold",
}
def _uniprot_keys(values):
"""Every spelling of a gene in a UniProt "Gene Names" cell, lower case.
UniProt puts the synonyms in one space-separated cell -- "TP53 P53" --
so a screen naming either one has to match. Returned as a list per row
so the frame can be exploded onto them.
"""
out = []
for value in values:
names = str(value or "").replace(",", " ").split()
out.append([n.strip().lower() for n in names if n.strip()])
return out
def _uniprot_key_column(frame) -> Optional[str]:
"""The column naming a gene, for a table that is NOT Toxoplasma's.
`_key_column` parses every candidate with :func:`gene_number`, which is
the bundled path's key space: a bare Toxoplasma gene NUMBER. Human,
mouse and Plasmodium screens name their targets `TP53`, `Cdkn1a`,
`PF3D7_1206100` -- none of which is a number -- so that finder rejected
every column and the UniProt join reported "this table names no gene
column" about tables that named nothing else.
Same preference order, different test: a column counts when it holds
text at all.
"""
for name in ("gene", "gene_name", "gene_id", "Gene ID", "gene_nr",
"grna", "feature", "variable", "index"):
if name not in getattr(frame, "columns", ()):
continue
values = frame[name].astype(str).str.strip()
if values.replace("", None).notna().any() and (values != "").any():
return name
return None
[docs]
def annotate_from_uniprot(frame, source, *, cache_dir=None,
key_column: Optional[str] = None,
quiet: bool = False):
"""``frame`` with UniProt's annotation for ``source`` joined onto it.
The UniProt branch of :func:`annotate_with`. Joins on the gene name --
UniProt's own, including its synonyms -- and on the accession, because a
screen library may name its targets either way.
Never raises and never multiplies rows: the annotation is collapsed to
one row per key before the merge, and the merge is ``many_to_one``, for
the reason this module's header gives.
:returns: ``(frame, note)``. The frame is unchanged when nothing could
be joined, and the note says why.
"""
import pandas as pd
from .uniprot import annotation_for
if frame is None or not len(getattr(frame, "columns", ())):
return frame, ""
asked = _uniprot_key_column(frame) if key_column is None else key_column
genes = (frame[asked].astype(str).tolist()
if asked and asked in frame.columns else None)
table, note = annotation_for(source, cache_dir=cache_dir, genes=genes)
if table is None or not len(table):
return frame, note
key = key_column or _uniprot_key_column(frame)
if key is None or key not in frame.columns:
return frame, ("this table names no gene column, so the UniProt "
"annotation was not joined onto it.")
table = table.rename(columns=UNIPROT_COLUMNS)
keep = [c for c in UNIPROT_COLUMNS.values() if c in table.columns]
if not keep:
return frame, "UniProt returned no usable columns."
table = table[keep].copy()
table["_key"] = _uniprot_keys(table.get("gene_name", ""))
exploded = table.explode("_key").dropna(subset=["_key"])
if "uniprot_accession" in table.columns:
by_accession = table.copy()
by_accession["_key"] = (by_accession["uniprot_accession"]
.astype(str).str.strip().str.lower())
exploded = pd.concat([exploded, by_accession], ignore_index=True)
exploded = exploded.drop_duplicates(subset=["_key"], keep="first")
joined = frame.copy()
joined["_key"] = joined[key].astype(str).str.strip().str.lower()
new = [c for c in keep if c not in frame.columns]
if not new:
return frame, ""
merged = joined.merge(exploded[["_key", *new]], on="_key", how="left",
validate="many_to_one")
merged = merged.drop(columns=["_key"])
matched = int(merged[new[0]].notna().sum()) if new else 0
if not quiet:
print(f"UniProt annotation ({source}): {matched} of {len(merged)} "
f"rows matched, {len(new)} column(s) joined.")
if matched == 0:
note = (note + " " if note else "") + (
f"none of the {len(frame)} gene identifiers matched a UniProt "
f"entry for {source}.")
return merged, note
[docs]
def annotate_with(frame, source, *, cache_dir=None,
key_column: Optional[str] = None, quiet: bool = False):
"""Annotate ``frame`` from whatever ``source`` names.
This is the function the pipeline calls. ``source`` is the
``annotation_source`` setting:
* empty or any spelling of Toxoplasma -- the bundled tables, offline.
This is the default and it does not touch
:mod:`spacr.uniprot` at all;
* an organism name or taxon id -- that organism from UniProt;
* an accession -- that entry.
:returns: ``(frame, note)``; the note is empty on a clean join.
"""
from .uniprot import resolve
if resolve(source).kind == "bundled":
return annotate(frame, key_column=key_column, quiet=quiet), ""
return annotate_from_uniprot(frame, source, cache_dir=cache_dir,
key_column=key_column, quiet=quiet)