"""Infection metrics, assembled from a finished Measure run.
This module builds a report from existing tables and runs no image
processing: thirteen of the sixteen candidate infection metrics can be
derived from tables Measure has already written.
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.
Every number this module produces carries its own denominator, because
most errors in infection metrics come from 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, not all
cells. 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.
The module adds no new terms. ``uninfected``, ``pathogen_count`` and
``pathogen_prcfo_count`` already exist across ``measure``, ``io`` and
``timelapse``; this module reuses them.
"""
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 determines what the cell table holds.
`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.
A column of 1.000 would read as 100% infection, and nothing in the table
would indicate otherwise, so this module withholds 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), so the answer is
unknown.
"""
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 use the same border rule;
otherwise 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 kind of problem as `include_uninfected`: a setting
that changes what a table contains, and therefore what a ratio over it
means, without any indication in the output. `include_uninfected` made
every well 1.000000 on a real plate; a border mismatch instead deflates
the rate, which can be mistaken for a biological effect.
: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), so the
answer is unknown.
"""
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 its parasite count, including cells with zero.
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 counts only infected
cells.
: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 a mismatched denominator is
the usual error in these numbers. ``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 writes the report automatically; it needs no button,
screen or setting.
Nothing is written when the report is empty. A plate with no cell
table, or none of the columns the metrics need, gives an empty report;
no empty CSV is written and None is returned.
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:
"""Per-well distribution of the number of parasites per cell.
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:
"""Compare host measurements of infected and uninfected cells per well.
The comparison is made within a well: comparing an infected well to an
uninfected one confounds infection with everything else that differs
between two wells, whereas here both populations share the same well.
: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)