"""Infection metrics, assembled from a finished Measure run.
A REPORT, NOT A PIPELINE, and that was a deliberate choice made after the
derivability was measured: thirteen of the sixteen candidate infection metrics
fall out of tables Measure has already written, with no new image processing.
The three that do not -- intracellular fraction, vacuoles per cell, distance
to the host boundary -- belong to the invasion module and to segmentation, and
are deliberately absent here rather than approximated.
THE DENOMINATOR IS THE WHOLE PROBLEM, and it is why every number this module
produces carries its own denominator beside it. Almost every way of getting an
infection metric wrong is a denominator that does not match its numerator:
* The relationship table has NO ROW for a cell with no parasites.
``measure.get_components`` explodes the per-cell child list and drops the
empty ones, so counting rows there counts INFECTED cells and calls it the
cell count. The denominator has to come from the ``cell`` table, which
holds every segmented cell whether or not anything is inside it.
* ``io.py`` drops pathogens with no ``cell_id`` and can filter rows by
``pathogen_prcfo_count``. Neither touches the cell table. A numerator built
from a filtered pathogen table over an unfiltered cell table counts two
different populations, and the ratio looks entirely plausible.
So :func:`infection_report` returns the denominator, and its size, on every
row. A reader who disagrees with a number can see immediately which of the two
halves they disagree with.
IT DOES NOT INVENT A FIFTH VOCABULARY. ``uninfected``, ``pathogen_count`` and
``pathogen_prcfo_count`` already exist across ``measure``, ``io`` and
``timelapse``. This module reuses them and adds no synonym.
"""
from __future__ import annotations
import os
import sqlite3
from contextlib import closing
from typing import Dict, Iterable, List, Optional, Sequence, Tuple
import pandas as pd
def _connect_read_only(db_path):
"""A read-only connection that WAITS for Measure rather than failing.
`sqlite3.connect(..., mode=ro)` takes SQLite's five-second default and
then raises "database is locked" -- which, against a measurements.db
a Measure run is still writing, is a report that fails for a reason
that has nothing to do with the data. `database_concurrency.connect`
sets `busy_timeout` from its own `timeout` and opens with
`query_only=ON`, so a reader waits out a writer's transaction instead
of racing it.
Pinned by `test_no_connection_relies_on_sqlites_five_second_default`.
THE CALLER CLOSES IT. A ``sqlite3.Connection`` used as a context manager
only commits; it stays open, and an open reader keeps a WAL database's
``-wal`` file alive, so a copy of the ``.db`` alone taken afterwards
misses every change still in the log. Wrap it in
:func:`contextlib.closing` or close it in a ``finally``.
"""
from .database_concurrency import connect
return connect(db_path, readonly=True)
#: The table holding one row per segmented host cell. THE DENOMINATOR.
CELL_TABLE = "cell"
#: The table holding one row per segmented parasite.
PATHOGEN_TABLE = "pathogen"
#: How a pathogen row names the host cell it sits inside.
HOST_KEY = "cell_id"
#: What a well is, for grouping. Field is deliberately NOT here: a well mean
#: is the unit a screen reports, and the per-field spread is reported
#: separately precisely because a mean hides a settling gradient.
WELL_KEYS: Tuple[str, ...] = ("plateID", "rowID", "columnID")
#: Added to the well keys when the spread across fields is wanted.
FIELD_KEY = "fieldID"
#: The Measure setting that decides whether the cell table is a POPULATION or
#: a SELECTION. With it false, Measure writes no row for a cell that had no
#: parasite, so `cell` holds only infected cells and every infection rate
#: computed from it is 1.0 by construction.
INCLUDE_UNINFECTED_KEY = "include_uninfected"
#: The per-object border rules, which are applied at SEGMENTATION time --
#: `clear_border(mask)` in `utils.py` -- so they decide what is in each table
#: rather than how it is counted. The cell rule sets the denominator's
#: population and the pathogen rule sets the numerator's.
CELL_BORDER_KEY = "cell_remove_border_objects"
PATHOGEN_BORDER_KEY = "pathogen_remove_border_objects"
#: What `denominator` says instead of a population, when the population is not
#: in the database to be counted.
NOT_DERIVABLE = ("not derivable: the Measure run set include_uninfected=False, "
"so uninfected cells were never written")
[docs]
def uninfected_cells_were_measured(db_path: str) -> Optional[bool]:
"""Whether the Measure run that wrote ``db_path`` kept uninfected cells.
THE ANSWER CHANGES WHAT THE CELL TABLE IS. `include_uninfected=False` --
the default for a screen that only crops infected cells, and what the
TSG101 plates were measured with -- means Measure wrote no row for a cell
with no parasite. The cell table is then a selection of infected cells
rather than the segmented population, and `infected / cells` is 1.0 for
every well no matter what the biology did.
That is not a rounding problem to note in a docstring. A reader handed a
column of 1.000 reads 100% infection as a RESULT, and nothing in the
table says otherwise, so this module refuses the rate instead of
printing it.
:param db_path: path to a ``measurements.db``.
:returns: ``True`` or ``False`` when the run recorded the setting,
``None`` when it did not -- an older database, where the honest
answer is that we cannot tell.
"""
try:
db = _connect_read_only(db_path)
except sqlite3.Error:
return None
try:
if "settings" not in {
r[0] for r in db.execute(
"SELECT name FROM sqlite_master WHERE type='table'")}:
return None
row = db.execute(
"SELECT setting_value FROM settings WHERE setting_key = ?",
(INCLUDE_UNINFECTED_KEY,)).fetchone()
except sqlite3.Error:
return None
finally:
db.close()
if row is None or row[0] is None:
return None
return str(row[0]).strip().lower() in {"true", "1", "yes"}
def _setting(db_path: str, key: str) -> Optional[bool]:
"""One boolean setting the Measure run recorded, or ``None``.
:param db_path: path to a ``measurements.db``.
:param key: the ``settings`` row to read.
:returns: the value, or ``None`` when the run did not record it.
"""
try:
db = _connect_read_only(db_path)
except sqlite3.Error:
return None
try:
if "settings" not in {
r[0] for r in db.execute(
"SELECT name FROM sqlite_master WHERE type='table'")}:
return None
row = db.execute(
"SELECT setting_value FROM settings WHERE setting_key = ?",
(key,)).fetchone()
except sqlite3.Error:
return None
finally:
db.close()
if row is None or row[0] is None:
return None
return str(row[0]).strip().lower() in {"true", "1", "yes"}
[docs]
def border_rules_agree(db_path: str) -> Optional[bool]:
"""Whether cells and parasites were border-filtered the same way.
THE NUMERATOR AND THE DENOMINATOR MUST OBEY THE SAME RULE, which is what
377's PART 1 asks for in as many words: "the infection denominator has to
use the same rule or the rate is computed against a different cell count
than the numerator".
`remove_border_objects` is applied at SEGMENTATION time, per object type,
so it decides what is in each table rather than how the table is counted.
The two settings are independent and default to False together, so they
agree unless somebody changed one:
* pathogens cleared and cells kept -- a host cell whose only parasite
touched the edge is still in the cell table, now with a count of
zero. It reads as UNINFECTED and deflates the rate.
* cells cleared and pathogens kept -- a parasite can name a cell that
is no longer in the cell table, so it is counted in neither
numerator nor denominator and the infection index drifts.
THIS IS THE SAME FAILURE AS `include_uninfected`, WHICH ALREADY SHIPPED:
a setting that silently changes what a table contains, and therefore what
a ratio over it means, with nothing in the output saying so. That one
turned every well into 1.000000 on a real plate. This one is quieter,
which is worse -- a deflated rate looks like biology.
:param db_path: path to a ``measurements.db``.
:returns: ``True`` when both rules match, ``False`` when they differ,
``None`` when the run recorded neither -- an older database, where
the honest answer is that we cannot tell.
"""
cells = _setting(db_path, CELL_BORDER_KEY)
pathogens = _setting(db_path, PATHOGEN_BORDER_KEY)
if cells is None and pathogens is None:
return None
if cells is None or pathogens is None:
return None
return cells == pathogens
def _present_columns(db: sqlite3.Connection, table: str) -> List[str]:
"""The columns ``table`` actually has, or an empty list if it has none.
:param db: an open connection.
:param table: the table to inspect.
:returns: column names, empty when the table does not exist.
"""
try:
rows = db.execute(f'PRAGMA table_info("{table}")').fetchall()
except sqlite3.Error:
return []
return [str(r[1]) for r in rows]
def _canonical(columns: Sequence[str], wanted: str) -> Optional[str]:
"""The spelling ``columns`` uses for the identity column ``wanted``.
spaCR has written both ``plateID`` and ``plate`` over the years and a
database can carry either; the aliases live in :mod:`spacr.filters` and
are reused here rather than duplicated.
:param columns: the columns a table has.
:param wanted: the canonical name.
:returns: the spelling present, or None.
"""
from .filters import IDENTITY_ALIASES
have = {c.lower(): c for c in columns}
for alias in IDENTITY_ALIASES.get(wanted, (wanted,)):
if alias.lower() in have:
return have[alias.lower()]
return None
def _load(db: sqlite3.Connection, table: str,
columns: Optional[Iterable[str]] = None) -> pd.DataFrame:
"""Read ``table``, or an empty frame when it is absent.
An absent table is a fact about the run -- a project measured without a
pathogen channel has no pathogen table -- and not an error to raise into
a report.
:param db: an open connection.
:param table: table name.
:param columns: restrict to these, when given.
:returns: the table as a frame.
"""
have = _present_columns(db, table)
if not have:
return pd.DataFrame()
if columns:
keep = [c for c in columns if c in have]
if not keep:
return pd.DataFrame()
select = ", ".join(f'"{c}"' for c in keep)
else:
select = "*"
from .tabular import _read_query
return _read_query(db, f'SELECT {select} FROM "{table}"',
report=None)
[docs]
def parasites_per_cell(db_path: str) -> pd.DataFrame:
"""Every host cell, with how many parasites it contains -- ZERO INCLUDED.
THE ZEROES ARE THE POINT. The relationship between a cell and its
parasites is stored on the parasite, so a cell containing none appears
nowhere in the pathogen table. Any count taken from that table alone is a
count of infected cells wearing the name of a cell count.
:param db_path: path to a ``measurements.db``.
:returns: one row per cell, with the identity columns, ``object_label``
and ``pathogen_count``. Empty when there is no cell table.
"""
from .filters import OBJECT_COLUMN
with closing(_connect_read_only(db_path)) as db:
cells = _load(db, CELL_TABLE)
pathogens = _load(db, PATHOGEN_TABLE)
if cells.empty:
return pd.DataFrame()
keys = [c for c in (_canonical(cells.columns, k) for k in WELL_KEYS + (FIELD_KEY,))
if c]
label = _canonical(cells.columns, OBJECT_COLUMN) or OBJECT_COLUMN
if label not in cells.columns:
return pd.DataFrame()
out = cells[keys + [label]].copy()
out = out.rename(columns={label: OBJECT_COLUMN})
out["pathogen_count"] = 0
if pathogens.empty or HOST_KEY not in pathogens.columns:
return out
p_keys = [c for c in (_canonical(pathogens.columns, k)
for k in WELL_KEYS + (FIELD_KEY,)) if c]
counted = (pathogens.dropna(subset=[HOST_KEY])
.groupby(p_keys + [HOST_KEY], dropna=False)
.size().reset_index(name="n"))
counted = counted.rename(columns={HOST_KEY: OBJECT_COLUMN})
for frame in (out, counted):
frame[OBJECT_COLUMN] = pd.to_numeric(frame[OBJECT_COLUMN],
errors="coerce")
merged = out.merge(counted, how="left", on=keys + [OBJECT_COLUMN])
merged["pathogen_count"] = merged["n"].fillna(0).astype(int)
return merged.drop(columns=["n"])
def _monolayer_by_group(db_path: str, keys: Sequence[str]) -> Dict[tuple, dict]:
"""Confluency per report group, when the Measure run measured it.
When the Measure run also measured confluency, every row of
:func:`infection_report` gains ``monolayer_ok`` and each group two more
metrics: ``confluency``, the covered fraction of the imaged area, and
``parasites_per_confluency``, the parasite count over that fraction,
which compares wells per unit of monolayer rather than per field imaged.
``infection_report(..., monolayer_filter=True)`` drops the groups whose
monolayer failed the confluency QC; it is ignored when no confluency was
measured.
:param db_path: path to a ``measurements.db``.
:param keys: the report's grouping columns; a group with ``fieldID``
takes that field's own confluency, otherwise the well's pooled one.
:returns: identity tuple (as strings) -> ``confluency``, ``n_fields``
and ``monolayer_ok``; empty when the run wrote no confluency.
"""
from .measure import _confluency_by_well, _read_confluency
try:
fields = _read_confluency(db_path)
except Exception:
return {}
if fields.empty:
return {}
if FIELD_KEY in keys:
table = fields.assign(n_fields=1)
else:
table = _confluency_by_well(fields)
wanted = [key for key in keys if key in table.columns]
if len(wanted) != len(keys):
return {}
out = {}
for _, row in table.iterrows():
identity = tuple(str(row[key]) for key in keys)
out[identity] = {"confluency": float(row["confluency"]),
"n_fields": int(row["n_fields"]),
"monolayer_ok": int(row["monolayer_ok"])}
return out
[docs]
def infection_report(db_path: str, *,
by_field: bool = False,
monolayer_filter: bool = False) -> pd.DataFrame:
"""Infection metrics per well, each with the denominator it was computed over.
EVERY ROW CARRIES ITS DENOMINATOR because that is where these numbers go
wrong. ``infection rate`` over "cells" and ``parasites per infected`` over
"infected cells" are different populations, and a table that reports only
the ratios cannot be checked.
:param db_path: path to a ``measurements.db``.
:param by_field: group by field as well as well, which is what shows a
settling gradient a well mean hides.
:returns: tidy frame -- identity columns, then ``metric``, ``value``,
``denominator`` and ``n_denominator``. Empty when there is no cell
table to count.
"""
per_cell = parasites_per_cell(db_path)
if per_cell.empty:
return pd.DataFrame(columns=["metric", "value", "denominator",
"n_denominator"])
measured_uninfected = uninfected_cells_were_measured(db_path)
cells_are_all_infected = measured_uninfected is False
cell_population = ("infected host cells (uninfected excluded by the "
"Measure run)" if cells_are_all_infected
else "segmented host cells")
if border_rules_agree(db_path) is False:
cell_population += (" -- WARNING: cells and parasites were filtered "
"differently at the plate border, so this "
"denominator and its numerator are not the same "
"population")
keys = [c for c in per_cell.columns
if c in {_canonical(per_cell.columns, k) for k in WELL_KEYS}]
if by_field:
field = _canonical(per_cell.columns, FIELD_KEY)
if field:
keys = keys + [field]
monolayer = _monolayer_by_group(db_path, keys) if keys else {}
rows: List[dict] = []
grouped = per_cell.groupby(keys, dropna=False) if keys else [((), per_cell)]
for name, block in grouped:
identity = dict(zip(keys, name if isinstance(name, tuple) else (name,)))
cover = monolayer.get(tuple(str(identity[key]) for key in keys))
if monolayer and monolayer_filter and (
cover is None or not cover["monolayer_ok"]):
continue
first_row = len(rows)
cells = len(block)
infected = int((block["pathogen_count"] > 0).sum())
parasites = int(block["pathogen_count"].sum())
def add(metric, value, denominator, n):
"""Record one metric together with the population it is over.
The denominator travels with the value rather than being implied
by the metric's name, because two of these are ratios over
different populations and a reader cannot check a ratio whose
denominator they have to guess.
:param metric: the metric's name.
:param value: its value for this group.
:param denominator: what the value was computed over, in words.
:param n: how many things that denominator contained.
"""
rows.append({**identity, "metric": metric, "value": value,
"denominator": denominator, "n_denominator": n})
if cells_are_all_infected:
add("infection_rate", float("nan"), NOT_DERIVABLE, cells)
else:
add("infection_rate", infected / cells if cells else float("nan"),
cell_population, cells)
add("infection_index", parasites / cells if cells else float("nan"),
cell_population, cells)
add("parasites_per_infected",
parasites / infected if infected else float("nan"),
"infected host cells", infected)
add("cell_count", float(cells), cell_population, cells)
add("infected_count", float(infected), cell_population, cells)
add("parasite_count", float(parasites), "parasites with a host cell",
parasites)
if monolayer:
fraction = cover["confluency"] if cover else float("nan")
fields = cover["n_fields"] if cover else 0
add("confluency", fraction, "imaged field area", fields)
add("parasites_per_confluency",
parasites / fraction if fraction and fraction > 0
else float("nan"),
"covered fraction of the imaged area", fields)
ok = cover["monolayer_ok"] if cover else float("nan")
for row in rows[first_row:]:
row["monolayer_ok"] = ok
return pd.DataFrame(rows)
#: What a Measure run writes its infection report as, beside the
#: measurements database it was computed from.
REPORT_NAME = "infection_report.csv"
[docs]
def write_infection_report(db_path: str, *, by_field: bool = False,
destination: Optional[str] = None) -> Optional[str]:
"""Write the infection report beside the database it came from.
A MEASURE RUN EMITS THE REPORT, so anyone who has measured a plate
already has it. There is no button and no screen, and nothing has to be
asked for -- which is the point, because the runs that most need these numbers
are the ones that would never have thought to ask.
NOTHING IS WRITTEN WHEN THERE IS NOTHING TO SAY. A plate with no cell
table, or none of the columns the metrics need, gives an empty report,
and an empty CSV beside a database is a file that invites somebody to
wonder what went wrong. The answer is None instead.
Written to a dot-name in the same folder and renamed over the target,
so a reader never sees half a file.
:param db_path: a ``measurements.db``.
:param by_field: group by field as well as well.
:param destination: where to write it; the default is
:data:`REPORT_NAME` beside ``db_path``.
:returns: the path written, or None when the report is empty.
"""
report = infection_report(db_path, by_field=by_field)
if report.empty:
return None
target = destination or os.path.join(os.path.dirname(os.fspath(db_path)),
REPORT_NAME)
folder, name = os.path.split(target)
if folder:
os.makedirs(folder, exist_ok=True)
temporary = os.path.join(folder, f".{name}.tmp")
report.to_csv(temporary, index=False)
os.replace(temporary, target)
return target
[docs]
def multiplicity_distribution(db_path: str) -> pd.DataFrame:
"""The histogram behind the means, which is what the means hide.
Two conditions can share an infection index and differ completely: a few
heavily infected cells against many lightly infected ones is different
biology with the same average. The distribution is the number that
distinguishes them, so it is reported rather than summarised.
:param db_path: path to a ``measurements.db``.
:returns: one row per (well, parasite count) with ``cells`` and the
fraction of the well's cells at that count.
"""
per_cell = parasites_per_cell(db_path)
if per_cell.empty:
return pd.DataFrame(columns=["pathogen_count", "cells", "fraction"])
keys = [c for c in per_cell.columns
if c in {_canonical(per_cell.columns, k) for k in WELL_KEYS}]
counted = (per_cell.groupby(keys + ["pathogen_count"], dropna=False)
.size().reset_index(name="cells"))
totals = counted.groupby(keys, dropna=False)["cells"].transform("sum") \
if keys else counted["cells"].sum()
counted["fraction"] = counted["cells"] / totals
return counted
[docs]
def host_contrast(db_path: str, columns: Sequence[str], *,
statistic: str = "mean") -> pd.DataFrame:
"""One host measurement, infected against uninfected, IN THE SAME WELL.
The comparison is made within a well on purpose: comparing an infected
well to an uninfected one confounds infection with everything else that
differs between two wells, and the whole reason this is worth computing
is that both populations sit in the same one.
:param db_path: path to a ``measurements.db``.
:param columns: the host-cell measurements to contrast, e.g. ``("area",
"eccentricity")``.
:param statistic: any name ``DataFrameGroupBy.agg`` accepts.
:returns: one row per (well, measurement) with the infected and
uninfected values and both group sizes.
"""
from .filters import OBJECT_COLUMN
per_cell = parasites_per_cell(db_path)
if per_cell.empty:
return pd.DataFrame()
with closing(_connect_read_only(db_path)) as db:
cells = _load(db, CELL_TABLE)
if cells.empty:
return pd.DataFrame()
label = _canonical(cells.columns, OBJECT_COLUMN) or OBJECT_COLUMN
keys = [c for c in per_cell.columns
if c in {_canonical(per_cell.columns, k) for k in WELL_KEYS}]
join_on = keys + [OBJECT_COLUMN]
cells = cells.rename(columns={label: OBJECT_COLUMN})
cells[OBJECT_COLUMN] = pd.to_numeric(cells[OBJECT_COLUMN], errors="coerce")
wanted = [c for c in columns if c in cells.columns]
if not wanted:
return pd.DataFrame()
merged = per_cell.merge(cells[join_on + wanted], how="left", on=join_on)
merged["infected"] = merged["pathogen_count"] > 0
rows: List[dict] = []
grouped = merged.groupby(keys, dropna=False) if keys else [((), merged)]
for name, block in grouped:
identity = dict(zip(keys, name if isinstance(name, tuple) else (name,)))
yes = block[block["infected"]]
no = block[~block["infected"]]
for column in wanted:
rows.append({
**identity,
"measurement": column,
"infected": (yes[column].agg(statistic) if len(yes) else float("nan")),
"uninfected": (no[column].agg(statistic) if len(no) else float("nan")),
"n_infected": len(yes),
"n_uninfected": len(no),
})
return pd.DataFrame(rows)