"""Click a gene, see everything spaCR knows about it -- in one panel.
Two modules already hold the two halves of the answer and
neither of them is a widget, which is deliberate: the parts that can be WRONG
are testable without a window.
:mod:`spacr.gene_tile` WHICH gene this dot is -- guide to gene, the
ambiguous protospacers, and THIS SCREEN's own
effect / p / q / guides, read out of the results
frame that is already on screen.
:mod:`spacr.gene_facts` WHAT that gene is -- product, topology with the
DeepTMHMM coordinates, hyperLOPIT compartment,
the published CRISPR fitness screens and the
stage expression, all of it out of
:mod:`spacr.annotation`.
This module puts them one above the other and does nothing else with the
numbers. THE PANEL IS NOT A SECOND SOURCE OF TRUTH: the coefficient, the
p-value, the q-value and the guide agreement are whatever the table on screen
says they are, because two places computing one number is how they start
disagreeing.
THE GUI THREAD DOES NOT READ FILES
----------------------------------
Cold, the first click costs 360 ms of CSV reading -- five bundled annotation
tables, DeepTMHMM's 8,140 rows, and the gRNA reference and metadata indices
:mod:`spacr.gene_tile` keeps -- inside a mouse press. A plot that freezes for
a third of a second when clicked reads as broken.
So :func:`warm_annotation` does all of it on a worker thread, through
:class:`spacr.qt.job_runner.JobRunner`, which is the module that already
gets the threading rules right: the worker's ``finished`` is relayed through
a Signal whose receiver is a BOUND METHOD of a GUI-thread object, never a
closure, so the handler runs on the GUI thread and the QThread is retired.
Given the screen's own terms it warms every gene in one join -- 400 genes
cost the same 21 ms as one -- after which a click is a dictionary lookup.
Until it is warm, the panel says so, and the one control it has is greyed out
with the reason on it. A gene with no annotation likewise
says "no row in the bundled annotation" rather than showing a form of empty
fields, which reads as "measured, found nothing".
"""
from __future__ import annotations
import logging
from typing import Any, Callable, List, Optional, Sequence, Tuple
from PySide6.QtCore import QSize, Qt, Signal
from PySide6.QtGui import QPainter, QPixmap
from PySide6.QtWidgets import (QApplication, QFileDialog, QHBoxLayout,
QLabel, QPushButton, QSizePolicy,
QTextBrowser, QVBoxLayout, QWidget)
from ..theme import SPACING, font_px
from .gene_tile import TILE_WIDTH, GeneTilePanel
LOG = logging.getLogger("spacr.qt.gene_panel")
__all__ = ["GenePanel", "warm_annotation"]
#: Shown in the lower half before the annotation has finished loading. Not a
#: blank: a blank pane beside a plot reads as broken rather than as busy.
LOADING_TEXT = "Loading the bundled Toxoplasma annotation…"
#: Shown in the lower half before anything has been clicked.
IDLE_TEXT = ("What spaCR knows about the gene appears here — product,"
" topology, compartment, fitness screens and stage expression.")
#: The term used to build :mod:`spacr.gene_tile`'s three indices during the
#: warm-up when the caller passed no terms of its own. Any term does; this one
#: is a control guide, so it resolves without depending on a gene being in the
#: bundled reference.
_WARMING_TERM = "fraction:grna[000000_1]"
[docs]
def warm_annotation(features: Sequence[Any] = ()) -> Tuple[str, ...]:
"""Read every table a gene tile needs. RUNS ON A WORKER THREAD.
:param features: the terms the user might click -- pass the results
table's whole ``feature`` column. Every gene among them is joined in
one pass, which costs the same as joining one.
:returns: the annotation columns that came out available; empty when the
bundled tables are not installed, which is a state the panel shows
rather than hides.
Touches no widget and returns only data -- that is the contract for
anything handed to :meth:`spacr.qt.job_runner.JobRunner.submit`.
It warms :mod:`spacr.gene_tile` too, by resolving one term. That module
keeps its own indices over the gRNA reference and the curated metadata,
and they are just as cold on the first click as the annotation tables
are; warming one and not the other would move the freeze rather than
remove it.
"""
from ... import gene_facts, gene_tile
columns = gene_facts.warm(features)
try:
gene_tile.gene_tile(features[0] if len(features) else _WARMING_TERM)
except Exception: # noqa: BLE001
LOG.debug("gene panel: could not warm the gene_tile indices",
exc_info=True)
return columns
[docs]
class GenePanel(QWidget):
"""The gene tile: which gene this is, and everything known about it.
:param frame_provider: called with no arguments for the current results
frame. A callable rather than a stored frame so a newly loaded
regression is never answered out of the previous one.
:param threaded: ``False`` warms inline instead of on a worker thread, so
a test can drive the panel synchronously without the behaviour
diverging -- :class:`~spacr.qt.job_runner.JobRunner` emits the same
signals in the same order either way.
:param parent: the usual.
"""
#: Emitted with the feature string whenever a tile is built for it.
tile_shown = Signal(str)
#: Emitted with the available column count once the annotation is loaded;
#: ``0`` means the bundled tables are not installed.
annotation_ready = Signal(int)
def __init__(self, frame_provider: Optional[Callable[[], Any]] = None,
*, threaded: bool = True, parent=None):
"""Build the gene panel: the record above what spaCR knows about it.
The annotation lookup is warmed on a worker, and only once the panel has
actually been shown: a panel re-set with the same table on every filter
move would otherwise start a thread per redraw for no new genes at all.
The thread's lifetime is guarded twice, because Qt aborts the process if
a running ``QThread`` is destroyed and a panel can be dropped without
ever being closed -- a tab rebuilt, a screen replaced, an interpreter
shutting down.
:param frame_provider: called for the coefficient table the record half
reads its numbers from.
:param threaded: warm the annotation lookup on a worker thread.
:param parent: parent widget, or ``None``.
"""
super().__init__(parent)
from ..job_runner import JobRunner
self._columns: Tuple[str, ...] = ()
self._warm = False
self._facts: Tuple[Any, ...] = ()
#: The terms the last warm-up covered. A panel is re-set_frame'd with
#: the SAME table whenever the gene/guide filter moves, and a QThread
#: per redraw is a thread for no new genes at all.
self._warmed: Tuple[Any, ...] = ()
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(SPACING.get("xs", 4))
from .collapsible_splitter import CollapsibleSplitter
split = CollapsibleSplitter(Qt.Vertical, self,
persist_key="regression::gene")
#: The record half: identity, this screen's numbers, the guides.
self.summary = GeneTilePanel(frame_provider=frame_provider)
split.add_section(self.summary, "Gene record", stretch=3,
persist_key="regression/Gene record")
self._known = QTextBrowser()
self._known.setOpenLinks(False)
self._known.setOpenExternalLinks(False)
self._known.setProperty("i18nSkipText", True)
self._known.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
split.add_section(self._known, "Known about this gene", stretch=4,
persist_key="regression/Known about this gene")
self.split = split
layout.addWidget(split, 1)
footer = QHBoxLayout()
footer.setContentsMargins(0, 0, 0, 0)
self._status = QLabel("")
self._status.setWordWrap(True)
self._status.setStyleSheet(
f"color: palette(mid); font-size: {font_px(10)}px;")
footer.addWidget(self._status, 1)
#: Writes this gene's full DeepTMHMM record -- every segment's
#: coordinates -- as its own CSV. It is the one thing on this panel
#: that leaves a file behind, so it is the one thing that has to say
#: why it cannot: see :meth:`topology_reason`.
self.topology_button = QPushButton("Save topology CSV…")
self.topology_button.clicked.connect(self._ask_to_save_topology)
footer.addWidget(self.topology_button)
layout.addLayout(footer)
self.clear()
self._runner = JobRunner(self, threaded=bool(threaded),
app_key="gene annotation",
user_visible=False)
#: Whether the warm-up has been started. See :meth:`showEvent`.
self._warming_started = False
#: True once the panel has actually been shown. Nothing starts a
#: thread before this: see :meth:`warm_for`.
self._shown = False
#: The features a pending warm-up should cover, or () for all.
self._pending_warm = ()
application = QApplication.instance()
if application is not None:
application.aboutToQuit.connect(self._shut_down_warming)
[docs]
def showEvent(self, event): # noqa: N802
"""Start annotation warm-up when the panel first becomes visible.
Deferring the worker until the panel is shown avoids creating a
background thread for panels that are constructed but never used.
:param event: the show event; passed on to the base class before the
warm-up starts.
"""
super().showEvent(event)
self._shown = True
self.warm_now()
[docs]
def warm_now(self) -> bool:
"""Start the pending annotation warm-up.
Returns
-------
bool
Whether a worker was started.
"""
if self._warming_started:
return False
self._warming_started = True
features = list(self._pending_warm) if self._pending_warm else None
job = ((lambda f=features: warm_annotation(f)) if features
else warm_annotation)
return bool(self._runner.submit(job, self._annotation_loaded))
@property
[docs]
def tile(self):
"""The :class:`spacr.gene_tile.GeneTile` on screen, or ``None``."""
return self.summary.tile
@property
[docs]
def feature(self) -> str:
"""The feature string the current tile was built from."""
return self.summary.feature
@property
[docs]
def facts(self) -> Tuple[Any, ...]:
"""One :class:`spacr.gene_facts.GeneFacts` per candidate gene."""
return self._facts
[docs]
def is_warm(self) -> bool:
"""Has the annotation finished loading off the GUI thread?"""
return self._warm
[docs]
def annotation_columns(self) -> Tuple[str, ...]:
"""The annotation columns this install can show. Empty until warm."""
return self._columns
[docs]
def set_frame_provider(self, provider: Optional[Callable[[], Any]]) -> None:
"""Point the panel at where the current results frame lives.
:param provider: a no-argument callable returning the current results
frame, called each time a feature is shown; ``None`` builds tiles
without a frame.
"""
self.summary.set_frame_provider(provider)
[docs]
def clear(self) -> None:
"""Back to the waiting state, which says what it is waiting for."""
self.summary.clear()
self._facts = ()
self._say(LOADING_TEXT if not self._warm else IDLE_TEXT)
self._update_topology_button()
[docs]
def warm_for(self, frame) -> bool:
"""Warm the annotation for every gene in ``frame``. Call on load.
:param frame: the results table that was just loaded.
:returns: whether a warm-up was started.
The terms are read off the frame HERE, on the GUI thread, because
that is a list comprehension over a column that is already in memory.
The join they feed is what goes to the worker.
"""
terms = tuple(_terms_of(frame))
if not terms or terms == self._warmed:
return False
self._warmed = terms
self._pending_warm = terms
self._warming_started = False
if not self._shown:
return False
return self.warm_now()
def _annotation_loaded(self, columns) -> None:
"""The warm-up landed. GUI THREAD ONLY -- JobRunner guarantees it."""
self._columns = tuple(columns or ())
self._warm = True
if self.summary.feature:
self.show_feature(self.summary.feature)
else:
self._say(IDLE_TEXT)
self._update_topology_button()
self.annotation_ready.emit(len(self._columns))
[docs]
def show_feature(self, key) -> None:
"""Build and show the tile for one clicked feature.
THE SLOT ``key_selected`` CONNECTS TO. It takes the feature string and
nothing else, so a volcano click and a results-row click reach it
identically -- and it is connected once, on the table, because that is
the funnel both directions already pass through.
Driven straight rather than off ``summary.tile_shown``: that signal
is not emitted when the resolver RAISES, and the one case where the
lower half must not be left showing the previous gene is exactly the
one where the upper half failed.
:param key: the clicked feature string, e.g. a gene or results-row key;
it is passed to the summary tile as given.
"""
self.summary.show_feature(key)
self._render_known(self.summary.tile)
self._update_topology_button()
self.tile_shown.emit(self.summary.feature or str(key))
def _render_known(self, tile) -> None:
"""Render what spaCR knows about the resolved gene, or why it knows nothing.
Every unavailable case says which one it is -- the gene did not resolve,
the lookup is still warming, the term carries no gene number, the fact
source is not installed -- because they are different problems and only
one of them is worth acting on.
An ambiguous term gets a heading per candidate: three products under one
heading would read as one protein with three names.
:param tile: the resolved gene tile, or ``None`` when nothing resolved.
"""
import html as _html
from ... import gene_facts
self._facts = ()
if tile is None:
self._say("The gene could not be resolved, so there is nothing "
"to look up. The plot is unaffected.")
return
if not self._warm:
self._say(LOADING_TEXT)
return
reason = _nothing_to_look_up(tile)
if reason:
self._say(reason)
return
unavailable = gene_facts.unavailable_reason()
if unavailable:
self._say(unavailable)
return
found = gene_facts.facts_for(c.gene for c in tile.candidates)
self._facts = tuple(found[c.gene] for c in tile.candidates
if c.gene in found)
if not self._facts:
self._say("No gene number could be parsed out of this term, so "
"there is no annotation to look up.")
return
parts: List[str] = []
for candidate, known in zip(tile.candidates, self._facts):
if len(self._facts) > 1:
parts.append("<h3 style='margin-bottom:0'>what spaCR knows "
f"about {_html.escape(candidate.name)}</h3>")
parts.append(known.to_html())
self._known.setHtml("".join(parts))
self._status.setText("")
def _say(self, text: str) -> None:
"""Put a sentence in the lower half instead of a table of facts."""
import html as _html
self._known.setHtml(
f"<p style='color:#888'>{_html.escape(text)}</p>")
self._status.setText("")
[docs]
def topology_reason(self) -> str:
"""Why "Save topology CSV" cannot run, or ``""`` when it can.
The design: a control that cannot do anything is greyed out AND
says why. This is the sentence, and it is the button's tooltip.
"""
if not self._warm:
return ("The DeepTMHMM table is still loading. The button turns "
"on when it is ready.")
if not self._columns:
return ("The bundled Toxoplasma annotation is not installed with "
"this copy of spaCR, so there is no topology to save.")
if not self._facts:
return "Click a gene first — there is no gene to save topology for."
genes = [known for known in self._facts if known.segments]
if not genes:
named = ", ".join(known.gene for known in self._facts)
return (f"DeepTMHMM found no signal peptide and no transmembrane "
f"segment in {named}, so its topology table would be "
f"empty.")
return ""
def _update_topology_button(self) -> None:
"""Enable the topology export, or say why it is unavailable.
It is the one control here that leaves a file behind, so it is the one
that has to explain a refusal rather than simply going grey.
"""
reason = self.topology_reason()
self.topology_button.setEnabled(not reason)
self.topology_button.setToolTip(
reason or "Write this gene's full DeepTMHMM record — every "
"segment's coordinates — as a CSV.")
[docs]
def save_topology(self, path) -> bool:
"""Write the clicked gene's full DeepTMHMM record to ``path``.
:param path: the CSV file to write, passed to
:func:`spacr.annotation.supplementary`. Nothing is written when no
clicked gene is known or DeepTMHMM is not bundled.
:returns: whether a file was written.
Straight through :func:`spacr.annotation.supplementary`, which is the
function that defines what that table is. Rewriting the columns here
would be a second definition of the supplementary file, differing
from the one an export writes in ways nobody would notice until a
reviewer compared them.
"""
from ... import annotation
genes = [known.gene for known in self._facts if known.gene]
if not genes:
return False
return annotation.supplementary(genes, path) is not None
def _ask_to_save_topology(self) -> None:
"""Ask where to write the DeepTMHMM topology CSV and write it.
A failure is reported on the status line rather than raised: losing the
panel because an export could not be written would be worse than not
having the file.
"""
genes = "_".join(known.gene for known in self._facts if known.gene)
path, _filter = QFileDialog.getSaveFileName(
self, "Save DeepTMHMM topology",
f"deeptmhmm_{genes or 'gene'}.csv", "CSV (*.csv)")
if not path:
return
try:
self.save_topology(path)
except Exception as error: # noqa: BLE001
LOG.exception("gene panel: could not write the topology table")
self._status.setText(f"Could not write {path}: {error}")
else:
self._status.setText(f"Topology written to {path}")
[docs]
def to_pixmap(self, width: int = TILE_WIDTH) -> QPixmap:
"""The whole tile -- both halves -- as one ``QPixmap``.
The figure grid's cells take a pixmap and size themselves from its
aspect ratio, so rendering to one lets the gene tile be a tile in
that grid without ``_FigureCell`` learning about text.
"""
top = self.summary.to_pixmap(width)
bottom = _document_pixmap(self._known.toHtml(), width)
height = max(top.height() + bottom.height(), 1)
out = QPixmap(QSize(max(int(width), 1), height))
out.fill(Qt.transparent)
painter = QPainter(out)
try:
painter.drawPixmap(0, 0, top)
painter.drawPixmap(0, top.height(), bottom)
finally:
painter.end()
return out
def _shut_down_warming(self) -> None:
"""Stop the warm-up. A BOUND METHOD, and that is the point.
It is connected to ``QApplication.aboutToQuit``; a closure there
would make the application object the receiver and the call would be
dropped when this panel is destroyed first, leaving the very thread
it was meant to stop still running.
"""
try:
self._runner.shutdown()
except RuntimeError:
pass
[docs]
def closeEvent(self, event): # noqa: N802
"""Stop the warm-up before the widget goes.
Qt aborts the process if a running QThread is destroyed, and a
warm-up outliving its panel is exactly that.
:param event: the close event; passed on to the base class after the
warm-up worker is stopped.
"""
self._shut_down_warming()
super().closeEvent(event)
[docs]
def __del__(self):
"""The last guard, and the only one that needs nothing to happen.
THREE THINGS CAN START THE WARM-UP -- the first show, `warm_now`, and
a frame arriving -- so guarding each start site is guarding the wrong
end. This guards the LIFETIME: whenever the panel is collected, with
or without a close, with or without an event loop, the thread is
asked to stop first.
The other two guards each need an event to fire. `closeEvent` needs
somebody to close the panel; `QApplication.aboutToQuit` needs an
event loop to quit. A panel built, handed a table and dropped -- a
tab rebuilt, a screen replaced, a headless script, a test -- reaches
neither, and Qt calls `abort()` on the running thread. That was
SIGABRT on `QApplication([]); RegressionResultsPanel()`.
Everything here is swallowed on purpose. `__del__` runs during
garbage collection and at interpreter shutdown, where the C++ half
may already be gone and the module globals may already be None; an
exception raised here is printed and ignored by Python anyway, and
the one thing worth doing is the shutdown attempt.
"""
try:
self._shut_down_warming()
except BaseException: # noqa: BLE001
pass
def _terms_of(frame) -> List[Any]:
"""The clickable terms of a results frame, or ``[]``.
``feature`` is the key every plot and the table join on -- see
:class:`spacr.qt.widgets.regression_results.RegressionResultsPanel`. A
frame without it is not a coefficient table and there is nothing here to
warm.
"""
columns = getattr(frame, "columns", None)
if columns is None or "feature" not in columns:
return []
try:
return list(frame["feature"])
except Exception: # noqa: BLE001
return []
def _nothing_to_look_up(tile) -> str:
"""Why this tile has no gene to annotate, or ``""``.
A control guide, a model covariate and an unrecognised string are three
DIFFERENT answers and each of them is an answer. The one thing none of
them may produce is a panel of empty fields, which reads as "measured,
found nothing".
"""
if tile.kind == "control":
return ("This is a non-targeting control guide. There is no gene "
"behind it, so there is nothing to annotate — its effect is "
"the assay's own baseline.")
if tile.kind == "nuisance":
return (f"{tile.feature} is a model covariate, not a gene. It is "
"fitted so the real effects are estimated cleanly and is not "
"itself a hypothesis.")
if not tile.candidates:
return (tile.unresolved[0] if tile.unresolved else
f"{tile.feature} does not name a gene spaCR recognises.")
return ""
def _document_pixmap(document_html: str, width: int) -> QPixmap:
"""One HTML document rendered to a pixmap of the given width."""
from PySide6.QtGui import QTextDocument
document = QTextDocument()
document.setHtml(document_html)
document.setTextWidth(max(int(width), 1))
size = document.size().toSize()
pixmap = QPixmap(QSize(max(int(width), 1), max(size.height(), 1)))
pixmap.fill(Qt.transparent)
painter = QPainter(pixmap)
try:
document.drawContents(painter)
finally:
painter.end()
return pixmap